Coverage for agentos/subagent/collaboration.py: 34%
204 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2Agent 协作模式 — Debate/Vote/Review/Pipeline/Ensemble。
3基于 SubAgentManager + 父子通信之上，提供高级多Agent协作原语。
5使用示例::
7 mgr = SubAgentManager()
8 collab = AgentCollaboration(mgr)
10 result = await collab.debate("Python vs Rust for web backend", agents=2)
11 result = await collab.vote(["方案A", "方案B", "方案C"], agents=5)
12 result = await collab.review("写一篇关于AI安全的文章", rounds=2)
13 result = await collab.pipeline("分析Q2财报数据", stages=3)
14 result = await collab.ensemble("设计系统架构方案", agents=3)
15"""
17from __future__ import annotations
19import re
20import time
21import uuid
22from dataclasses import dataclass, field
23from enum import Enum
24from typing import Any, Callable, Awaitable
26from .manager import SubAgentManager, SubAgentSpec, SubAgentResult
27from .parent_child import ChildContext, SharedState
30# ──────────────────────────────────────────────
31# 枚举与数据类型
32# ──────────────────────────────────────────────
35class CollaborationMode(str, Enum):
36 DEBATE = "debate"
37 VOTE = "vote"
38 REVIEW = "review"
39 PIPELINE = "pipeline"
40 ENSEMBLE = "ensemble"
43class VoteStrategy(str, Enum):
44 MAJORITY = "majority"
45 WEIGHTED = "weighted"
46 RANKED = "ranked"
47 UNANIMOUS = "unanimous"
50@dataclass
51class DebateRound:
52 """一轮辩论。"""
53 round: int
54 arguments: list[str] # 各方论点
55 rebuttals: list[str] # 反驳
56 winner: int | None = None
59@dataclass
60class VoteBallot:
61 """一张选票。"""
62 agent_id: str
63 choice: str
64 confidence: float = 1.0
65 reasoning: str = ""
68@dataclass
69class ReviewPass:
70 """一轮审查。"""
71 round: int
72 draft: str
73 feedback: str
74 revised: str
75 score: float = 0.0
78@dataclass
79class CollaborationResult:
80 """协作结果。"""
81 mode: CollaborationMode
82 agents: list[str]
83 rounds: int
84 final_output: str
85 intermediate: list[Any] = field(default_factory=list)
86 consensus: float = 0.0
87 duration: float = 0.0
90# ──────────────────────────────────────────────
91# 角色性格库
92# ──────────────────────────────────────────────
94_PERSONAS = [
95 "Be logical, data-driven, and cite evidence.",
96 "Be creative, think outside the box, challenge assumptions.",
97 "Focus on practical concerns: cost, timeline, feasibility.",
98 "Advocate for user experience and human-centered design.",
99 "Take a contrarian stance, find flaws in all arguments.",
100]
102# ──────────────────────────────────────────────
103# 核心协作引擎
104# ──────────────────────────────────────────────
107class AgentCollaboration:
108 """多Agent协作引擎。
110 参数:
111 manager: SubAgentManager 实例
112 run_func: 执行函数 (task_str, ctx) -> (output, iterations)
113 default_timeout: 每个子Agent默认超时（秒）
114 """
116 def __init__(
117 self,
118 manager: SubAgentManager | None = None,
119 run_func: Callable[[SubAgentSpec, ChildContext], Awaitable[tuple[str, int]]] | None = None,
120 default_timeout: float | None = 300.0,
121 ):
122 self._mgr = manager or SubAgentManager()
123 self._run = run_func
124 self._timeout = default_timeout
126 # ── 便捷属性 ────────────────────────────
128 @property
129 def manager(self) -> SubAgentManager:
130 return self._mgr
132 @property
133 def shared_state(self) -> SharedState:
134 return self._mgr.shared_state
136 async def cancel_all(self) -> None:
137 await self._mgr.cancel_all()
139 @property
140 def active_agents(self) -> int:
141 return self._mgr.active_children
143 # ── 内部辅助 ────────────────────────────
145 async def _spawn(self, task: str, timeout: float | None = None) -> SubAgentResult:
146 """Fork 一个单Agent执行 task。"""
147 return await self._mgr.spawn_fork(
148 task=task,
149 run_func=self._run,
150 timeout=timeout or self._timeout,
151 )
153 async def _spawn_many(self, tasks: list[str], timeout: float | None = None) -> list[SubAgentResult]:
154 """Swarm 并行执行多个 task。"""
155 return await self._mgr.spawn_swarm(
156 tasks=tasks,
157 run_func=self._run,
158 timeout=timeout or self._timeout,
159 )
161 @staticmethod
162 def _parse_score(output: str) -> float:
163 """从输出中解析 SCORE: N 格式的评分。"""
164 m = re.search(r'SCORE:\s*([\d.]+)', output, re.IGNORECASE)
165 if m:
166 return max(0.0, min(10.0, float(m.group(1)))) / 10.0
167 return 0.7
169 @staticmethod
170 def _parse_choice(
171 output: str, option_count: int
172 ) -> tuple[int | None, float, str]:
173 """解析 CHOICE: N | CONFIDENCE: X | REASONING: text。"""
174 choice = None
175 confidence = 0.5
176 reasoning = output[:200]
178 m_choice = re.search(r'CHOICE:\s*(\d+)', output, re.IGNORECASE)
179 if m_choice:
180 num = int(m_choice.group(1))
181 if 1 <= num <= option_count:
182 choice = num
184 m_conf = re.search(r'CONFIDENCE:\s*([\d.]+)', output, re.IGNORECASE)
185 if m_conf:
186 confidence = max(0.0, min(1.0, float(m_conf.group(1))))
188 m_reason = re.search(r'REASONING:\s*(.+?)(?:\n|$)', output, re.IGNORECASE | re.DOTALL)
189 if m_reason:
190 reasoning = m_reason.group(1).strip()[:200]
192 return choice, confidence, reasoning
194 # ══════════════════════════════════════════
195 # 1. Debate 辩论模式
196 # ══════════════════════════════════════════
198 async def debate(
199 self,
200 topic: str,
201 agents: int = 2,
202 rounds: int = 3,
203 timeout: float | None = None,
204 ) -> CollaborationResult:
205 """多个Agent辩论 topic，裁判总结。
207 流程:
208 Round 0: 各方发表初始论点
209 Round 1~N: 反驳对方 + 强化己方
210 裁判: 综合所有论点给出最终裁决
211 """
212 t0 = time.time()
213 history: list[DebateRound] = []
214 all_agent_ids: list[str] = []
216 # Round 0 — 初始论点
217 r0_tasks = [
218 f"Debate topic: {topic}\n"
219 f"You are debater {chr(65+i)}. {_PERSONAS[i % len(_PERSONAS)]}\n"
220 f"Present your opening argument."
221 for i in range(agents)
222 ]
223 r0_results = await self._spawn_many(r0_tasks, timeout)
224 r0_args = [r.output for r in r0_results]
225 history.append(DebateRound(round=0, arguments=r0_args, rebuttals=[]))
226 all_agent_ids.extend(r.agent_id for r in r0_results)
228 # Round 1..N — 反驳强化
229 for rnd in range(1, rounds):
230 rebut_tasks = []
231 for i in range(agents):
232 opponent_args = [
233 history[-1].arguments[j]
234 for j in range(agents) if j != i
235 ]
236 rebut_tasks.append(
237 f"Debate topic: {topic}\n"
238 f"You are debater {chr(65+i)}. {_PERSONAS[i % len(_PERSONAS)]}\n"
239 f"Opponent arguments: {' | '.join(opponent_args)}\n"
240 f"Provide your rebuttal and strengthen your position."
241 )
242 rebut_results = await self._spawn_many(rebut_tasks, timeout)
243 rebuttals = [r.output for r in rebut_results]
244 history.append(DebateRound(
245 round=rnd,
246 arguments=[history[-1].arguments[i] for i in range(agents)],
247 rebuttals=rebuttals,
248 ))
249 all_agent_ids.extend(r.agent_id for r in rebut_results)
251 # 裁判总结
252 all_args = "\n\n".join([
253 f"Debater {chr(65+i)} initial: {history[0].arguments[i]}\n"
254 f"Debater {chr(65+i)} final rebuttal: "
255 f"{history[-1].rebuttals[i] if i < len(history[-1].rebuttals) else 'N/A'}"
256 for i in range(agents)
257 ])
258 judge = await self._spawn(
259 f"As an impartial judge, synthesize this debate on '{topic}' and give your verdict:\n"
260 f"{all_args}",
261 timeout,
262 )
263 all_agent_ids.append(judge.agent_id)
265 return CollaborationResult(
266 mode=CollaborationMode.DEBATE,
267 agents=all_agent_ids,
268 rounds=rounds + 1,
269 final_output=judge.output,
270 intermediate=history,
271 consensus=0.5 + 0.1 * min(agents, 5),
272 duration=time.time() - t0,
273 )
275 # ══════════════════════════════════════════
276 # 2. Vote 投票模式
277 # ══════════════════════════════════════════
279 async def vote(
280 self,
281 options: list[str],
282 agents: int = 3,
283 strategy: VoteStrategy = VoteStrategy.MAJORITY,
284 timeout: float | None = None,
285 ) -> CollaborationResult:
286 """多个Agent投票选择最优方案。
288 参数:
289 options: 候选选项列表
290 agents: 投票Agent数
291 strategy: 统计策略
292 """
293 t0 = time.time()
294 option_list = "\n".join(f"{i+1}. {opt}" for i, opt in enumerate(options))
295 ballots: list[VoteBallot] = []
297 vote_tasks = [
298 f"You are voter {i+1}/{agents}. {_PERSONAS[i % len(_PERSONAS)]}\n"
299 f"Evaluate these options and vote for ONE:\n{option_list}\n"
300 f"Format: CHOICE: <number> | CONFIDENCE: <0.0-1.0> | REASONING: <text>"
301 for i in range(agents)
302 ]
303 results = await self._spawn_many(vote_tasks, timeout)
305 for r in results:
306 c, conf, reason = self._parse_choice(r.output, len(options))
307 if c is not None:
308 ballots.append(VoteBallot(
309 agent_id=r.agent_id,
310 choice=options[c - 1],
311 confidence=conf,
312 reasoning=reason,
313 ))
315 _tally: dict[str, int] = {}
316 _weighted: dict[str, float] = {}
317 for b in ballots:
318 _tally[b.choice] = _tally.get(b.choice, 0) + 1
319 _weighted[b.choice] = _weighted.get(b.choice, 0) + b.confidence
321 if not ballots:
322 return CollaborationResult(
323 mode=CollaborationMode.VOTE,
324 agents=[],
325 rounds=1,
326 final_output="No valid votes cast.",
327 consensus=0.0,
328 duration=time.time() - t0,
329 )
331 if strategy == VoteStrategy.WEIGHTED:
332 winner = max(_weighted, key=_weighted.get)
333 consensus = _weighted[winner] / sum(_weighted.values())
334 summary = f"Weighted winner: {winner} (score: {_weighted[winner]:.2f})"
335 elif strategy == VoteStrategy.UNANIMOUS:
336 if len(_tally) == 1 and list(_tally.values())[0] == agents:
337 winner, consensus = list(_tally.keys())[0], 1.0
338 summary = f"Unanimous: {winner}"
339 else:
340 winner, consensus = "NO CONSENSUS", 0.0
341 summary = "Unanimous vote FAILED"
342 else:
343 winner = max(_tally, key=_tally.get)
344 consensus = _tally[winner] / len(ballots)
345 summary = f"Majority winner: {winner} ({_tally[winner]}/{len(ballots)} votes)"
347 report = f"{summary}\n\nVote details:\n"
348 for b in ballots:
349 report += f"- [{b.agent_id}] '{b.choice}' (conf={b.confidence:.2f}): {b.reasoning}\n"
351 return CollaborationResult(
352 mode=CollaborationMode.VOTE,
353 agents=[b.agent_id for b in ballots],
354 rounds=1,
355 final_output=report,
356 intermediate=ballots,
357 consensus=consensus,
358 duration=time.time() - t0,
359 )
361 # ══════════════════════════════════════════
362 # 3. Review 审查模式
363 # ══════════════════════════════════════════
365 async def review(
366 self,
367 task: str,
368 rounds: int = 2,
369 timeout: float | None = None,
370 ) -> CollaborationResult:
371 """Writer产出 → Reviewer审查 → 多轮迭代。
373 流程:
374 1. Writer 生成初稿
375 2. Reviewer 审查 → 打分 + 反馈
376 3. Writer 根据反馈修改
377 4. 重复 rounds 次
378 """
379 t0 = time.time()
380 passes: list[ReviewPass] = []
381 writer_id = uuid.uuid4().hex[:8]
382 reviewer_id = uuid.uuid4().hex[:8]
384 draft_result = await self._spawn(
385 f"As a writer, complete: {task}", timeout,
386 )
387 draft = draft_result.output
389 for rnd in range(rounds):
390 review_result = await self._spawn(
391 f"As a reviewer, evaluate this draft (round {rnd+1}):\n{draft}\n"
392 f"Provide specific feedback. Format: SCORE: <0-10> | FEEDBACK: <text>",
393 timeout,
394 )
395 feedback = review_result.output
396 score = self._parse_score(feedback)
398 revise_result = await self._spawn(
399 f"As a writer, revise your draft based on this feedback:\n{feedback}\n\n"
400 f"Original draft:\n{draft}\n\nProvide the revised version.",
401 timeout,
402 )
403 revised = revise_result.output
405 passes.append(ReviewPass(
406 round=rnd + 1,
407 draft=draft,
408 feedback=feedback,
409 revised=revised,
410 score=score,
411 ))
412 draft = revised
414 return CollaborationResult(
415 mode=CollaborationMode.REVIEW,
416 agents=[writer_id, reviewer_id],
417 rounds=rounds,
418 final_output=draft,
419 intermediate=passes,
420 consensus=passes[-1].score if passes else 0.0,
421 duration=time.time() - t0,
422 )
424 # ══════════════════════════════════════════
425 # 4. Pipeline 流水线模式
426 # ══════════════════════════════════════════
428 async def pipeline(
429 self,
430 task: str,
431 stages: int = 3,
432 stage_names: list[str] | None = None,
433 timeout: float | None = None,
434 ) -> CollaborationResult:
435 """多Agent串联处理，前一输出是后一输入。
437 参数:
438 task: 初始输入
439 stages: 流水线段数
440 stage_names: 自定义阶段名，默认 ['Analyzer','Processor','Refiner',...]
441 """
442 t0 = time.time()
443 if stage_names is None:
444 stage_names = ["Analyzer", "Processor", "Refiner", "Polisher", "Validator"][:stages]
446 intermediates: list[str] = []
447 agent_ids: list[str] = []
448 current_input = task
450 for name in stage_names:
451 result = await self._spawn(
452 f"You are the {name} stage in a processing pipeline.\n"
453 f"Input: {current_input}\nProcess and output for the next stage.",
454 timeout,
455 )
456 current_input = result.output
457 intermediates.append(current_input)
458 agent_ids.append(result.agent_id)
460 return CollaborationResult(
461 mode=CollaborationMode.PIPELINE,
462 agents=agent_ids,
463 rounds=stages,
464 final_output=current_input,
465 intermediate=intermediates,
466 consensus=1.0,
467 duration=time.time() - t0,
468 )
470 # ══════════════════════════════════════════
471 # 5. Ensemble 集成模式
472 # ══════════════════════════════════════════
474 async def ensemble(
475 self,
476 task: str,
477 agents: int = 3,
478 merge_strategy: str = "best_of",
479 timeout: float | None = None,
480 ) -> CollaborationResult:
481 """多个Agent独立求解，合并最优。
483 参数:
484 task: 需求描述
485 agents: 求解Agent数
486 merge_strategy: 'best_of' | 'merge' | 'weighted'
487 """
488 t0 = time.time()
490 tasks = [
491 f"You are solver {i+1}/{agents}. {_PERSONAS[i % len(_PERSONAS)]}\n"
492 f"Complete: {task}\nAt the end self-evaluate: SELF_SCORE: <0-10>"
493 for i in range(agents)
494 ]
495 results = await self._spawn_many(tasks, timeout)
497 scored: list[tuple[float, str, str]] = []
498 for r in results:
499 s = self._parse_score(r.output) * 10 # 0-10
500 scored.append((s, r.output, r.agent_id))
501 scored.sort(reverse=True, key=lambda x: x[0])
503 if not scored:
504 return CollaborationResult(
505 mode=CollaborationMode.ENSEMBLE,
506 agents=[], rounds=1,
507 final_output="No results.",
508 duration=time.time() - t0,
509 )
511 if merge_strategy == "best_of":
512 bs, bo, ba = scored[0]
513 final = f"Best solution (score: {bs:.1f}/10) from [{ba}]:\n\n{bo}"
514 elif merge_strategy == "merge":
515 parts = []
516 for i, (s, o, a) in enumerate(scored):
517 parts.append(f"[Solution {i+1}, score={s:.1f}]:\n{o[:500]}")
518 merge_result = await self._spawn(
519 f"As a meta-synthesizer, merge these {agents} solutions:\n\n"
520 + "\n---\n".join(parts),
521 timeout,
522 )
523 final = merge_result.output
524 else: # weighted
525 total = sum(s for s, _, _ in scored)
526 weights = [(s / total) if total else 0 for s, _, _ in scored]
527 parts = []
528 for (s, o, a), w in zip(scored, weights):
529 parts.append(f"[{a}] weight={w:.2f}:\n{o[:300]}")
530 final = f"Weighted ensemble ({agents} solvers):\n\n" + "\n---\n".join(parts)
532 return CollaborationResult(
533 mode=CollaborationMode.ENSEMBLE,
534 agents=[a for _, _, a in scored],
535 rounds=1,
536 final_output=final,
537 intermediate=scored,
538 consensus=scored[0][0] / 10.0,
539 duration=time.time() - t0,
540 )