825 lines
34 KiB
Markdown
825 lines
34 KiB
Markdown
---
|
||
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 周
|
||
|
||
<!-- week-topics:start -->
|
||
|
||
## 本周相关专题
|
||
|
||
> 本节由 `tools/gen_week_topics.py` 根据专题文件的 `week` 字段自动生成，请勿手改；想调整就修改对应专题文件的 frontmatter 后重新生成。
|
||
|
||
- **AI 编程练习（每周）**：本周至少一个功能用 AI 编程工具（Cursor / Claude Code / Codex 等）完成，按 [AI 编程：工作流](../topics/ai/ai-coding/01-workflow-rules-skills.md) 写清需求和规则，按 [验证与证据](../topics/ai/ai-coding/02-verify-evidence.md) 保存测试结果和会话记录（脱敏后），第 15 周整理成“AI 编程 + 验证方法”的应聘案例。
|
||
|
||
| 技术栈 | 课程 / 场景 / 面试题 |
|
||
| --- | --- |
|
||
| [Agent 设计](../topics/ai/agent-design/index.md) | 课程：[Agent 推理模式](../topics/ai/agent-design/01-react-planning.md)；课程：[上下文工程、Skills、多 Agent 和 A2A](../topics/ai/agent-design/02-context-skills-multiagent.md)；场景：[运营报表 Agent](../topics/ai/agent-design/scenarios/report-agent.md)；[面试题](../topics/ai/agent-design/interview.md) |
|
||
| [LangChain](../topics/ai/langchain/index.md) | 课程：[LangChain 1.x](../topics/ai/langchain/01-lcel-structured-agent.md)；场景：[工单自动分类服务](../topics/ai/langchain/scenarios/ticket-triage.md)；[面试题](../topics/ai/langchain/interview.md) |
|
||
| [LangGraph](../topics/ai/langgraph/index.md) | 课程：[LangGraph 状态图](../topics/ai/langgraph/01-state-graph.md)；[面试题](../topics/ai/langgraph/interview.md) |
|
||
| [MCP 模型上下文协议](../topics/ai/mcp/index.md) | 课程：[MCP 协议和服务开发](../topics/ai/mcp/01-protocol-server.md)；场景：[企业内部 MCP 服务](../topics/ai/mcp/scenarios/internal-tools-gateway.md)；[面试题](../topics/ai/mcp/interview.md) |
|
||
| [提示词与上下文工程](../topics/ai/prompt-engineering/index.md) | 课程：[提示词注入和防护](../topics/ai/prompt-engineering/02-context-injection.md) |
|
||
|
||
**跨技术栈场景**：[审批流 + 人工介入的 Agent](../scenarios/approval-human-in-loop.md)、[工单系统](../scenarios/ticket-system.md)
|
||
|
||
<!-- week-topics:end -->
|