1062 lines
46 KiB
Markdown
1062 lines
46 KiB
Markdown
---
|
||
id: plan.week10
|
||
title: "第 10 周：项目三 M1——手写工具调用循环 + 手写多 Agent 循环 + MCP 服务"
|
||
type: week
|
||
domain: plan
|
||
stack: [mcp, function-calling]
|
||
week: [10]
|
||
est_hours: 12
|
||
order: 10
|
||
---
|
||
|
||
# 第 10 周：项目三 M1——手写工具调用循环 + 手写多 Agent 循环 + MCP 服务
|
||
|
||
> 第三段的第 3 周，**项目三 `agent-mini`（阶段 3 主交付）** 的第 1 周。上周 `week09-case-cards.md`，下周 `week11-agent-m2-langgraph.md`。
|
||
|
||
## 项目三总览
|
||
|
||
**做什么**：把你在公司见过的多 Agent 产品“削”成一个**没有业务外壳**、可公开演示的内核，证明你“会造”，而不只是“会改”。
|
||
|
||
做完后（第 13 周）要能演示：
|
||
|
||
1. 提一个问题 → 协调者派活给提议者、评审者 → 专家可以调用工具（包括通过 **MCP** 调用你的 RAG 知识库）→ 多轮讨论 → 达成共识或到达最大轮数后结束 → 生成可审计的报告。
|
||
2. 中途 `kill -9` 杀掉服务进程 → 重启 → 任务从最近的检查点接着跑完。
|
||
3. 浏览器通过 SSE 看进度；关掉页面任务照跑；重新打开能补齐漏掉的事件。
|
||
4. 信息不够时停下来问人（`interrupt`），人回答后继续；模型失败后换模型从失败点重试。
|
||
5. 记得用户的偏好（长期记忆），下次提问时用上。
|
||
6. 有评测集，能用数字说明改动的效果；有可观测性（每次运行的耗时、调用了什么）。
|
||
|
||
**节奏**
|
||
|
||
| 周 | 里程碑 | 内容 |
|
||
| --- | --- | --- |
|
||
| 10 | M1 | **手写**工具调用循环、手写多 Agent 循环（理解原理）、写 MCP 服务 |
|
||
| 11 | M2 | 用 **LangGraph** 重写编排 |
|
||
| 12 | M3 | 检查点续跑、`interrupt`、业务表 + 租约、后台作业、SSE、长期记忆 |
|
||
| 13 | M4 | 评测集、可观测性（LangSmith）、文档和映射表、Dify / Coze 体验 |
|
||
|
||
**范围**
|
||
|
||
| 必须有 | 禁止有（本阶段） |
|
||
| --- | --- |
|
||
| 协调者 + 提议者 + 评审者；结束条件写在代码里 | 照搬公司 UI、提示词或代码 |
|
||
| 工具调用，至少一个工具来自自己写的 MCP 服务 | 依赖公司内网才能演示 |
|
||
| 检查点持久化 + 重启续跑；编排在服务端后台 | LangGraph 之外再引入别的 Agent 框架 |
|
||
| SSE 断线补发；人机协作；长期记忆；评测集 | |
|
||
| **不需要真实模型密钥也能跑通测试和演示**（假模型） | |
|
||
|
||
讨论题目用通用例子（“5 人创业团队做内部工具该选什么技术栈”“给读书 App 设计会员规则”），不要用公司业务题目。
|
||
|
||
**技术选型**：Python 3.12 + uv、httpx、pydantic；MCP 官方 Python SDK（2.x）；LangGraph + langchain-core；FastAPI + SSE；SQLite（检查点、业务表，生产换 PostgreSQL）；pytest。
|
||
|
||
> **版本提醒**：LangGraph、LangChain、MCP SDK 更新很快。本计划的代码在 2026-10 用 `mcp 2.3`、`langgraph 1.2`、`langchain-core 1.7` 实际跑通。**动手前先看官方文档当前版本**，导入路径或参数有出入时按文档改：
|
||
> - MCP Python SDK：https://py.sdk.modelcontextprotocol.io/ （v1 → v2 迁移说明：https://py.sdk.modelcontextprotocol.io/v2/migration/ ）
|
||
> - MCP 协议：https://modelcontextprotocol.io/
|
||
> - LangGraph：https://docs.langchain.com/oss/python/langgraph/overview
|
||
|
||
### 最终项目结构（4 周完成后）
|
||
|
||
```text
|
||
agent-mini/
|
||
├── README.md
|
||
├── pyproject.toml
|
||
├── .env.example # 默认 LLM_PROVIDER=fake
|
||
├── Dockerfile / docker-compose.yml # 第 13 周
|
||
├── docs/ # 架构、设计取舍、映射表、对照文档、讲解稿
|
||
├── mcp_servers/
|
||
│ ├── __init__.py
|
||
│ └── knowledge_server.py # 第 10 周：MCP 服务（查知识库、建待办、资源、提示词）
|
||
├── src/agent_mini/
|
||
│ ├── __init__.py
|
||
│ ├── handwritten/ # 第 10 周：手写版（保留作对照）
|
||
│ │ ├── __init__.py
|
||
│ │ ├── llm.py # 最小模型客户端（支持 tools）+ 脚本化假模型
|
||
│ │ ├── tools.py # 工具定义、安全计算器、日期工具
|
||
│ │ ├── tool_loop.py # 手写工具调用循环
|
||
│ │ ├── multi_agent.py # 手写协调者 + 两个专家循环（JSON 文件持久化）
|
||
│ │ └── mcp_bridge.py # 把 MCP 工具接进手写循环
|
||
│ ├── models.py # 第 11 周：Decision 等结构
|
||
│ ├── llm.py # 第 11 周：真实 / 假对话模型
|
||
│ ├── graph/ # 第 11~12 周：LangGraph 版
|
||
│ ├── store/ # 第 12 周：runs / events 业务表、租约
|
||
│ ├── runner.py # 第 12 周：后台运行、看门狗
|
||
│ ├── api/ # 第 12 周：FastAPI + SSE
|
||
│ └── observability.py # 第 13 周：运行指标
|
||
├── evals/ # 第 13 周
|
||
├── web/index.html # 第 12 周：极简页面
|
||
└── tests/
|
||
```
|
||
|
||
---
|
||
|
||
## 第 1 天：初始化 + 支持工具调用的最小模型客户端
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 消息有四种角色：`system`（规则）、`user`（用户）、`assistant`（模型）、`tool`（工具执行结果）。
|
||
- **工具调用（Function / Tool Calling）**的本质：你把“有哪些工具、每个工具的参数格式（JSON Schema）”发给模型；模型**不执行任何东西**，只返回“我想调用哪个工具、参数是什么”（`tool_calls`）；**你的代码**执行工具，把结果作为 `tool` 消息追加回去，再问模型，直到它给出最终回答。
|
||
- 为了测试，写一个**脚本化假模型**：按预先写好的顺序返回回答或工具调用。
|
||
- 案例：
|
||
|
||
```bash
|
||
uv init --package agent-mini && cd agent-mini
|
||
uv add httpx pydantic pydantic-settings "mcp>=2"
|
||
uv add --dev pytest pytest-asyncio ruff pyright
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/__init__.py
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/__init__.py
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/llm.py
|
||
import json
|
||
from dataclasses import dataclass, field
|
||
from typing import Protocol
|
||
|
||
import httpx
|
||
|
||
|
||
@dataclass
|
||
class ToolCall:
|
||
id: str
|
||
name: str
|
||
arguments: str # 模型给的是 JSON 字符串，可能不合法，要自己校验
|
||
|
||
|
||
@dataclass
|
||
class AssistantMessage:
|
||
content: str | None = None
|
||
tool_calls: list[ToolCall] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> dict:
|
||
"""转回 OpenAI 格式，追加到消息列表里。"""
|
||
d: dict = {"role": "assistant", "content": self.content}
|
||
if self.tool_calls:
|
||
d["tool_calls"] = [
|
||
{"id": c.id, "type": "function", "function": {"name": c.name, "arguments": c.arguments}}
|
||
for c in self.tool_calls
|
||
]
|
||
return d
|
||
|
||
|
||
class ChatLLM(Protocol):
|
||
async def chat(self, messages: list[dict], tools: list[dict] | None = None) -> AssistantMessage: ...
|
||
|
||
|
||
class OpenAICompatLLM:
|
||
def __init__(self, client: httpx.AsyncClient, model: str):
|
||
self.client, self.model = client, model
|
||
|
||
async def chat(self, messages: list[dict], tools: list[dict] | None = None) -> AssistantMessage:
|
||
body: dict = {"model": self.model, "messages": messages, "temperature": 0.2}
|
||
if tools:
|
||
body["tools"] = tools
|
||
resp = await self.client.post("/chat/completions", json=body)
|
||
resp.raise_for_status()
|
||
msg = resp.json()["choices"][0]["message"]
|
||
calls = [ToolCall(c["id"], c["function"]["name"], c["function"]["arguments"]) for c in msg.get("tool_calls") or []]
|
||
return AssistantMessage(msg.get("content"), calls)
|
||
|
||
|
||
class ScriptedLLM:
|
||
"""按顺序返回预设的回答；记录每次收到的消息，方便断言。"""
|
||
|
||
def __init__(self, script: list[AssistantMessage]):
|
||
self.script = list(script)
|
||
self.calls: list[list[dict]] = []
|
||
|
||
async def chat(self, messages: list[dict], tools: list[dict] | None = None) -> AssistantMessage:
|
||
self.calls.append(json.loads(json.dumps(messages))) # 深拷贝，防止后续修改影响断言
|
||
if not self.script:
|
||
return AssistantMessage(content="（脚本已用完）")
|
||
return self.script.pop(0)
|
||
|
||
|
||
def call(name: str, args: dict | str, id: str = "c1") -> AssistantMessage:
|
||
"""测试辅助：构造一条“想调用工具”的模型回复。"""
|
||
return AssistantMessage(tool_calls=[ToolCall(id, name, args if isinstance(args, str) else json.dumps(args))])
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：OpenAI 兼容接口里，模型要调工具时返回的 `tool_calls[].function.arguments` 是 **JSON 字符串**，不是对象，要自己 `json.loads`，并处理解析失败。
|
||
- 易错点：把工具结果发回模型时，消息角色是 `tool`，并且要带上对应的 `tool_call_id`；漏了会报错或让模型对不上号。
|
||
- 面试常问：Function Calling 的完整流程（带工具定义请求 → 模型返回要调用的工具和参数 → 程序执行 → 把结果发回 → 模型生成最终回答）；模型自己会执行工具吗（不会，执行的永远是你的程序）。
|
||
|
||
### 要建的文件
|
||
|
||
- `pyproject.toml`（加 `[tool.pytest.ini_options] asyncio_mode = "auto"`）、`.env.example`、`.gitignore`
|
||
- `src/agent_mini/__init__.py`、`handwritten/__init__.py`、`handwritten/llm.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run python -c "import agent_mini.handwritten.llm as m; print(m.call('now', {}).to_dict())"
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 能画出工具调用的消息往返：`user → assistant(tool_calls) → tool → assistant(最终回答)`。
|
||
|
||
---
|
||
|
||
## 第 2 天：手写工具调用循环
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 每个工具 = 名字 + 说明 + 参数模型 + 执行函数。**说明写得好不好直接决定模型会不会用对**——写清楚“什么时候用、参数什么意思”。
|
||
- 可靠性要点：
|
||
1. 参数用 pydantic 校验；不合法时**把错误返回给模型**让它自己改，而不是整个流程崩溃。
|
||
2. 工具执行出错同理，返回错误信息。
|
||
3. **限制最大步数**，防止模型来回调用停不下来。
|
||
4. 计算器**绝不能用 `eval`**（会执行任意代码），用 `ast` 只允许数字和四则运算。
|
||
5. 有副作用的工具（建待办、发邮件）要幂等、可审计（第 4 天）。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/tools.py
|
||
import ast
|
||
import inspect
|
||
import operator
|
||
from collections.abc import Awaitable, Callable
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, Field
|
||
|
||
|
||
@dataclass
|
||
class Tool:
|
||
name: str
|
||
description: str
|
||
parameters: dict # JSON Schema
|
||
func: Callable[[dict], Any | Awaitable[Any]]
|
||
args_model: type[BaseModel] | None = None # 本地工具用 pydantic 校验；MCP 工具由服务端校验
|
||
|
||
def schema(self) -> dict:
|
||
return {"type": "function", "function": {"name": self.name, "description": self.description,
|
||
"parameters": self.parameters}}
|
||
|
||
async def run(self, args: dict) -> str:
|
||
if self.args_model is not None:
|
||
args = self.args_model.model_validate(args).model_dump() # 不合法会抛 ValidationError
|
||
result = self.func(args)
|
||
if inspect.isawaitable(result):
|
||
result = await result
|
||
return str(result)
|
||
|
||
|
||
def local_tool(name: str, description: str, args_model: type[BaseModel], func: Callable[[dict], Any]) -> Tool:
|
||
return Tool(name, description, args_model.model_json_schema(), func, args_model)
|
||
|
||
|
||
_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
|
||
ast.Div: operator.truediv, ast.USub: operator.neg}
|
||
|
||
|
||
def safe_eval(expr: str) -> float:
|
||
def ev(node: ast.AST) -> float:
|
||
if isinstance(node, ast.Expression):
|
||
return ev(node.body)
|
||
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
|
||
return node.value
|
||
if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
|
||
return _OPS[type(node.op)](ev(node.left), ev(node.right))
|
||
if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
|
||
return _OPS[type(node.op)](ev(node.operand))
|
||
raise ValueError("只允许数字和 + - * /")
|
||
|
||
if len(expr) > 100:
|
||
raise ValueError("表达式太长")
|
||
return ev(ast.parse(expr, mode="eval"))
|
||
|
||
|
||
class CalcArgs(BaseModel):
|
||
expression: str = Field(description="只含数字和 + - * / 括号的算式，例如 (3+4)*2")
|
||
|
||
|
||
class DaysArgs(BaseModel):
|
||
start: date = Field(description="开始日期，YYYY-MM-DD")
|
||
end: date = Field(description="结束日期，YYYY-MM-DD")
|
||
|
||
|
||
class NoArgs(BaseModel):
|
||
pass
|
||
|
||
|
||
def default_tools(today: Callable[[], date] = date.today) -> list[Tool]:
|
||
return [
|
||
local_tool("calculator", "计算四则运算。需要做数学计算时使用，不要自己心算。", CalcArgs,
|
||
lambda a: safe_eval(a["expression"])),
|
||
local_tool("today", "返回今天的日期（YYYY-MM-DD）。问题涉及“今天”“现在”时先调用它。", NoArgs,
|
||
lambda a: today().isoformat()),
|
||
local_tool("days_between", "计算两个日期之间相差的天数。", DaysArgs,
|
||
lambda a: (a["end"] - a["start"]).days),
|
||
]
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/tool_loop.py
|
||
import json
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from agent_mini.handwritten.llm import ChatLLM
|
||
from agent_mini.handwritten.tools import Tool
|
||
|
||
|
||
@dataclass
|
||
class StepLog:
|
||
tool: str
|
||
arguments: str
|
||
result: str
|
||
ok: bool
|
||
ms: int
|
||
|
||
|
||
@dataclass
|
||
class LoopResult:
|
||
answer: str
|
||
steps: list[StepLog] = field(default_factory=list)
|
||
stop_reason: str = "answered" # answered / max_steps
|
||
|
||
|
||
async def run_tool_loop(llm: ChatLLM, question: str, tools: list[Tool], max_steps: int = 6,
|
||
system: str = "你是一个会使用工具的助手。需要时调用工具，得到结果后用中文简洁回答。") -> LoopResult:
|
||
by_name = {t.name: t for t in tools}
|
||
messages: list[dict] = [{"role": "system", "content": system}, {"role": "user", "content": question}]
|
||
result = LoopResult(answer="")
|
||
for _ in range(max_steps):
|
||
msg = await llm.chat(messages, [t.schema() for t in tools])
|
||
messages.append(msg.to_dict())
|
||
if not msg.tool_calls:
|
||
result.answer = msg.content or ""
|
||
return result
|
||
for c in msg.tool_calls: # 模型可能一次要求调用多个工具
|
||
start = time.perf_counter()
|
||
ok = True
|
||
try:
|
||
tool = by_name.get(c.name)
|
||
if tool is None:
|
||
raise LookupError(f"没有名为 {c.name} 的工具，可用：{', '.join(by_name)}")
|
||
output = await tool.run(json.loads(c.arguments or "{}"))
|
||
except (json.JSONDecodeError, ValidationError) as e:
|
||
ok, output = False, f"参数错误：{e}。请修正参数后重试。"
|
||
except Exception as e: # 工具内部错误也返回给模型，而不是让整个循环崩溃
|
||
ok, output = False, f"工具执行失败：{e}"
|
||
result.steps.append(StepLog(c.name, c.arguments, output, ok, int((time.perf_counter() - start) * 1000)))
|
||
messages.append({"role": "tool", "tool_call_id": c.id, "content": output})
|
||
result.answer, result.stop_reason = "超过最大步数，已停止。", "max_steps"
|
||
return result
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/tests/test_tool_loop.py
|
||
from datetime import date
|
||
|
||
import pytest
|
||
|
||
from agent_mini.handwritten.llm import AssistantMessage, ScriptedLLM, call
|
||
from agent_mini.handwritten.tool_loop import run_tool_loop
|
||
from agent_mini.handwritten.tools import default_tools, safe_eval
|
||
|
||
TOOLS = default_tools(today=lambda: date(2026, 10, 23))
|
||
|
||
|
||
async def test_calls_tools_then_answers():
|
||
llm = ScriptedLLM([
|
||
call("today", {}, "c1"),
|
||
call("days_between", {"start": "2026-10-23", "end": "2027-01-01"}, "c2"),
|
||
call("calculator", {"expression": "70*3"}, "c3"),
|
||
AssistantMessage(content="还有 70 天，乘以 3 是 210。"),
|
||
])
|
||
r = await run_tool_loop(llm, "距离 2027 年元旦还有多少天，再乘以 3", TOOLS)
|
||
assert [s.tool for s in r.steps] == ["today", "days_between", "calculator"]
|
||
assert [s.result for s in r.steps] == ["2026-10-23", "70", "210"]
|
||
assert r.answer.endswith("210。")
|
||
last_messages = llm.calls[-1]
|
||
assert last_messages[-1] == {"role": "tool", "tool_call_id": "c3", "content": "210"}
|
||
|
||
|
||
async def test_bad_arguments_are_returned_to_model():
|
||
llm = ScriptedLLM([call("days_between", {"start": "明天"}), AssistantMessage(content="抱歉，请给出具体日期")])
|
||
r = await run_tool_loop(llm, "x", TOOLS)
|
||
assert r.steps[0].ok is False and "参数错误" in r.steps[0].result
|
||
assert "参数错误" in llm.calls[1][-1]["content"] # 模型下一轮能看到错误
|
||
|
||
|
||
async def test_unknown_tool_and_max_steps():
|
||
llm = ScriptedLLM([call("rm_rf", {})] * 10)
|
||
r = await run_tool_loop(llm, "x", TOOLS, max_steps=3)
|
||
assert r.stop_reason == "max_steps" and len(r.steps) == 3
|
||
assert "没有名为 rm_rf 的工具" in r.steps[0].result
|
||
|
||
|
||
@pytest.mark.parametrize("expr", ["__import__('os').system('ls')", "2**999999", "a+1"])
|
||
def test_safe_eval_rejects_code(expr):
|
||
with pytest.raises(ValueError):
|
||
safe_eval(expr)
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `handwritten/tools.py`、`handwritten/tool_loop.py`、`tests/test_tool_loop.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_tool_loop.py -v
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 测试通过；有密钥的话，用真实模型问“距离 2027 年元旦还有多少天，再乘以 3”，日志里能看到先调 `today`、再调 `days_between`、再调 `calculator`。
|
||
- 面试常问：工具调用的完整流程？模型参数给错了怎么办？怎么防止死循环？为什么不能用 `eval`？
|
||
|
||
---
|
||
|
||
## 第 3 天：手写多 Agent 循环
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 多 Agent 的本质：几个“角色 = 系统提示词 + 能看到的上下文 + 可用工具”，加上一个“下一步谁来做、什么时候结束”的调度逻辑。
|
||
- **协调者**输出结构化决定（JSON，用 pydantic 校验，格式错误时把错误告诉它重试一次）。
|
||
- **评审者**最后一行必须是 `VERDICT: APPROVE` 或 `VERDICT: REVISE`，用代码解析。
|
||
- **硬规则优先于模型**：评审通过 → 结束；达到最大轮数 → 结束。不能只靠模型自觉。
|
||
- 每一步把状态写进 JSON 文件（原子写入），进程中途退出后可以从文件继续——先体会“持久化 + 续跑”要解决什么问题，下两周再用检查点和数据库正式解决。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/multi_agent.py
|
||
import json
|
||
import os
|
||
import re
|
||
import tempfile
|
||
from pathlib import Path
|
||
from typing import Literal
|
||
|
||
from pydantic import BaseModel, Field, ValidationError
|
||
|
||
from agent_mini.handwritten.llm import ChatLLM
|
||
|
||
ROLE_PROMPTS = {
|
||
"coordinator": """你是协调者，负责决定下一步。只输出 JSON：
|
||
{"action": "dispatch" | "ask_human" | "finish", "target": "proposer" | "critic" | null,
|
||
"instruction": "给对方的具体指令", "reason": "理由"}""",
|
||
"proposer": "你是提议者。根据问题和评审意见给出具体、可执行的方案。",
|
||
"critic": "你是评审者。指出方案的问题。最后一行必须是 VERDICT: APPROVE 或 VERDICT: REVISE。",
|
||
"summarizer": "你是总结者。根据讨论给出最终结论，列出关键理由。",
|
||
}
|
||
|
||
|
||
class Decision(BaseModel):
|
||
action: Literal["dispatch", "ask_human", "finish"]
|
||
target: Literal["proposer", "critic"] | None = None
|
||
instruction: str = ""
|
||
reason: str = Field(min_length=1)
|
||
|
||
|
||
class Turn(BaseModel):
|
||
role: str
|
||
instruction: str
|
||
content: str
|
||
|
||
|
||
class LoopState(BaseModel):
|
||
question: str
|
||
max_rounds: int = 3
|
||
round: int = 0
|
||
turns: list[Turn] = []
|
||
last_verdict: Literal["APPROVE", "REVISE"] | None = None
|
||
status: Literal["running", "awaiting_input", "done"] = "running"
|
||
human_question: str | None = None
|
||
final_answer: str | None = None
|
||
end_reason: str | None = None
|
||
|
||
|
||
def parse_verdict(text: str) -> Literal["APPROVE", "REVISE"]:
|
||
m = re.search(r"VERDICT:\s*(APPROVE|REVISE)\s*$", text.strip())
|
||
return m.group(1) if m else "REVISE" # 格式不对按“不通过”处理，偏保守 # type: ignore[return-value]
|
||
|
||
|
||
def save_atomic(path: Path, state: LoopState) -> None:
|
||
fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
f.write(state.model_dump_json(indent=2))
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def context_for(state: LoopState, last_n: int = 4) -> str:
|
||
"""上下文工程：只给问题 + 最近几条发言，而不是全部历史。"""
|
||
recent = "\n".join(f"[{t.role}] {t.content[:400]}" for t in state.turns[-last_n:])
|
||
return f"问题：{state.question}\n最近的讨论：\n{recent or '（无）'}"
|
||
|
||
|
||
async def decide(llm: ChatLLM, state: LoopState) -> Decision:
|
||
messages = [{"role": "system", "content": ROLE_PROMPTS["coordinator"]},
|
||
{"role": "user", "content": context_for(state)}]
|
||
for attempt in range(2):
|
||
raw = (await llm.chat(messages)).content or ""
|
||
try:
|
||
m = re.search(r"\{.*\}", raw, re.S)
|
||
return Decision.model_validate_json(m.group(0) if m else raw)
|
||
except ValidationError as e:
|
||
if attempt == 1:
|
||
raise
|
||
messages += [{"role": "assistant", "content": raw},
|
||
{"role": "user", "content": f"格式错误：{e.errors()[0]['msg']}。只输出合法 JSON。"}]
|
||
raise AssertionError("unreachable")
|
||
|
||
|
||
async def run_multi_agent(llm: ChatLLM, state: LoopState, path: Path, human_answer: str | None = None) -> LoopState:
|
||
if state.status == "awaiting_input":
|
||
if human_answer is None:
|
||
return state
|
||
state.turns.append(Turn(role="human", instruction=state.human_question or "", content=human_answer))
|
||
state.status, state.human_question = "running", None
|
||
|
||
while state.status == "running":
|
||
# 1. 硬规则
|
||
if state.last_verdict == "APPROVE":
|
||
state.end_reason = "consensus"
|
||
elif state.round >= state.max_rounds:
|
||
state.end_reason = "max_rounds"
|
||
if state.end_reason:
|
||
break
|
||
# 2. 协调者决定
|
||
d = await decide(llm, state)
|
||
if d.action == "finish":
|
||
state.end_reason = "coordinator_finish"
|
||
break
|
||
if d.action == "ask_human":
|
||
state.status, state.human_question = "awaiting_input", d.instruction
|
||
save_atomic(path, state)
|
||
return state
|
||
# 3. 执行专家
|
||
target = d.target or "proposer"
|
||
reply = await llm.chat([{"role": "system", "content": ROLE_PROMPTS[target]},
|
||
{"role": "user", "content": f"{context_for(state)}\n\n指令：{d.instruction}"}])
|
||
content = reply.content or ""
|
||
state.turns.append(Turn(role=target, instruction=d.instruction, content=content))
|
||
if target == "proposer":
|
||
state.round += 1
|
||
else:
|
||
state.last_verdict = parse_verdict(content)
|
||
save_atomic(path, state) # 每一步都落盘
|
||
|
||
summary = await llm.chat([{"role": "system", "content": ROLE_PROMPTS["summarizer"]},
|
||
{"role": "user", "content": context_for(state, last_n=8)}])
|
||
state.final_answer, state.status = summary.content, "done"
|
||
save_atomic(path, state)
|
||
return state
|
||
|
||
|
||
def load_state(path: Path) -> LoopState:
|
||
return LoopState.model_validate_json(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(json.dumps(Decision.model_json_schema(), ensure_ascii=False)[:200])
|
||
```
|
||
|
||
用一个“按角色回答”的假模型测试：
|
||
|
||
```python
|
||
# 文件：agent-mini/tests/test_multi_agent.py
|
||
import json
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from agent_mini.handwritten.llm import AssistantMessage
|
||
from agent_mini.handwritten.multi_agent import LoopState, load_state, parse_verdict, run_multi_agent
|
||
|
||
|
||
class RoleLLM:
|
||
"""根据系统提示词判断当前角色，从对应队列里取回答。"""
|
||
|
||
def __init__(self, **queues: list[str]):
|
||
self.queues = {k: list(v) for k, v in queues.items()}
|
||
self.count = {k: 0 for k in queues}
|
||
|
||
async def chat(self, messages, tools=None):
|
||
system = messages[0]["content"]
|
||
role = next(r for r in ("coordinator", "proposer", "critic", "summarizer")
|
||
if {"coordinator": "协调者", "proposer": "提议者", "critic": "评审者", "summarizer": "总结者"}[r] in system)
|
||
self.count[role] = self.count.get(role, 0) + 1
|
||
return AssistantMessage(content=self.queues[role].pop(0))
|
||
|
||
|
||
def d(action, target=None, instruction="请处理", reason="需要"):
|
||
return json.dumps({"action": action, "target": target, "instruction": instruction, "reason": reason})
|
||
|
||
|
||
async def test_consensus_ends_early(tmp_path):
|
||
llm = RoleLLM(
|
||
coordinator=[d("dispatch", "proposer"), d("dispatch", "critic"), d("dispatch", "proposer"), d("dispatch", "critic")],
|
||
proposer=["方案一", "方案二"],
|
||
critic=["不够具体\nVERDICT: REVISE", "可以\nVERDICT: APPROVE"],
|
||
summarizer=["结论：采用方案二"],
|
||
)
|
||
s = await run_multi_agent(llm, LoopState(question="选技术栈", max_rounds=5), tmp_path / "s.json")
|
||
assert (s.end_reason, s.round, s.final_answer) == ("consensus", 2, "结论：采用方案二")
|
||
assert load_state(tmp_path / "s.json").status == "done"
|
||
|
||
|
||
async def test_max_rounds(tmp_path):
|
||
llm = RoleLLM(coordinator=[d("dispatch", "proposer"), d("dispatch", "critic")] * 2,
|
||
proposer=["a", "b"], critic=["VERDICT: REVISE"] * 2, summarizer=["到达上限"])
|
||
s = await run_multi_agent(llm, LoopState(question="q", max_rounds=2), tmp_path / "s.json")
|
||
assert s.end_reason == "max_rounds" and s.round == 2
|
||
|
||
|
||
async def test_bad_json_retried_once(tmp_path):
|
||
llm = RoleLLM(coordinator=["好的我来派活", d("finish")], summarizer=["结束"])
|
||
s = await run_multi_agent(llm, LoopState(question="q"), tmp_path / "s.json")
|
||
assert s.end_reason == "coordinator_finish" and llm.count["coordinator"] == 2
|
||
|
||
llm2 = RoleLLM(coordinator=["坏", "还是坏"])
|
||
with pytest.raises(ValidationError):
|
||
await run_multi_agent(llm2, LoopState(question="q"), tmp_path / "s2.json")
|
||
|
||
|
||
async def test_ask_human_then_resume_from_file(tmp_path):
|
||
path = tmp_path / "s.json"
|
||
llm = RoleLLM(coordinator=[d("ask_human", instruction="预算是多少？"), d("finish")], summarizer=["按 5 万预算推荐"])
|
||
s = await run_multi_agent(llm, LoopState(question="做网站"), path)
|
||
assert s.status == "awaiting_input" and s.human_question == "预算是多少？"
|
||
restored = load_state(path) # 模拟进程重启：只从文件恢复
|
||
s = await run_multi_agent(llm, restored, path, human_answer="5 万")
|
||
assert s.status == "done" and s.turns[-1].role == "human"
|
||
|
||
|
||
def test_parse_verdict():
|
||
assert parse_verdict("好\nVERDICT: APPROVE") == "APPROVE"
|
||
assert parse_verdict("没写结论") == "REVISE"
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：多 Agent 循环一定要有最大轮数，否则两个角色可能一直互相“再改改”。
|
||
- 易错点：状态要在每一步之后保存（本周存 JSON 文件），保存要用“写临时文件再改名”，避免崩溃时写坏文件。
|
||
- 面试常问：多 Agent 和单 Agent + 多工具的区别和代价（角色分工更清楚、可以互相审查，但调用次数和费用成倍增加，调试更难）。
|
||
|
||
### 要建的文件
|
||
|
||
- `handwritten/multi_agent.py`（150~250 行）、`tests/test_multi_agent.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_multi_agent.py -v
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 测试通过；能不看代码讲清循环的每一步。
|
||
- 写下“用 JSON 文件持久化”的问题（为下两周铺垫）：多个进程同时写会互相覆盖；查询“所有等待输入的任务”要扫所有文件；正在调用模型时崩溃，这一步会重做。
|
||
|
||
---
|
||
|
||
## 第 4 天：MCP 概念 + 写一个 MCP 服务
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- **MCP（Model Context Protocol）**：让 AI 应用用统一方式接入外部工具和数据的协议。角色：**宿主**（Cursor、Claude 桌面版、你自己的 Agent）、**客户端**（宿主里负责连接的部分）、**服务端**（你写的，提供能力）。
|
||
- 服务端能提供三类东西：
|
||
- **tools**：模型可以调用的函数（有副作用也行），比如“查知识库”“建待办”；
|
||
- **resources**：只读数据，用 URI 标识，比如 `notes://cards`；
|
||
- **prompts**：可复用的提示词模板，比如“评审某个方案”。
|
||
- **传输方式**：本地子进程用 **stdio**（宿主启动你的程序，通过标准输入输出通信）；远程服务用 **Streamable HTTP**（老的 HTTP+SSE 传输已被它取代）。
|
||
- **和 Function Calling 的关系**：Function Calling 是“模型怎么表达要调用工具”；MCP 是“工具怎么被统一发现和调用”。写一次 MCP 服务，所有支持 MCP 的宿主都能用。
|
||
- 返回值写成 pydantic 模型，SDK 会自动生成**输出 Schema**，客户端拿到结构化结果；想把错误信息告诉调用方要抛 `ToolError`，其他异常在 2.x 里只会显示“执行出错”（防止泄露内部细节）。
|
||
- SDK 版本：MCP Python SDK 2.x 里服务端类是 `MCPServer`（`from mcp.server.mcpserver import MCPServer`）；1.x 里叫 `FastMCP`（`from mcp.server.fastmcp import FastMCP`）。网上很多教程还是 1.x 写法，**以官方文档为准**：https://py.sdk.modelcontextprotocol.io/
|
||
- 案例：知识库 MCP 服务——查询第 5~6 周的 `rag-qa`，提供一个幂等的“建待办”工具、一个资源和一个提示词。
|
||
|
||
```python
|
||
# 文件：agent-mini/mcp_servers/__init__.py
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/mcp_servers/knowledge_server.py
|
||
"""知识库 MCP 服务。运行：uv run python mcp_servers/knowledge_server.py（stdio）"""
|
||
import os
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
from mcp.server.mcpserver import MCPServer
|
||
from mcp.server.mcpserver.exceptions import ToolError
|
||
from pydantic import BaseModel
|
||
|
||
RAG_QA_URL = os.getenv("RAG_QA_URL", "http://localhost:8000")
|
||
NOTES_DIR = Path(os.getenv("NOTES_DIR", "notes/cards"))
|
||
RAG_TRANSPORT: httpx.AsyncBaseTransport | None = None # 测试时替换成 MockTransport
|
||
|
||
mcp = MCPServer("knowledge", instructions="查询个人知识库（rag-qa）和学习笔记。")
|
||
_todos: dict[str, dict] = {} # 演示用；真实实现调用项目一的接口并带上幂等键
|
||
|
||
|
||
# 返回值用 pydantic 模型：SDK 会自动生成输出 Schema，客户端拿到结构化结果（structured_content）
|
||
class Source(BaseModel):
|
||
n: int
|
||
file: str
|
||
heading: str
|
||
|
||
|
||
class SearchResult(BaseModel):
|
||
answer: str
|
||
sources: list[Source]
|
||
|
||
|
||
class DocItem(BaseModel):
|
||
filename: str
|
||
chunks: int
|
||
|
||
|
||
class TodoResult(BaseModel):
|
||
id: int
|
||
title: str
|
||
duplicate: bool
|
||
|
||
|
||
def _client() -> httpx.AsyncClient:
|
||
return httpx.AsyncClient(base_url=RAG_QA_URL, timeout=30, transport=RAG_TRANSPORT)
|
||
|
||
|
||
@mcp.tool()
|
||
async def search_knowledge(query: str, top_k: int = 4) -> SearchResult:
|
||
"""在个人知识库中检索并回答问题，返回回答和出处。
|
||
当问题涉及“我的笔记 / 知识库 / 学过的内容”时使用。query 用完整的中文问题。"""
|
||
async with _client() as c:
|
||
try:
|
||
r = await c.post("/ask", json={"question": query})
|
||
except httpx.HTTPError as e:
|
||
# ToolError 的信息会原样返回给客户端；其他异常只会显示“执行出错”，避免泄露内部细节
|
||
raise ToolError(f"知识库服务不可用（{type(e).__name__}），请稍后再试") from e
|
||
if r.status_code != 200:
|
||
raise ToolError(f"知识库返回 {r.status_code}")
|
||
data = r.json()
|
||
return SearchResult(answer=data["text"], sources=[
|
||
Source(n=c["n"], file=c["filename"], heading=c["heading_path"]) for c in data["citations"][:top_k]])
|
||
|
||
|
||
@mcp.tool()
|
||
async def list_documents() -> list[DocItem]:
|
||
"""列出知识库里的文档（文件名和片段数）。"""
|
||
async with _client() as c:
|
||
r = await c.get("/documents")
|
||
if r.status_code != 200:
|
||
raise ToolError(f"知识库返回 {r.status_code}")
|
||
return [DocItem(filename=d["filename"], chunks=d["chunks"]) for d in r.json()]
|
||
|
||
|
||
@mcp.tool()
|
||
def create_todo(title: str, request_id: str) -> TodoResult:
|
||
"""创建一条待办。request_id 由调用方生成；同一个 request_id 重复调用只会创建一次（幂等）。"""
|
||
if request_id in _todos:
|
||
return TodoResult(**_todos[request_id], duplicate=True)
|
||
_todos[request_id] = {"id": len(_todos) + 1, "title": title}
|
||
return TodoResult(**_todos[request_id], duplicate=False)
|
||
|
||
|
||
@mcp.resource("notes://cards")
|
||
def card_titles() -> str:
|
||
"""案例卡标题列表（只读脱敏后的 notes/cards）。"""
|
||
if not NOTES_DIR.is_dir():
|
||
return "（没有找到案例卡目录）"
|
||
return "\n".join(p.read_text(encoding="utf-8").splitlines()[0] for p in sorted(NOTES_DIR.glob("*.md")))
|
||
|
||
|
||
@mcp.prompt()
|
||
def review_plan(plan: str) -> str:
|
||
"""让模型按固定维度评审一个方案。"""
|
||
return f"请从成本、风险、可维护性三个方面评审下面的方案，最后给出 VERDICT: APPROVE 或 VERDICT: REVISE。\n\n{plan}"
|
||
|
||
|
||
if __name__ == "__main__": # 不自动运行
|
||
mcp.run() # 默认 stdio；远程部署用 mcp.run(transport="streamable-http")
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/tests/test_mcp_server.py
|
||
import httpx
|
||
from mcp import Client
|
||
|
||
import mcp_servers.knowledge_server as ks
|
||
|
||
|
||
def fake_rag(request: httpx.Request) -> httpx.Response:
|
||
if request.url.path == "/ask":
|
||
return httpx.Response(200, json={"text": "用 channel 通信 [1]", "citations": [
|
||
{"n": 1, "filename": "go.md", "heading_path": "Go > 并发"}]})
|
||
return httpx.Response(200, json=[{"filename": "go.md", "chunks": 3}])
|
||
|
||
|
||
async def test_tools_resources_prompts(monkeypatch, tmp_path):
|
||
monkeypatch.setattr(ks, "RAG_TRANSPORT", httpx.MockTransport(fake_rag))
|
||
(tmp_path / "A.md").write_text("标题：编排停在浏览器里\n...", encoding="utf-8")
|
||
monkeypatch.setattr(ks, "NOTES_DIR", tmp_path)
|
||
|
||
async with Client(ks.mcp) as client: # 内存连接：不起子进程，测试最方便
|
||
names = {t.name for t in (await client.list_tools()).tools}
|
||
assert names == {"search_knowledge", "list_documents", "create_todo"}
|
||
|
||
r = await client.call_tool("search_knowledge", {"query": "goroutine 怎么通信"})
|
||
assert not r.is_error and r.structured_content["sources"][0]["file"] == "go.md"
|
||
|
||
first = await client.call_tool("create_todo", {"title": "复习", "request_id": "req-1"})
|
||
again = await client.call_tool("create_todo", {"title": "复习", "request_id": "req-1"})
|
||
assert first.structured_content["duplicate"] is False and again.structured_content["duplicate"] is True
|
||
|
||
res = await client.read_resource("notes://cards")
|
||
assert "编排停在浏览器里" in res.contents[0].text
|
||
|
||
p = await client.get_prompt("review_plan", {"plan": "用 Redis 缓存"})
|
||
assert "VERDICT" in p.messages[0].content.text
|
||
|
||
|
||
async def test_rag_down_returns_error(monkeypatch):
|
||
def down(request):
|
||
raise httpx.ConnectError("refused", request=request)
|
||
|
||
monkeypatch.setattr(ks, "RAG_TRANSPORT", httpx.MockTransport(down))
|
||
async with Client(ks.mcp) as client:
|
||
r = await client.call_tool("search_knowledge", {"query": "x"})
|
||
assert r.is_error and "不可用" in r.content[0].text
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `mcp_servers/__init__.py`、`mcp_servers/knowledge_server.py`、`tests/test_mcp_server.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_mcp_server.py -v
|
||
# 官方调试工具 MCP Inspector：浏览器里查看并手动调用工具
|
||
npx @modelcontextprotocol/inspector uv run python mcp_servers/knowledge_server.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 测试通过；在 Inspector 里能看到 3 个工具、1 个资源、1 个提示词并手动调用（先启动 `rag-qa`）。
|
||
- 面试常问：MCP 是什么？tools / resources / prompts 的区别？stdio 和 Streamable HTTP 怎么选？有副作用的工具怎么保证安全（幂等键、权限、人工确认）？
|
||
|
||
---
|
||
|
||
## 第 5 天：在 Cursor 和自己的 Agent 里用 MCP
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 同一个 MCP 服务，**Cursor 能用，你自己的 Agent 也能用**——这就是协议的价值。
|
||
- 在自己的 Agent 里：用 MCP 客户端连接服务 → `list_tools` 拿到工具名、说明、参数 Schema → 转成第 2 天的 `Tool` 对象 → 放进工具调用循环。工具是**运行时动态发现**的，不用写死。
|
||
- 连接方式：测试里用内存连接 `Client(server)`；真实使用时用 `Client(StdioServerParameters(command=..., args=[...]))` 启动子进程，或连远程 HTTP 地址。
|
||
- LangChain 也提供了 MCP 适配器（独立包 `langchain-mcp-adapters` 的 `MultiServerMCPClient`；较新的 LangChain 把它并进了 `langchain` 包，处于测试阶段）。**以官方文档为准**：https://docs.langchain.com/oss/python/langchain/mcp 。下周我们用自己写的 20 行桥接，原理更清楚。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/mcp_bridge.py
|
||
from mcp import Client
|
||
|
||
from agent_mini.handwritten.tools import Tool
|
||
|
||
|
||
class ToolExecError(RuntimeError):
|
||
pass
|
||
|
||
|
||
async def load_mcp_tools(client: Client, allow: set[str] | None = None) -> list[Tool]:
|
||
"""把 MCP 服务上的工具转成手写循环用的 Tool。allow 是白名单：不要把所有工具都暴露给模型。"""
|
||
tools = []
|
||
for t in (await client.list_tools()).tools:
|
||
if allow is not None and t.name not in allow:
|
||
continue
|
||
|
||
async def run(args: dict, _name: str = t.name) -> str:
|
||
r = await client.call_tool(_name, args)
|
||
text = "\n".join(c.text for c in r.content if getattr(c, "type", "") == "text")
|
||
if r.is_error:
|
||
raise ToolExecError(text)
|
||
return text
|
||
|
||
tools.append(Tool(name=t.name, description=t.description or "", parameters=t.input_schema, func=run))
|
||
return tools
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/tests/test_mcp_bridge.py
|
||
import httpx
|
||
from mcp import Client
|
||
|
||
import mcp_servers.knowledge_server as ks
|
||
from agent_mini.handwritten.llm import AssistantMessage, ScriptedLLM, call
|
||
from agent_mini.handwritten.mcp_bridge import load_mcp_tools
|
||
from agent_mini.handwritten.tool_loop import run_tool_loop
|
||
from agent_mini.handwritten.tools import default_tools
|
||
|
||
|
||
async def test_agent_uses_mcp_tool(monkeypatch):
|
||
monkeypatch.setattr(ks, "RAG_TRANSPORT", httpx.MockTransport(
|
||
lambda r: httpx.Response(200, json={"text": "Pod 是最小调度单位 [1]",
|
||
"citations": [{"n": 1, "filename": "k8s.md", "heading_path": "K8s"}]})))
|
||
async with Client(ks.mcp) as client:
|
||
mcp_tools = await load_mcp_tools(client, allow={"search_knowledge"})
|
||
assert [t.name for t in mcp_tools] == ["search_knowledge"]
|
||
llm = ScriptedLLM([call("search_knowledge", {"query": "Pod 是什么"}),
|
||
AssistantMessage(content="根据你的笔记，Pod 是最小调度单位。")])
|
||
r = await run_tool_loop(llm, "我的笔记里 Pod 是什么", default_tools() + mcp_tools)
|
||
assert r.steps[0].ok and "k8s.md" in r.steps[0].result
|
||
assert "根据你的笔记" in r.answer
|
||
```
|
||
|
||
Cursor 里配置 MCP（项目根目录 `.cursor/mcp.json`，**格式以 Cursor 官方文档为准**；路径换成你自己的，不要提交含个人信息的路径）：
|
||
|
||
```json
|
||
{
|
||
"mcpServers": {
|
||
"knowledge": {
|
||
"command": "uv",
|
||
"args": ["run", "--directory", "/path/to/agent-mini", "python", "mcp_servers/knowledge_server.py"],
|
||
"env": { "RAG_QA_URL": "http://localhost:8000" }
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：stdio 方式的 MCP 服务**不能往标准输出打印日志**，标准输出是协议通道，乱打印会让客户端解析失败；日志写到标准错误。
|
||
- 易错点：Cursor 的 `mcp.json` 里命令和路径要写对（建议用绝对路径或 `uv --directory`），改完配置要重新加载。
|
||
- 易错点：MCP Python SDK 2.x 和 1.x 的导入路径、类名不同（本计划按 2.x 写），网上的例子多数是 1.x，以官方文档为准：https://py.sdk.modelcontextprotocol.io/ 。
|
||
- 面试常问：MCP 解决什么问题（工具接入的统一协议：一个服务可以被各种客户端复用）；tools、resources、prompts 的区别；stdio 和 Streamable HTTP 两种传输方式各适合什么场景（本机进程 / 远程服务）。
|
||
|
||
### 要建的文件
|
||
|
||
- `handwritten/mcp_bridge.py`、`tests/test_mcp_bridge.py`、`docs/mcp-usage.md`（Cursor 配置示例 + 截图）
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest -v
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 在 Cursor 对话里让它“查一下我的知识库里关于 X 的内容”，能看到它调用了 `search_knowledge`。
|
||
- 你自己的工具调用循环通过 MCP 用上了同一个工具。
|
||
|
||
---
|
||
|
||
## 第 6 天：真实模型试跑 + 对照文档（上半）
|
||
|
||
### 学习内容
|
||
|
||
- 说明：用真实模型跑手写版，观察模型的“不听话”：协调者 JSON 格式错、评审者忘写 VERDICT、乱调工具、重复调用同一个工具。每个现象都对应一条防护（校验重试、保守默认值、白名单、最大步数）。
|
||
- 案例：一个命令行入口（有密钥时用真实模型，没有时提示）。
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/handwritten/demo.py
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
|
||
from agent_mini.handwritten.llm import OpenAICompatLLM
|
||
from agent_mini.handwritten.multi_agent import LoopState, run_multi_agent
|
||
|
||
|
||
async def main(question: str) -> None: # 不自动运行：需要真实模型密钥
|
||
if not os.getenv("LLM_API_KEY"):
|
||
sys.exit("请先在环境变量里设置 LLM_BASE_URL、LLM_API_KEY、CHAT_MODEL")
|
||
async with httpx.AsyncClient(base_url=os.environ["LLM_BASE_URL"], timeout=60,
|
||
headers={"Authorization": f"Bearer {os.environ['LLM_API_KEY']}"}) as c:
|
||
llm = OpenAICompatLLM(c, os.environ["CHAT_MODEL"])
|
||
path = Path(tempfile.gettempdir()) / "agent_mini_state.json"
|
||
state = await run_multi_agent(llm, LoopState(question=question, max_rounds=3), path)
|
||
for t in state.turns:
|
||
print(f"\n## {t.role}（指令：{t.instruction}）\n{t.content}")
|
||
print(f"\n== 结束原因：{state.end_reason}\n{state.final_answer}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main(" ".join(sys.argv[1:]) or "5 人创业团队做内部工具，该选什么技术栈？"))
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `handwritten/demo.py`；`docs/samples/handwritten-*.md`（真实运行记录，检查无敏感信息）
|
||
- `docs/handwritten-vs-langgraph.md`：先写“手写版”这一半：循环、状态、持久化、工具调用、结束条件分别怎么做的，哪里最容易出错
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run python -m agent_mini.handwritten.demo "给读书 App 设计会员规则"
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 跑 3 个题目，记录观察到的“不听话”现象和对应防护。
|
||
|
||
---
|
||
|
||
## 第 7 天：M1 验收
|
||
|
||
### 学习内容
|
||
|
||
- 说明：整理代码和测试；把本周的面试题用自己的话写进 `docs/handwritten-vs-langgraph.md`。
|
||
|
||
### 要建的文件
|
||
|
||
- README 初版：项目介绍、目录、怎么跑测试、怎么启动 MCP 服务
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run ruff check . && uv run pyright src && uv run pytest -v
|
||
```
|
||
|
||
### 完成标准（M1 验收）
|
||
|
||
- [ ] 手写工具调用循环和多 Agent 循环都能跑，测试通过
|
||
- [ ] 能讲清 Function Calling 的完整过程，以及参数不合法、工具出错、死循环怎么处理
|
||
- [ ] 写了 MCP 服务（tools / resources / prompts），Inspector 和 Cursor 都能调用
|
||
- [ ] 自己的 Agent 能动态加载 MCP 工具
|
||
- [ ] 能讲清 MCP 和 Function Calling 的关系
|
||
- [ ] 在总纲进度表勾掉第 10 周
|
||
|
||
<!-- 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) |
|
||
| [AI 编程工具与验证](../topics/ai/ai-coding/index.md) | 课程：[AI 编程工具工作流](../topics/ai/ai-coding/01-workflow-rules-skills.md) |
|
||
| [Function Calling 工具调用](../topics/ai/function-calling/index.md) | 课程：[工具调用原理](../topics/ai/function-calling/01-tool-loop.md)；场景：[让 Agent 安全地调用内部业务接口](../topics/ai/function-calling/scenarios/safe-business-tools.md)；[面试题](../topics/ai/function-calling/interview.md) |
|
||
| [模型侧基础（了解级）](../topics/ai/llm-serving/index.md) | 课程：[模型侧基础](../topics/ai/llm-serving/01-routing-vllm-lora.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) |
|
||
|
||
**跨技术栈场景**：[报表生成 Agent](../scenarios/report-agent.md)
|
||
|
||
<!-- week-topics:end -->
|