Subagent 与多智能体编排

from dataclasses import dataclass, asdict
from typing import Literal
import json
@dataclass
class SubResult:
    """
    子 Agent 返回给父 Agent 的唯一格式。
    设计要点：带证据、带置信度、带来源——否则父 Agent 无法判断该不该采信。
    """
    sub_task: str
    status: Literal["ok", "partial", "failed"]
    conclusion: str                 # 结论，尽量短
    evidence: list[str]             # 支撑结论的最小证据（每条一句）
    sources: list[str]              # 来源标识，可点开核对
    confidence: float               # 0-1，子 Agent 自评
    tokens_used: int = 0
    caveats: list[str] = None       # 已知局限 / 未覆盖范围
    def render(self) -> str:
        """父 Agent 看到的形态：结构化、紧凑、可核验"""
        return json.dumps(asdict(self), ensure_ascii=False, indent=1)
def dispatch(sub_tasks: list[str], run_subagent) -> list[SubResult]:
    """
    并行派发。注意三点：
      1. 每个子任务必须有明确的产出要求（否则子 Agent 会返回无法使用的东西）
      2. 失败不能中断整体（partial 也要收）
      3. 汇总前先做冲突检测
    """
    results: list[SubResult] = []
    for t in sub_tasks:
        spec = (f"{t}\n\n"
                f"要求：只返回结论 + 最少必要证据 + 来源 + 你的置信度（0-1）。"
                f"不要返回过程、不要返回原始文档全文。")
        try:
            results.append(run_subagent(spec))
        except Exception as e:
            results.append(SubResult(
                sub_task=t, status="failed",
                conclusion=f"子任务执行失败：{type(e).__name__}",
                evidence=[], sources=[], confidence=0.0))
    return results
def merge(results: list[SubResult]) -> dict:
    """汇总：先归并，再显式暴露冲突，绝不静默选边"""
    ok = [r for r in results if r.status == "ok"]
    partial = [r for r in results if r.status == "partial"]
    failed = [r for r in results if r.status == "failed"]
    # 低置信度结论不进主结论，只作为参考项
    strong = [r for r in ok if r.confidence >= 0.7]
    weak = [r for r in ok if r.confidence < 0.7]
    conflicts = detect_conflicts(strong)
    return {
        "primary": [{"conclusion": r.conclusion, "sources": r.sources} for r in strong],
        "reference": [{"conclusion": r.conclusion, "confidence": r.confidence} for r in weak],
        "unresolved": conflicts,        # 交给父 Agent 或用户裁决，不自己选
        "partial": [r.sub_task for r in partial],
        "failed": [{"task": r.sub_task, "reason": r.conclusion} for r in failed],
        "coverage": f"{len(ok)}/{len(results)} 个子任务完成",
    }
def detect_conflicts(results: list[SubResult]) -> list[dict]:
    """
    冲突检测：真实系统里可以用语义比对或让父 Agent 判断；
    这里用一个可插拔的占位实现，重点是**把冲突显式暴露出来**。
    """
    found = []
    for i in range(len(results)):
        for j in range(i + 1, len(results)):
            a, b = results[i], results[j]
            if contradicts(a.conclusion, b.conclusion):     # 占位
                found.append({"a": a.conclusion, "b": b.conclusion,
                              "a_sources": a.sources, "b_sources": b.sources})
    return found
def contradicts(a: str, b: str) -> bool:
    return False        # 占位：真实实现用语义模型或规则