Multi-Agent 多智能体

import operator
from typing import Annotated, TypedDict

class TeamState(TypedDict):
    # 黑板：所有成员的汇报自动追加，谁都不覆盖谁
    messages: Annotated[list, operator.add]
    # 结构化交付物：后写覆盖先写，由收尾节点维护
    report: str

from pydantic import BaseModel, Field

class Handoff(BaseModel):
    """控制权转移时必须携带的交接单"""
    to: str = Field(description="目标成员，只能是当前在线名单中的一个")
    task: str = Field(description="一句话说清要对方完成什么")
    context: str = Field(
        description="对方需要但无法自行获取的背景：用户是谁、已经决定了什么"
    )
    deliverable: str = Field(
        description="期望产出形式：结论摘要 / 结构化数据 / 待选项"
    )

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.prebuilt import create_agent

llm = init_chat_model("openai:gpt-4o-mini")

@tool
def web_search(query: str) -> str:
    """搜索公开网络信息，返回带来源链接的摘要。输入应是聚焦的搜索词。"""
    ...  # 接 Tavily / SerpAPI 等搜索后端

researcher = create_agent(
    llm,
    tools=[web_search],
    system_prompt=(
        "你是调研专员，为技术报告收集事实。\n"
        "1. 每条事实必须附来源 URL，找不到就明说，不许编造。\n"
        "2. 汇报格式：按主题分组的事实清单，每条一行。\n"
        "3. 只调研：不写结论段，不评价事实。"
    ),
)

writer = create_agent(
    llm,
    tools=[],
    system_prompt=(
        "你是主笔，把调研清单写成正式报告。\n"
        "1. 只使用调研记录中出现的事实，缺料处标注[待补充]。\n"
        "2. 关键论断后标注来源编号，如 [1]。\n"
        "3. 不做任何新的检索。"
    ),
)

critic = create_agent(
    llm,
    tools=[],
    system_prompt=(
        "你是审稿人，只挑错不重写。输出固定两节：\n"
        "## 通过项\n## 问题清单（每条：位置 / 问题 / 建议）\n"
        "没有问题时，问题清单一节写'无'。"
    ),
)

import operator
from typing import Annotated, Literal, TypedDict
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class TeamState(TypedDict):
    messages: Annotated[list, operator.add]  # 黑板（第三章）
    report: str                              # 交付物，由收尾节点填充

class Routing(BaseModel):
    """主管的路由决策"""
    next: Literal["researcher", "writer", "critic", "FINISH"]
    reason: str = Field(description="选择该成员的一句话理由，供人工审查")

SUPERVISOR_PROMPT = """你是团队主管，负责在成员间路由任务。

成员能力：
- researcher：联网调研，产出带来源的事实清单
- writer：把事实清单写成报告草稿
- critic：审稿，输出通过项与问题清单

路由纪律：
1. researcher 尚未产出事实清单之前，不允许派 writer。
2. writer 产出草稿后，必须派 critic。
3. critic 列出实质问题（问题清单不是"无"）时回到 writer，最多修改 2 轮。
4. critic 通过，或修改轮数用尽，选 FINISH。"""

router = llm.with_structured_output(Routing)

def supervisor(
    state: TeamState,
) -> Command[Literal["researcher", "writer", "critic", "__end__"]]:
    decision = router.invoke([
        {"role": "system", "content": SUPERVISOR_PROMPT},
        *state["messages"],
    ])
    if decision.next == "FINISH":
        return Command(goto=END, update={"messages": ["主管：任务完成，定稿。"]})
    # 把路由理由写回黑板：下一个成员知道为什么轮到自己，事后可回放
    return Command(
        goto=decision.next,
        update={"messages": [f"主管派活 -> {decision.next}：{decision.reason}"]},
    )

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor)

members = {"researcher": researcher, "writer": writer, "critic": critic}
for name, agent in members.items():
    builder.add_node(name, agent)
    builder.add_edge(name, "supervisor")  # 成员干完活必回主管

builder.add_edge(START, "supervisor")
graph = builder.compile()

inputs = {"messages": [("user",
    "写一份 LangGraph 1.x 学习路线报告，面向有 LangChain 基础的工程师")]}

config = {"recursion_limit": 60, "thread_id": "team-001"}  # 一轮=主管+成员两跳

for chunk in graph.stream(inputs, config, stream_mode="updates"):
    print(chunk)  # 每跳打印一次：谁在执行、黑板新增了什么

# 最终报告：黑板上 writer 的最后一条定稿消息
# 需要结构化交付时，加一个收尾节点从 messages 抽取并写入 report

from typing import Annotated
from langchain_core.messages import ToolMessage
from langchain.tools import tool
from langgraph.prebuilt import InjectedToolCallId
from langgraph.types import Command

def make_handoff(target: str, description: str):
    """生成一个转接工具：谁装上它，谁就能把控制权交给 target 成员"""
    @tool(description=description)
    def handoff(
        task: str,                                          # 交接单核心字段
        tool_call_id: Annotated[str, InjectedToolCallId],   # 框架自动注入
    ) -> Command:
        return Command(
            goto=target,           # 父图中的目标成员节点
            graph=Command.PARENT,  # 跳出当前子图，在父图里路由
            update={
                "messages": [
                    ToolMessage(f"已转接给 {target}。", tool_call_id=tool_call_id),
                    {"role": "user", "content": task},  # 对方接手的任务说明
                ]
            },
        )
    return handoff

from langgraph.graph import StateGraph, START, MessagesState

billing = create_agent(
    llm,
    tools=[make_handoff("tech", "当问题转向产品故障、报错、配置排查时，转接技术支持")],
    system_prompt="你是账务专员，只处理账单、退款、发票问题。超出职责立即转接。",
)

tech = create_agent(
    llm,
    tools=[make_handoff("billing", "当问题转向账单金额、退款进度时，转接账务专员")],
    system_prompt="你是技术支持，只处理故障排查与配置问题。超出职责立即转接。",
)

builder = StateGraph(MessagesState)   # 共享对话状态：会话历史全程连续
builder.add_node("billing", billing)
builder.add_node("tech", tech)
builder.add_edge(START, "billing")    # 前台默认路由：先到账务
swarm = builder.compile()

# 控制权随对话流动：账单问题 -> 转接 -> 故障排查
swarm.invoke({"messages": [("user", "这个月账单多扣了钱，而且 App 一直闪退")]})

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class ResearchState(TypedDict):
    topic: str
    subtopics: list[str]                          # 计划节点拆出的子题
    findings: Annotated[list[str], operator.add]  # 各分支自动汇聚
    report: str

def fan_out(state: ResearchState) -> list[Send]:
    """每个子题派一个调研分支，输入互不干扰"""
    return [
        Send("researcher", {"topic": state["topic"], "sub": sub})
        for sub in state["subtopics"]
    ]

def researcher_branch(state: dict) -> dict:
    # 分支私有输入：只有 topic + sub，看不到其他分支（天然隔离）
    result = researcher.invoke({"messages": [
        ("user", f"只调研子题「{state['sub']}」，产出 5 条带来源的事实")
    ]})
    summary = result["messages"][-1].content   # 分支只上抛结论
    return {"findings": [f"## {state['sub']}\n{summary}"]}

def aggregate(state: ResearchState) -> dict:
    """收拢：分支结论已按子题分段，交给主笔"""
    merged = "\n\n".join(state["findings"])   # reducer 已把各分支结果追加在一起
    draft = writer.invoke(f"把以下分组事实综合成报告：\n{merged}").content
    return {"report": draft}

builder = StateGraph(ResearchState)
builder.add_node("plan", plan_node)             # 拆子题的 LLM 节点
builder.add_node("researcher", researcher_branch)
builder.add_node("aggregate", aggregate)
builder.add_conditional_edges("plan", fan_out, ["researcher"])
builder.add_edge("researcher", "aggregate")
builder.add_edge(START, "plan")
builder.add_edge("aggregate", END)
graph = builder.compile()

graph.invoke(
    {"topic": "向量数据库选型", "subtopics": ["主流产品对比", "价格模型", "运维成本"]},
    config={"max_concurrency": 3},   # 限制同时在跑的分支数
)

from pydantic import BaseModel, Field

class ResearchReport(BaseModel):
    """调研成员对父图的唯一输出接口"""
    facts: list[str] = Field(description="事实清单，每条附来源 URL")
    gaps: list[str] = Field(description="没查到的关键问题")
    confidence: float = Field(description="整体置信度 0-1")

reporter = llm.with_structured_output(ResearchReport)

def researcher_node(state: TeamState) -> dict:
    # 1. 私有会话：完整工具循环只存在于这个局部变量里
    scratch = researcher.invoke(
        {"messages": [("user", f"调研：{state['task']}")]}
    )
    # 2. 压缩：把长会话压成结构化结论
    report = reporter.invoke([
        {"role": "system", "content": "把调研记录压成结构化报告，逐条保留来源。"},
        *scratch["messages"],
    ])
    # 3. 上抛：父图只拿到结论，不拿到过程
    return {
        "messages": [f"调研完成：{len(report.facts)} 条事实，置信度 {report.confidence}"],
        "research": report,
    }

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 任意成员写入：团队级决策，命名空间 (类型, 范围)
store.put(("decision", "team"), "style-guide", {
    "text": "报告采用中文书面语，代码示例统一 Python 3.12",
    "author": "critic",        # 谁写入的，出问题可归因
})

# 任意成员读取（跨线程持久，生产环境换 PostgresStore）
item = store.get(("decision", "team"), "style-guide")
items = store.search(("preference", "user-42"))  # 列出命名空间内条目

def route_sequence(graph, inputs, config) -> list[str]:
    """跑一遍图，按顺序记录成员跳（过滤掉 supervisor 本身）"""
    seq = []
    for update in graph.stream(inputs, config, stream_mode="updates"):
        seq.extend(update.keys())          # 每个 update 的键 = 本跳执行的节点
    return [n for n in seq if n != "supervisor"]

def eval_routing(graph, cases) -> float:
    """cases: [{"inputs": ..., "expect": ["researcher", "writer", "critic", "FINISH"]}]"""
    hit = total = 0
    for case in cases:
        seq = route_sequence(graph, case["inputs"], {"recursion_limit": 60})
        for expect, actual in zip(case["expect"], seq):
            total += 1
            hit += expect == actual
    return hit / total   # 低于 0.9：先修主管提示，再谈别的优化

from collections import defaultdict

def attribute_tokens(spans) -> dict[str, int]:
    """spans：一次运行的 trace 导出，每条含节点名与 token 用量"""
    per_node = defaultdict(int)
    for span in spans:
        if span["type"] == "llm":            # 只统计 LLM 调用
            per_node[span["node"]] += span["total_tokens"]
    return dict(per_node)

# {'supervisor': 42000, 'researcher': 310000, 'writer': 68000, 'critic': 51000}
# researcher 一家烧掉六成：先优化它的检索策略，而不是全团队换模型

TOKEN_BUDGET = 400_000   # 每次运行的全局预算，按 trace 成本归因校准

def supervisor_with_budget(
    state: TeamState,
) -> Command[Literal["researcher", "writer", "critic", "__end__"]]:
    # tokens_used：中间件在每个 LLM 调用后累加 usage_metadata（TeamState 加 int 字段）
    if state["tokens_used"] > TOKEN_BUDGET:
        # 预算耗尽：不再派新任务，让主笔用手头材料收尾，而不是报错
        return Command(
            goto="writer",
            update={"messages": ["主管：预算耗尽，基于已有材料直接收尾，不要再索取新料。"]},
        )
    decision = router.invoke([
        {"role": "system", "content": SUPERVISOR_PROMPT},
        *state["messages"],
    ])
    if decision.next == "FINISH":
        return Command(goto=END, update={"messages": ["主管：任务完成，定稿。"]})
    return Command(
        goto=decision.next,
        update={"messages": [f"主管派活 -> {decision.next}：{decision.reason}"]},
    )