多智能体编排与工作流引擎

循环：
  Thought: 我需要查一下今天的汇率
  Action:  search("USD to CNY rate today")
  Observation: 7.12
  Thought: 有了汇率，可以计算了
  Action:  calculate("1000 * 7.12")
  Observation: 7120
  Thought: 可以给出最终答案
  Final Answer: 1000 美元约等于 7120 人民币

Plan: [
  "查询公司 2025 年财报",
  "提取营收与净利润",
  "计算同比增长率",
  "生成对比图表数据",
]
Execute: 逐步执行，每步结果可触发 Replan

Attempt 1: 失败 → Reflection: "我漏了检查参数单位，下次应先确认单位"
Attempt 2: 带上反思重试 → 成功

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 消息列表，自动追加
    intent: str                               # 意图分类结果
    attempts: int                             # 重试计数
    needs_human: bool                         # 是否需要人工

def classify(state: AgentState) -> dict:
    """意图分类节点"""
    intent = classify_intent(state["messages"][-1].content)
    return {"intent": intent}

def route(state: AgentState) -> str:
    """条件边：按意图路由到不同处理节点"""
    if state["intent"] == "refund":
        return "refund_handler"
    if state["intent"] == "tech":
        return "tech_handler"
    return "general_handler"

def guard(state: AgentState) -> dict:
    """兜底节点：重试超限则转人工"""
    if state["attempts"] >= 3:
        return {"needs_human": True}
    return {"attempts": state["attempts"] + 1}

builder = StateGraph(AgentState)
builder.add_node("classify", classify)
builder.add_node("refund_handler", refund_handler)
builder.add_node("tech_handler", tech_handler)
builder.add_node("general_handler", general_handler)
builder.add_node("guard", guard)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route,
    {"refund_handler": "refund_handler",
     "tech_handler": "tech_handler",
     "general_handler": "general_handler"})
builder.add_edge("refund_handler", "guard")
builder.add_edge("tech_handler", "guard")
builder.add_edge("general_handler", "guard")
builder.add_conditional_edges("guard",
    lambda s: "human" if s["needs_human"] else END,
    {"human": END})

graph = builder.compile(checkpointer=MemorySaver())

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(          # 生产用 Postgres 持久化，不用内存
    "postgresql://user:pass@db:5432/agent"
) as checkpointer:
    checkpointer.setup()                    # 建表，首次运行
    graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "session-8821"}}
graph.invoke({"messages": [("user", "帮我退掉订单 A123")]}, config)  # 首次执行

state = graph.get_state(config)
if state.next:                             # 服务重启后，还有未完成的节点
    graph.invoke(None, config)             # 传入 None 表示从检查点续跑

from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(          # 子 Agent：聚焦的提示与工具集
    model, tools=[web_search, read_pdf],
    prompt="你是研究员，负责查找与整理资料，不写最终答案。",
)
writer_agent = create_react_agent(
    model, tools=[write_file],
    prompt="你是撰稿人，只负责把资料组织成文，不检索。",
)


def supervisor(state: AgentState) -> dict:
    """主管 Agent：只做委派决策，不直接调工具"""
    decision = model.chat(
        system="根据任务决定下一步交给 research 还是 writer，"
               "若已完成则回答 FINISH。",
        user=state["messages"][-1].content,
    )
    return {"next": decision.content.strip()}

def delegate(state: AgentState) -> dict:
    target = research_agent if state["next"] == "research" else writer_agent
    result = target.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

import hashlib, json

class LoopGuard:
    def __init__(self, max_repeats: int = 2, max_steps: int = 15):
        self.seen: list[str] = []
        self.max_repeats = max_repeats
        self.max_steps = max_steps

    def _fingerprint(self, tool: str, args: dict) -> str:
        payload = json.dumps({"t": tool, "a": args}, sort_keys=True)
        return hashlib.sha256(payload.encode()).hexdigest()[:16]

    def check(self, step: int, tool: str, args: dict) -> tuple[bool, str]:
        if step >= self.max_steps:
            return True, f"步数超限 {self.max_steps}"
        fp = self._fingerprint(tool, args)
        if self.seen.count(fp) >= self.max_repeats:
            return True, f"检测到重复调用 {tool}"
        self.seen.append(fp)
        return False, "ok"

from langgraph.types import interrupt

def refund_node(state: AgentState) -> dict:
    amount = state["refund_amount"]
    if amount > 500:
        # 中断图执行，等待人工审批
        decision = interrupt({
            "type": "approval",
            "amount": amount,
            "reason": "退款金额超过自动审批阈值 500 元",
        })
        if decision != "approved":
            return {"messages": [("assistant", "退款申请已提交人工审核。")]}
    return execute_refund(state)

class Budget:
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens
        self.used = 0

    def spend(self, tokens: int) -> None:
        self.used += tokens

    @property
    def remaining(self) -> int:
        return self.max_tokens - self.used

    def should_degrade(self) -> bool:
        return self.remaining < self.max_tokens * 0.2      # 剩余不足 20%