---
id: plan.week11
title: "第 11 周：项目三 M2——用 LangGraph 实现多 Agent 编排"
type: week
domain: plan
stack: [langchain, langgraph, mcp, prompt-engineering]
week: [11]
est_hours: 12
order: 11
---

# 第 11 周：项目三 M2——用 LangGraph 实现多 Agent 编排

> 项目三的第 2 周。上周 `week10-agent-m1-tools-mcp.md`（手写版 + MCP），下周 `week12-agent-m3-persistence.md`（续跑、中断、后台作业、SSE、记忆）。

**本周目标**：用 LangGraph 把上周的手写循环重写一遍：状态（State）、节点、条件边、工具调用、结构化决定。手写版的每一部分都要在 LangGraph 里找到对应。

**接口以官方文档为准**：LangGraph 更新快，本周代码在 `langgraph 1.2`、`langchain-core 1.7` 上跑通。文档：https://docs.langchain.com/oss/python/langgraph/overview （Graph API：https://docs.langchain.com/oss/python/langgraph/graph-api ）

```bash
uv add langgraph langchain-core
uv add langchain langchain-openai     # 接真实模型时才需要（init_chat_model）
```

**本周新增文件**

```text
agent-mini/src/agent_mini/
├── models.py              # Decision、parse_verdict
├── llm.py                 # 真实模型 / 假模型
├── tools.py               # LangChain 版本地工具 + MCP 工具桥接
├── fakes.py               # 按角色回答的假模型（测试和 --fake 演示共用）
├── graph/
│   ├── __init__.py
│   ├── state.py           # State 定义（含 reducer）
│   ├── prompts.py
│   ├── routing.py         # 条件边：硬规则 + 协调者决定
│   ├── nodes.py           # coordinator / proposer / critic / ask_human / summarize
│   └── build.py           # 组装并编译图
└── cli.py                 # agent-mini graph / run
tests/
├── __init__.py
├── test_routing.py
├── test_graph.py
└── test_lc_tools.py
```

---

## 第 1 天：LangGraph 基础

### 学习内容

- 说明（对照上周手写版理解）：

| LangGraph 概念 | 是什么 | 手写版里对应 |
| --- | --- | --- |
| State | 一个 `TypedDict`，整个流程共享的数据 | `LoopState` |
| 节点（node） | 普通函数：读 State，返回**要更新的字段** | 协调者、专家各自的那段代码 |
| 边 / 条件边 | 节点之间的跳转；条件边由一个函数决定下一步 | `while` 循环里的 `if` |
| reducer | 字段怎么合并：默认覆盖，`Annotated[list, operator.add]` 表示追加 | `turns.append(...)` |
| `START` / `END` | 起点和终点 | 循环开始 / `break` |
| compile | 把图编译成可运行对象，支持 `invoke` / `stream` | — |
| checkpointer | 每个节点执行完自动保存状态（下周） | 写 JSON 文件 |

- 案例：一个最小的“提议 ↔ 评审”循环图，跑一遍看清节点、条件边、reducer：

```python
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph


class S(TypedDict):
    round: int
    log: Annotated[list[str], operator.add]  # reducer：节点返回的列表会被追加，而不是覆盖


def propose(s: S) -> dict:
    return {"round": s["round"] + 1, "log": [f"提议 v{s['round'] + 1}"]}


def review(s: S) -> dict:
    ok = s["round"] >= 2
    return {"log": ["评审：通过" if ok else "评审：再改改"]}


def after_review(s: S) -> str:
    return "done" if s["log"][-1].endswith("通过") else "again"


g = StateGraph(S)
g.add_node("propose", propose)
g.add_node("review", review)
g.add_edge(START, "propose")
g.add_edge("propose", "review")
g.add_conditional_edges("review", after_review, {"again": "propose", "done": END})
app = g.compile()

for update in app.stream({"round": 0, "log": []}, stream_mode="updates"):
    print(update)   # 每个节点执行完输出它返回的更新
print(app.invoke({"round": 0, "log": []})["log"])
print(app.get_graph().draw_mermaid())   # 导出 mermaid，可以贴进文档
```

- 预置组件：`langgraph.prebuilt` 里有 `ToolNode`（执行工具调用）、LangChain 的 `create_agent`（现成的“模型 + 工具”循环）。知道有、什么时候用；本项目的专家节点里**自己写一个小工具循环**，这样每一步可控、好测试。

**易错点 / 面试常问（补充）**

- 易错点：节点返回的是“要更新的字段”，不是整个状态；列表字段如果要追加而不是覆盖，要在 State 里用 `Annotated[list, operator.add]` 这类 reducer 声明。
- 易错点：图有环时要设退出条件；LangGraph 还有 `recursion_limit`（默认步数上限），超过会抛错，不要靠它当正常的结束条件。
- 面试常问：LangGraph 和 LangChain 的关系（LangChain 提供模型、工具等组件；LangGraph 负责有状态、可循环、可持久化的流程编排）；为什么用图而不是写死的 if-else（流程可视化、可中断、可从检查点恢复）。

### 要建的文件

- `scratch/langgraph_basics.py`（上面的代码，不进最终项目）

### 运行方式

```bash
uv run python scratch/langgraph_basics.py
```

### 完成标准

- 能说出 State、节点、条件边、reducer 和手写版的对应关系；能把 mermaid 图贴进 Markdown 看到图。

---

## 第 2 天：State、结构化决定、路由规则

### 学习内容

- 说明：
  - State 里放什么：问题、轮数、发言记录（追加型）、最近一次评审结论、协调者的决定、人的回答（追加型）、最终结论、结束原因、长期记忆（下周）。
  - **路由函数是纯函数**：输入 State，输出下一个节点名——最容易测试，也是“结束条件写在代码里”的落地点。硬规则顺序：评审已通过 → 总结；达到最大轮数 → 总结；否则按协调者决定。
- 案例：

```python
# 文件：agent-mini/src/agent_mini/models.py
import re
from typing import Literal

from pydantic import BaseModel, Field


class Decision(BaseModel):
    action: Literal["dispatch", "ask_human", "finish"]
    target: Literal["proposer", "critic"] | None = None
    instruction: str = ""
    reason: str = Field(min_length=1)


def parse_decision(raw: str) -> Decision:
    m = re.search(r"\{.*\}", raw, re.S)
    return Decision.model_validate_json(m.group(0) if m else raw)


def parse_verdict(text: str) -> Literal["APPROVE", "REVISE"]:
    m = re.search(r"VERDICT:\s*(APPROVE|REVISE)\s*$", text.strip())
    return "APPROVE" if m and m.group(1) == "APPROVE" else "REVISE"
```

```python
# 文件：agent-mini/src/agent_mini/graph/__init__.py
```

```python
# 文件：agent-mini/src/agent_mini/graph/state.py
import operator
from typing import Annotated, Literal, TypedDict


class Turn(TypedDict):
    role: str
    instruction: str
    content: str
    tools: list[str]  # 这一步调用过的工具（审计用）


class AgentState(TypedDict, total=False):
    question: str
    user_id: str
    max_rounds: int
    round: int
    turns: Annotated[list[Turn], operator.add]
    last_verdict: Literal["APPROVE", "REVISE"] | None
    decision: dict | None
    human_answers: Annotated[list[str], operator.add]
    memories: list[str]
    final_answer: str
    end_reason: str
```

```python
# 文件：agent-mini/src/agent_mini/graph/routing.py
from typing import Literal

from agent_mini.graph.state import AgentState

Next = Literal["proposer", "critic", "ask_human", "summarize"]


def hard_stop(state: AgentState) -> str | None:
    """硬规则：返回结束原因；不需要结束时返回 None。"""
    if state.get("last_verdict") == "APPROVE":
        return "consensus"
    if state.get("round", 0) >= state.get("max_rounds", 3):
        return "max_rounds"
    return None


def route_after_coordinator(state: AgentState) -> Next:
    if hard_stop(state):
        return "summarize"
    d = state.get("decision") or {}
    if d.get("action") == "dispatch":
        return "critic" if d.get("target") == "critic" else "proposer"
    if d.get("action") == "ask_human":
        return "ask_human"
    return "summarize"
```

```python
# 文件：agent-mini/tests/test_routing.py
import pytest

from agent_mini.graph.routing import route_after_coordinator


@pytest.mark.parametrize("state, expected", [
    ({"last_verdict": "APPROVE", "decision": {"action": "dispatch", "target": "proposer"}}, "summarize"),
    ({"round": 3, "max_rounds": 3, "decision": {"action": "dispatch", "target": "proposer"}}, "summarize"),
    ({"round": 1, "decision": {"action": "dispatch", "target": "critic"}}, "critic"),
    ({"round": 1, "decision": {"action": "dispatch", "target": None}}, "proposer"),
    ({"round": 1, "decision": {"action": "ask_human"}}, "ask_human"),
    ({"round": 1, "decision": {"action": "finish"}}, "summarize"),
    ({"round": 0}, "summarize"),   # 没有决定（异常情况）时保守结束
])
def test_route(state, expected):
    assert route_after_coordinator(state) == expected
```

**易错点 / 面试常问（补充）**

- 易错点：让模型输出结构化决定时，一定要校验（本项目用 pydantic），并在解析失败时有兜底路由，不要让一次格式错误让整个流程崩溃。
- 易错点：路由函数要是纯函数，只根据状态决定下一步，这样才能单独测试。
- 面试常问：“谁来决定下一步”有几种做法（规则路由、模型决定、两者结合），各自的可控性和灵活性。

### 要建的文件

- `models.py`、`graph/__init__.py`、`graph/state.py`、`graph/routing.py`、`tests/test_routing.py`

### 运行方式

```bash
uv run pytest tests/test_routing.py -v
```

### 完成标准

- 每条路由规则都有用例；能解释为什么“硬规则优先于模型决定”。

---

## 第 3 天：模型、工具、提示词、节点

### 学习内容

- 说明：
  - **模型**：真实模型用 LangChain 的 `init_chat_model`（OpenAI 兼容接口填 `base_url`）；测试用自己写的假模型（继承 `BaseChatModel`，按角色返回预设回答，支持 `bind_tools`）。
  - **工具**：本地工具用 `@tool` 装饰器；MCP 工具用 `StructuredTool` 包一层（参数 Schema 直接用 MCP 给的 JSON Schema）。
  - **节点**：每个节点只做一件事，返回要更新的字段。
    - 协调者：先检查硬规则（命中就不调用模型，省钱）；否则让模型输出 JSON 决定，校验失败重试一次。
    - 提议者：带工具的小循环（`bind_tools` → 模型要调用工具 → 执行 → `ToolMessage` 回填 → 再问），有最大步数。
    - 评审者：解析 `VERDICT`。
    - 问人：`interrupt()`（下周配合检查点使用）。
    - 总结者：生成结论和结束原因。
  - **上下文工程**：每个节点只给模型需要的东西——问题、最近几条发言、本次指令、相关记忆，而不是全部历史。
- 案例：

```python
# 文件：agent-mini/src/agent_mini/llm.py
import os
from collections.abc import Callable
from typing import Any

from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatGeneration, ChatResult


class FakeChatModel(BaseChatModel):
    """测试用：responder 根据收到的消息决定回答（可以带 tool_calls）。"""

    responder: Callable[[list[BaseMessage]], AIMessage]
    calls: int = 0

    @property
    def _llm_type(self) -> str:
        return "fake-role"

    def _generate(self, messages: list[BaseMessage], stop: list[str] | None = None,
                  run_manager: Any = None, **kwargs: Any) -> ChatResult:
        self.calls += 1
        return ChatResult(generations=[ChatGeneration(message=self.responder(messages))])

    def bind_tools(self, tools: Any, **kwargs: Any) -> "FakeChatModel":  # type: ignore[override]
        return self


def build_chat_model(model: str | None = None) -> BaseChatModel:
    """真实模型：OpenAI 兼容接口。需要 uv add langchain langchain-openai。参数以官方文档为准。"""
    from langchain.chat_models import init_chat_model

    return init_chat_model(
        model or os.environ["CHAT_MODEL"], model_provider="openai",
        base_url=os.environ["LLM_BASE_URL"], api_key=os.environ["LLM_API_KEY"],
        temperature=0.2, timeout=60, max_retries=2,
    )
```

```python
# 文件：agent-mini/src/agent_mini/tools.py
from datetime import date

from langchain_core.tools import BaseTool, StructuredTool, tool
from mcp import Client

from agent_mini.handwritten.tools import safe_eval


@tool
def calculator(expression: str) -> str:
    """计算四则运算，例如 (3+4)*2。需要做数学计算时使用。"""
    return str(safe_eval(expression))


@tool
def today() -> str:
    """返回今天的日期（YYYY-MM-DD）。"""
    return date.today().isoformat()


LOCAL_TOOLS: list[BaseTool] = [calculator, today]


async def mcp_tools_for_langchain(client: Client, allow: set[str]) -> list[BaseTool]:
    """把 MCP 工具包装成 LangChain 工具（也可以用官方的 MCP 适配器，接口以文档为准）。"""
    out: list[BaseTool] = []
    for t in (await client.list_tools()).tools:
        if t.name not in allow:
            continue

        async def run(_name: str = t.name, **kwargs) -> str:
            r = await client.call_tool(_name, kwargs)
            text = "\n".join(c.text for c in r.content if getattr(c, "type", "") == "text")
            return f"工具出错：{text}" if r.is_error else text

        out.append(StructuredTool.from_function(coroutine=run, name=t.name,
                                                description=t.description or "", args_schema=t.input_schema))
    return out
```

```python
# 文件：agent-mini/src/agent_mini/graph/prompts.py
COORDINATOR = """你是协调者，负责决定下一步由谁做什么。规则：
- 还没有方案时派给提议者（proposer）；有新方案时派给评审者（critic）；评审要求修改时派回提议者。
- 问题缺少关键信息（预算、规模、约束）且无法合理假设时，选择 ask_human。
只输出 JSON：{"action": "dispatch|ask_human|finish", "target": "proposer|critic|null", "instruction": "具体指令", "reason": "理由"}"""

PROPOSER = "你是提议者。给出具体、可执行的方案；需要事实或计算时使用工具，不要编造。"

CRITIC = "你是评审者。检查方案的成本、风险、可行性。最后一行必须是 VERDICT: APPROVE 或 VERDICT: REVISE。"

SUMMARIZER = "你是总结者。根据讨论给出最终结论（不超过 300 字），列出关键理由和主要风险。"


def context(state: dict, last_n: int = 4) -> str:
    turns = state.get("turns", [])[-last_n:]
    recent = "\n".join(f"[{t['role']}] {t['content'][:500]}" for t in turns) or "（无）"
    memories = "\n".join(f"- {m}" for m in state.get("memories", [])) or "（无）"
    answers = "\n".join(f"- {a}" for a in state.get("human_answers", [])) or "（无）"
    return f"问题：{state['question']}\n用户补充：\n{answers}\n相关记忆：\n{memories}\n最近讨论：\n{recent}"
```

```python
# 文件：agent-mini/src/agent_mini/graph/nodes.py
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import BaseTool
from langgraph.types import interrupt
from pydantic import ValidationError

from agent_mini.graph import prompts
from agent_mini.graph.routing import hard_stop
from agent_mini.graph.state import AgentState
from agent_mini.models import parse_decision, parse_verdict


class Nodes:
    def __init__(self, model: BaseChatModel, tools: list[BaseTool] | None = None, max_tool_steps: int = 4):
        self.model, self.tools, self.max_tool_steps = model, tools or [], max_tool_steps

    async def coordinator(self, state: AgentState) -> dict:
        if hard_stop(state):
            return {"decision": {"action": "finish", "reason": "命中硬规则"}}  # 不调用模型
        messages = [SystemMessage(prompts.COORDINATOR), HumanMessage(prompts.context(state))]
        for attempt in range(2):
            raw = str((await self.model.ainvoke(messages)).content)
            try:
                return {"decision": parse_decision(raw).model_dump()}
            except ValidationError as e:
                if attempt == 1:
                    raise
                messages += [HumanMessage(f"格式错误：{e.errors()[0]['msg']}。只输出合法 JSON。")]
        raise AssertionError("unreachable")

    async def proposer(self, state: AgentState) -> dict:
        instruction = (state.get("decision") or {}).get("instruction", "")
        messages = [SystemMessage(prompts.PROPOSER), HumanMessage(f"{prompts.context(state)}\n\n指令：{instruction}")]
        bound = self.model.bind_tools(self.tools) if self.tools else self.model
        by_name = {t.name: t for t in self.tools}
        used: list[str] = []
        for _ in range(self.max_tool_steps):
            ai = await bound.ainvoke(messages)
            messages.append(ai)
            if not ai.tool_calls:
                break
            for call in ai.tool_calls:
                tool = by_name.get(call["name"])
                try:
                    result = await tool.ainvoke(call["args"]) if tool else f"没有工具 {call['name']}"
                except Exception as e:  # 工具出错返回给模型
                    result = f"工具出错：{e}"
                used.append(call["name"])
                messages.append(ToolMessage(content=str(result), tool_call_id=call["id"]))
        content = str(messages[-1].content) if not getattr(messages[-1], "tool_calls", None) else "（工具调用次数超限）"
        return {"round": state.get("round", 0) + 1,
                "turns": [{"role": "proposer", "instruction": instruction, "content": content, "tools": used}]}

    async def critic(self, state: AgentState) -> dict:
        instruction = (state.get("decision") or {}).get("instruction", "")
        ai = await self.model.ainvoke([SystemMessage(prompts.CRITIC),
                                       HumanMessage(f"{prompts.context(state)}\n\n指令：{instruction}")])
        content = str(ai.content)
        return {"last_verdict": parse_verdict(content),
                "turns": [{"role": "critic", "instruction": instruction, "content": content, "tools": []}]}

    async def ask_human(self, state: AgentState) -> dict:
        # interrupt 会暂停整个图并把问题交给调用方；恢复时这个节点从头重新执行，
        # 所以 interrupt 之前不要放有副作用的操作
        question = (state.get("decision") or {}).get("instruction") or "请补充更多信息"
        answer = interrupt({"question": question})
        return {"human_answers": [f"{question} → {answer}"], "decision": None}

    async def summarize(self, state: AgentState) -> dict:
        ai = await self.model.ainvoke([SystemMessage(prompts.SUMMARIZER), HumanMessage(prompts.context(state, last_n=8))])
        reason = hard_stop(state) or "coordinator_finish"
        return {"final_answer": str(ai.content), "end_reason": reason}
```

**易错点 / 面试常问（补充）**

- 易错点：每个角色的提示词要写清“输出格式”，例如审查者最后一行必须是 `VERDICT: APPROVE / REVISE`，否则路由规则无法判断。
- 易错点：工具调用的结果要限制长度再放回上下文，太长会挤掉重要信息、增加费用。
- 面试常问：怎么让多个角色不互相“附和”（不同的系统提示词、审查者有明确的检查清单、最大轮数兜底）。

### 要建的文件

- `llm.py`、`tools.py`、`graph/prompts.py`、`graph/nodes.py`

### 运行方式

- 明天组装成图后一起测试。

### 完成标准

- 提示词全部自己写，不复制公司的。
- 能讲清每个节点“读了 State 的哪些字段、返回了哪些字段”。

---

## 第 4 天：组装图 + 测试

### 学习内容

- 说明：把节点和条件边组装起来；`checkpointer` 和 `store` 作为参数传入（本周传 `None`，下周接上）。用 `stream_mode="updates"` 逐个节点看执行过程。
- 案例：

```python
# 文件：agent-mini/src/agent_mini/graph/build.py
from langchain_core.language_models import BaseChatModel
from langchain_core.tools import BaseTool
from langgraph.graph import END, START, StateGraph

from agent_mini.graph.nodes import Nodes
from agent_mini.graph.routing import route_after_coordinator
from agent_mini.graph.state import AgentState


def build_graph(model: BaseChatModel, tools: list[BaseTool] | None = None, checkpointer=None, store=None):
    n = Nodes(model, tools)
    g = StateGraph(AgentState)
    g.add_node("coordinator", n.coordinator)
    g.add_node("proposer", n.proposer)
    g.add_node("critic", n.critic)
    g.add_node("ask_human", n.ask_human)
    g.add_node("summarize", n.summarize)
    g.add_edge(START, "coordinator")
    g.add_conditional_edges("coordinator", route_after_coordinator,
                            ["proposer", "critic", "ask_human", "summarize"])
    for node in ("proposer", "critic", "ask_human"):
        g.add_edge(node, "coordinator")
    g.add_edge("summarize", END)
    return g.compile(checkpointer=checkpointer, store=store)
```

```python
# 文件：agent-mini/src/agent_mini/fakes.py
import json

from langchain_core.messages import AIMessage, BaseMessage

from agent_mini.llm import FakeChatModel

ROLE_MARKERS = {"协调者": "coordinator", "提议者": "proposer", "评审者": "critic", "总结者": "summarizer",
                "记忆整理": "memory"}


def role_of(messages: list[BaseMessage]) -> str:
    system = str(messages[0].content)
    return next(role for marker, role in ROLE_MARKERS.items() if marker in system)


def decision(action: str, target: str | None = None, instruction: str = "请处理") -> str:
    return json.dumps({"action": action, "target": target, "instruction": instruction, "reason": "测试"})


def scripted_model(**queues: list) -> FakeChatModel:
    """每个角色一个回答队列；元素是字符串或 AIMessage（可以带 tool_calls）。队列用完时重复最后一个。"""
    q = {k: list(v) for k, v in queues.items()}

    def respond(messages: list[BaseMessage]) -> AIMessage:
        items = q[role_of(messages)]
        item = items.pop(0) if len(items) > 1 else items[0]
        return item if isinstance(item, AIMessage) else AIMessage(content=item)

    return FakeChatModel(responder=respond)


def happy_path_model() -> FakeChatModel:
    """提议者先调用计算器 → 评审要求修改 → 再提议 → 评审通过。"""
    return scripted_model(
        coordinator=[decision("dispatch", "proposer"), decision("dispatch", "critic"),
                     decision("dispatch", "proposer"), decision("dispatch", "critic")],
        proposer=[AIMessage(content="", tool_calls=[{"name": "calculator", "args": {"expression": "5*2000"}, "id": "t1"}]),
                  "方案一：每月预算 10000 元", "方案二：加上缓存一致性方案"],
        critic=["没考虑一致性\nVERDICT: REVISE", "可以\nVERDICT: APPROVE"],
        summarizer=["结论：采用方案二"],
        memory=['{"memories": []}'],
    )
```

```python
# 文件：agent-mini/tests/test_graph.py
from agent_mini.graph.build import build_graph
from agent_mini.tools import LOCAL_TOOLS
from agent_mini.fakes import decision, happy_path_model, scripted_model


async def test_consensus_with_tool_call():
    graph = build_graph(happy_path_model(), LOCAL_TOOLS)
    out = await graph.ainvoke({"question": "5 人团队怎么选技术栈", "max_rounds": 5, "round": 0})
    assert out["end_reason"] == "consensus" and out["round"] == 2
    assert out["turns"][0]["tools"] == ["calculator"]
    assert [t["role"] for t in out["turns"]] == ["proposer", "critic", "proposer", "critic"]
    assert out["final_answer"] == "结论：采用方案二"


async def test_max_rounds_and_stream_order():
    model = scripted_model(coordinator=[decision("dispatch", "proposer"), decision("dispatch", "critic")] * 2,
                           proposer=["方案"], critic=["VERDICT: REVISE"], summarizer=["到达上限"])
    graph = build_graph(model)
    nodes = []
    async for update in graph.astream({"question": "q", "max_rounds": 2, "round": 0}, stream_mode="updates"):
        nodes.extend(update.keys())
    assert nodes[nodes.index("summarize") - 1] == "coordinator"   # 最后由协调者（命中轮数上限）转到总结
    assert nodes.count("proposer") == 2


async def test_hard_stop_skips_model_call():
    model = scripted_model(coordinator=[decision("dispatch", "proposer")], proposer=["方案"],
                           critic=["好\nVERDICT: APPROVE"], summarizer=["结论"])
    graph = build_graph(model)
    await graph.ainvoke({"question": "q", "max_rounds": 1, "round": 0})
    # 协调者 1 次（派给提议者）+ 提议者 1 次 + 总结 1 次；第二次进入协调者时命中“轮数上限”，没有调用模型
    assert model.calls == 3


def test_mermaid():
    text = build_graph(happy_path_model()).get_graph().draw_mermaid()
    assert "coordinator" in text and "summarize" in text
```

```python
# 文件：agent-mini/tests/__init__.py
```

**易错点 / 面试常问（补充）**

- 易错点：测试里用脚本化的假模型，按角色给出固定回复，这样图的每条路径都能稳定测试，不花钱、不受模型随机性影响。
- 面试常问：Agent 怎么做自动化测试（假模型测流程和路由、评测集测效果、真实模型只在少量冒烟测试里用）。

### 要建的文件

- `graph/build.py`、`fakes.py`（放在 `src` 里，命令行的 `--fake` 演示也要用）、`tests/__init__.py`、`tests/test_graph.py`

### 运行方式

```bash
uv run pytest tests/test_graph.py -v
```

### 完成标准

- 测试通过；能解释 `test_hard_stop_skips_model_call` 为什么是 3 次调用。

---

## 第 5 天：MCP 工具接入 LangGraph + 命令行

### 学习内容

- 说明：上周的 MCP 服务通过 `mcp_tools_for_langchain` 变成 LangChain 工具，提议者就能“查知识库”。命令行提供两个子命令：导出图结构、流式运行（打印每个节点做了什么）。
- 案例：

```python
# 文件：agent-mini/tests/test_lc_tools.py
import httpx
from langchain_core.messages import AIMessage
from mcp import Client

import mcp_servers.knowledge_server as ks
from agent_mini.graph.build import build_graph
from agent_mini.tools import mcp_tools_for_langchain
from agent_mini.fakes import decision, scripted_model


async def test_proposer_uses_mcp_tool(monkeypatch):
    monkeypatch.setattr(ks, "RAG_TRANSPORT", httpx.MockTransport(lambda r: httpx.Response(200, json={
        "text": "团队熟悉 Vue 和 Go [1]", "citations": [{"n": 1, "filename": "team.md", "heading_path": "技术栈"}]})))
    async with Client(ks.mcp) as client:
        tools = await mcp_tools_for_langchain(client, allow={"search_knowledge"})
        model = scripted_model(
            coordinator=[decision("dispatch", "proposer"), decision("finish")],
            proposer=[AIMessage(content="", tool_calls=[{"name": "search_knowledge",
                                                         "args": {"query": "团队技术栈"}, "id": "k1"}]),
                      "根据知识库：前端 Vue、后端 Go"],
            summarizer=["结论：Vue + Go"],
        )
        out = await build_graph(model, tools).ainvoke({"question": "选技术栈", "max_rounds": 3, "round": 0})
    assert out["turns"][0]["tools"] == ["search_knowledge"]
    assert out["end_reason"] == "coordinator_finish"
```

```python
# 文件：agent-mini/src/agent_mini/cli.py
import argparse
import asyncio
import json

from agent_mini.graph.build import build_graph
from agent_mini.tools import LOCAL_TOOLS


def make_model(fake: bool):
    if fake:
        from agent_mini.fakes import happy_path_model  # 演示用假模型

        return happy_path_model()
    from agent_mini.llm import build_chat_model

    return build_chat_model()


async def run(question: str, fake: bool, max_rounds: int) -> None:
    graph = build_graph(make_model(fake), LOCAL_TOOLS)
    async for update in graph.astream({"question": question, "max_rounds": max_rounds, "round": 0},
                                      stream_mode="updates"):
        for node, data in update.items():
            print(f"== {node}")
            print(json.dumps(data, ensure_ascii=False, indent=2)[:800])


def main() -> None:
    p = argparse.ArgumentParser(prog="agent-mini")
    sub = p.add_subparsers(dest="cmd", required=True)
    sub.add_parser("graph")
    r = sub.add_parser("run")
    r.add_argument("question")
    r.add_argument("--fake", action="store_true", help="使用假模型（不需要密钥）")
    r.add_argument("--max-rounds", type=int, default=3)
    args = p.parse_args()
    if args.cmd == "graph":
        print(build_graph(make_model(True)).get_graph().draw_mermaid())
    else:
        asyncio.run(run(args.question, args.fake, args.max_rounds))


if __name__ == "__main__":  # 不自动运行
    main()
```

**易错点 / 面试常问（补充）**

- 易错点：把 MCP 工具转成 LangChain 工具时，参数 Schema 要原样保留，否则模型生成的参数会和服务端要求对不上。
- 易错点：本项目用自己写的桥接；以后换成官方适配（`langchain-mcp-adapters` 或 `langchain` 里的 MCP 支持）时，这些包更新快，接口以官方文档为准：https://docs.langchain.com/oss/python/langchain/mcp 。
- 面试常问：工具多了以后怎么办（按场景挑选少量工具、工具描述写清楚、对有副作用的工具要求确认）。

### 要建的文件

- `tests/test_lc_tools.py`、`cli.py`；`pyproject.toml` 里加 `[project.scripts] agent-mini = "agent_mini.cli:main"`

### 运行方式

```bash
uv run pytest -v
uv run agent-mini graph > docs/graph.mmd
uv run agent-mini run "5 人创业团队做内部工具该选什么技术栈" --fake
```

### 完成标准

- 测试通过；图结构导出到 `docs/graph.mmd` 并贴进 `docs/architecture.md`。

---

## 第 6 天：真实模型试跑和调提示词

### 学习内容

- 说明：用真实模型跑 5 个不同题目，观察：协调者是否按格式输出、评审是否太容易通过（总是第一轮就 APPROVE）、提议者是否乱调工具、平均轮数和耗时。每次改提示词记一行到 `docs/prompt-changelog.md`，第 13 周评测时会用到。
- 案例：`docs/prompt-changelog.md` 的格式

```markdown
| 日期 | 改了什么 | 为什么 | 观察到的变化（以实际运行为准） |
| --- | --- | --- | --- |
| 2026-xx-xx | 评审者提示词加“至少指出一个风险” | 评审总是第一轮就通过 | 平均轮数从 x 变成 y |
```

### 要建的文件

- `docs/samples/langgraph-*.md`（真实运行记录，检查无敏感信息）、`docs/prompt-changelog.md`

### 运行方式

```bash
export LLM_BASE_URL=... LLM_API_KEY=... CHAT_MODEL=...    # 写在 .env，不提交
uv run agent-mini run "给读书 App 设计会员规则"
```

### 完成标准

- 5 个题目都跑完；changelog 至少 2 条。

---

## 第 7 天：对照文档 + M2 验收

### 学习内容

- 说明：补完 `docs/handwritten-vs-langgraph.md` 的另一半——手写版每一部分在 LangGraph 里对应什么，框架帮你做了什么，又有哪些事框架不管、还得自己做。
- 案例：

```markdown
| 关注点 | 手写版 | LangGraph 版 | 框架没管、要自己做的 |
| --- | --- | --- | --- |
| 循环 / 跳转 | while + if | 条件边 | 结束条件（硬规则）仍要自己写 |
| 状态合并 | 手动 append | reducer | — |
| 持久化 | JSON 文件 | checkpointer（下周） | 业务状态表、权限、审计 |
| 工具调用 | 手写循环 | bind_tools + 节点内循环 / ToolNode | 参数白名单、副作用幂等 |
| 中断问人 | 自己存 awaiting_input | interrupt + Command(resume) | 前端交互、超时处理 |
| 防重复执行 | 进程内集合 | 无 | 租约（下周） |
```

### 要建的文件

- `docs/handwritten-vs-langgraph.md`

### 运行方式

```bash
uv run ruff check . && uv run pyright src && uv run pytest -v
```

### 完成标准（M2 验收）

- [ ] LangGraph 版能跑完一次带工具调用的多轮讨论，结束条件正确
- [ ] 路由规则、图的主要路径都有测试（假模型，不需要密钥）
- [ ] MCP 工具接入了 LangGraph
- [ ] 能讲清 State、节点、条件边、reducer，以及和手写版的对应关系
- [ ] 在总纲进度表勾掉第 11 周