多 Agent 系统怎么搭？Manager-Worker协作模式的编排、可观测与回滚实战

{
  "task": "撰写一份《2026 智能体平台》行业报告并做合规审查",
  "plan": [
    { "id": "r1", "worker": "researcher", "instruction": "检索近一年主流平台动态，整理要点" },
    { "id": "r2", "worker": "writer",     "instruction": "基于 r1 产出 1500 字报告初稿" },
    { "id": "r3", "worker": "reviewer",   "instruction": "依据内部合规清单审查 r2，给出风险点" }
  ]
}

# 极简黑板实现
blackboard = {
    "r1": {"status": "done", "output": "检索要点：..."},
    "r2": {"status": "running", "output": None},
    "r3": {"status": "pending", "output": None},
}

def read(blackboard, task_id):
    return blackboard.get(task_id, {}).get("output")

def write(blackboard, task_id, output):
    blackboard[task_id] = {"status": "done", "output": output}

trace = []
def call_worker(worker, instruction, parent_id=None):
    span = {"worker": worker, "instruction": instruction,
            "parent": parent_id, "start": time.time(), "ok": False}
    try:
        output = worker.run(instruction)
        span["ok"], span["output"] = True, output
        return output
    finally:
        span["cost"] = time.time() - span["start"]
        trace.append(span)   # 无论成败都记录，便于复盘

checkpoints = []
def checkpoint(blackboard):
    checkpoints.append(copy.deepcopy(blackboard))

# 执行某步前存盘，失败后回退
checkpoint(blackboard)
try:
    run_step(blackboard, "r3")
except ComplianceError:
    blackboard = checkpoints[-1]   # 退回审查前状态，转人工兜底

import copy, time

WORKERS = {
    "researcher": "你是行业研究员，只做检索与要点整理。",
    "writer":     "你是报告写手，基于给定资料产出初稿。",
    "reviewer":   "你是合规审查员，依据清单给风险点。",
}

def llm(system, prompt):
    # 此处替换为真实模型调用（OpenAI / 百炼 / 混元 / 智谱 等均可）
    return f"[来自 {system[:6]} 的结果] {prompt[:20]}..."

blackboard, trace, checkpoints = {}, [], []

def call_worker(name, instruction, parent=None):
    span = {"worker": name, "instruction": instruction,
            "parent": parent, "start": time.time(), "ok": False}
    try:
        out = llm(WORKERS[name], instruction)
        blackboard[name] = {"status": "done", "output": out}
        span["ok"] = True
        return out
    finally:
        span["cost"] = round(time.time() - span["start"], 3)
        trace.append(span)

def run_manager(task):
    # 1. 分解（示意：固定三步流水线）
    steps = [
        ("researcher", f"调研：{task}"),
        ("writer",     "基于 researcher 的结论写初稿"),
        ("reviewer",   "审查 writer 的初稿"),
    ]
    checkpoints.append(copy.deepcopy(blackboard))
    for i, (w, ins) in enumerate(steps):
        prev = steps[i-1][0] if i > 0 else None
        ctx = blackboard.get(prev, {}).get("output", "") if prev else ""
        call_worker(w, f"{ins}。上下文：{ctx}", parent=prev)
    # 2. 汇总
    return {
        "report": blackboard["writer"]["output"],
        "review": blackboard["reviewer"]["output"],
        "trace":  trace,
    }

result = run_manager("2026 智能体平台趋势")
print(result["report"], "\n--- trace ---", result["trace"])