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