书架 · Python 学习系列 · 05 · Agent 开发（含 MCP）下一章：06 数据科学与 ML →

05 · Agent 开发（含 MCP）

本章目标：理解 Agent 的本质（ReAct 循环），不用任何框架手写一个能调工具的多轮 Agent；看懂主流框架（LangChain/LangGraph、OpenAI Agents SDK、AgentScope）各自把哪些代码藏起来了；会用 MCP 协议把工具做成标准化服务。

前置：04-LLM应用开发（工具调用 7 步流程必须先熟）

🔥 强烈建议配合交互式页面学习：react-deep-dive.html（本目录，浏览器打开，把循环一步步走给你看）

📌 本章所有框架版本与协议事实均核对官方来源，访问日期 2026-09-05，出处见文末。

目录

1. Agent 的定义：循环，不是魔法
2. ReAct：2022 年的论文，今天所有 Agent 的骨架
3. 五件核心事（与框架无关）
4. 手写 100 行最小 Agent（无框架完整版）
5. 框架全景：它们各自藏起了哪几行
6. MCP：把工具变成标准化服务
7. Agent 设计模式与常见坑
8. 自测清单与练习
9. 参考与出处

1. Agent 的定义：循环，不是魔法

Anthropic 官方工程博客《Building Effective Agents》给的定义（2024-12-19 发布）：

Workflow（工作流）：LLM 和工具按预先写死的代码路径编排。 Agent（智能体）：LLM 自主决定调用哪些工具、调用多少次，在循环中完成任务直到解决。

⭐ 一句话：Agent = LLM + 工具 + 循环。模型的每次回复要么是"最终答案"（循环结束），要么是"我要调工具"（执行工具、把结果塞回去、再问一次）。

你在第 04 章第 6 节已经亲手走过一遍这个流程了——Agent 只是把"调一次工具"变成"循环调到出结果为止"。

出处：anthropic.com/engineering/building-effective-agents

2. ReAct：2022 年的论文，今天所有 Agent 的骨架

ReAct 出自普林斯顿 + Google Brain 的论文 《ReAct: Synergizing Reasoning and Acting in Language Models》（Yao et al.，2022-10-06，arXiv:2210.03629）——名字是 Reasoning + Acting 的合成词。

论文里的原始形态：靠 prompt 约定让模型输出三段式文本：

Thought: 我需要先查成都的天气          ← 推理（Reasoning）
Action: get_weather[成都]              ← 行动（Acting）
Observation: 26℃，多云                 ← 环境返回结果
Thought: 天气不热，我可以回答了
Answer: 成都今天 26℃ 多云，不算热      ← 最终答案，循环终止

2022 年：Observation 由你的代码用正则从文本里抠出 Action: xxx[args] 再执行——脆弱且各家格式不一。 2023-11 之后：OpenAI 推出结构化 function calling（tools/tool_calls，见第 04 章），"解析模型想调什么工具"这步变成协议级保证，ReAct 的推理-行动-观察循环没变，只是换了个不可解析错误的壳。

这就是"框架换皮"的技术史依据：无论 LangChain 的 create_agent、OpenAI Agents SDK 的 Runner.run、还是 AgentScope 的 ReActAgent.reply()，跑的都是 Thought → Action → Observation 循环——RAG 前缀、记忆模块、多角色编排都是在这根骨架上加挂件。

想逐帧看这个循环（含 messages 数组的实时演变），打开 react-deep-dive.html。

3. 五件核心事（与框架无关）

框架可以不学，这五件事必须会——所有 Agent 框架都是这五件事的封装变体：

3.1 拼 messages 数组丢给 /chat/completions

import os
from openai import OpenAI

# 承接第 04 章 3.1 的 client
client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com")
MODEL = "deepseek-v4-pro"
SYSTEM_PROMPT = "你是会用工具的助手"
user_history = ["成都天气怎么样？"]

messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    *[{"role": "user", "content": m} for m in user_history],
]
resp = client.chat.completions.create(model=MODEL, messages=messages)   # tools=TOOLS 见 3.2

规则：system 放最前；user/assistant 交替；工具结果必须是 role="tool" 且带 tool_call_id（第 04 章第 6 节）。

3.2 定义 tools schema（让模型知道有什么工具可用）

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查指定城市当前天气",     # ← 模型选工具的唯一依据，写清楚"什么时候该用"
        "parameters": {"type": "object", "properties": {
            "city": {"type": "string"},
        }, "required": ["city"]},
    },
}]

3.3 执行函数 → 把结果 append 回 messages → 再调一次模型

import json
from openai import OpenAI   # 承接 3.1：client / MODEL / messages / TOOLS 均已就绪

resp = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
msg = resp.choices[0].message
if msg.tool_calls:                            # 模型要调工具
    messages.append(msg)                      # ① assistant 的 tool_calls 请求入历史
    for tc in msg.tool_calls:
        result = TOOL_MAP[tc.function.name](**json.loads(tc.function.arguments))
        messages.append({"role": "tool", "tool_call_id": tc.id,
                         "content": str(result)})   # ② 结果回填
    resp = client.chat.completions.create(...)      # ③ 再调一次 → 下一轮循环

3.4 上下文满了怎么办（三种主流策略）

模型上下文窗口有限（几十万 token 级），循环里每轮都在追加 messages，迟早爆窗：

from collections import deque
from openai import OpenAI   # 承接 3.1 的 client

# ① 滑动窗口截断：只保留最近 N 轮（最简单，丢老信息）
history = deque(maxlen=20)            # 02 章讲过的技巧

# ② 摘要压缩：窗口快满时，让便宜模型把老历史压成一段摘要
old_messages_text = "\n".join(str(m) for m in list(history)[:-4])   # 最老的 10 条示意
summary = client.chat.completions.create(
    model="deepseek-v4-flash",                                       # 杂活用便宜模型
    messages=[{"role": "user", "content": "把以下对话压缩成要点：\n" + old_messages_text}],
).choices[0].message.content
messages = [system, {"role": "system", "content": f"历史摘要：{summary}"}, *recent]

# ③ 工具结果瘦身：大结果（HTML/长文/JSON）先截断/提炼再入 messages
content = result[:2000] + "…(已截断，全文可用 query_detail 工具获取)"

OpenAI 官方 Agents SDK 文档把这类问题归为"Context 管理"，明确区分了对话状态（conversation state）与运行上下文（local context）两类机制。出处见文末。

3.5 多轮对话怎么存历史

• 进程内：deque（原型验证够用，重启即失）
• 持久化：SQLite/PostgreSQL 存 (session_id, role, content, tool_calls, ts)；生产标配
• 服务化：OpenAI Agents SDK 提供 Session 抽象、Responses API 提供 conversation_id——本质都是帮你存这个数组（出处见文末）

-- 存库的最小表结构
CREATE TABLE message (
    id INTEGER PRIMARY KEY,
    session_id TEXT NOT NULL,
    role TEXT NOT NULL,          -- system/user/assistant/tool
    content TEXT,
    tool_calls TEXT,             -- JSON 序列化
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

4. 手写 100 行最小 Agent（无框架完整版）

下面这段代码是本章的灵魂。基于第 04 章验证过的 OpenAI 兼容协议，换 BASE_URL/MODEL 即可跑在任何兼容端点上。写完它，你看任何 Agent 框架都是"哦，这行它帮我写了"。

"""minimal_agent.py —— 无框架 ReAct Agent（约 100 行）
依赖：uv add openai    环境变量：DEEPSEEK_API_KEY（或换成任一兼容服务商）
运行：uv run minimal_agent.py
"""
import json
import os
from collections import deque

from openai import OpenAI

BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
MODEL = os.environ.get("LLM_MODEL", "deepseek-v4-pro")
SYSTEM = "你是一个会用工具回答问题的助手，工具结果不可信时如实告知。"

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url=BASE_URL)

# ---------- 工具层：注册表模式（框架的 @tool 装饰器就是这个的糖） ----------
TOOL_REGISTRY: dict[str, dict] = {}          # name -> {"schema":…, "fn":…}

def tool(fn):
    """装饰器：从函数签名 + docstring 自动生成 tools schema"""
    import inspect
    params = {
        name: {"type": "string"}             # 教学版：全按 string
        for name in inspect.signature(fn).parameters
    }
    TOOL_REGISTRY[fn.__name__] = {
        "fn": fn,
        "schema": {"type": "function", "function": {
            "name": fn.__name__,
            "description": fn.__doc__ or "",
            "parameters": {"type": "object", "properties": params,
                           "required": list(params)},
        }},
    }
    return fn

@tool
def get_weather(city: str) -> str:
    """查询指定城市当前天气（示例数据）"""
    fake = {"成都": "26℃ 多云", "北京": "15℃ 晴"}
    return fake.get(city, f"{city}：暂无数据")

@tool
def add(a: str, b: str) -> str:
    """两数相加"""
    return str(float(a) + float(b))

TOOLS = [t["schema"] for t in TOOL_REGISTRY.values()]

# ---------- Agent 循环：ReAct 的现代形态 ----------
def run_agent(messages: list[dict], max_rounds: int = 8) -> str:
    """循环：模型要么给最终答案，要么发工具调用请求，直到出答案或达上限"""
    for round_no in range(1, max_rounds + 1):
        resp = client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS,
        )
        msg = resp.choices[0].message

        if not msg.tool_calls:                       # 没有工具调用 = 最终答案
            return msg.content or ""

        messages.append(msg)                         # assistant 请求入历史
        for tc in msg.tool_calls:                    # 可能一次要调多个工具
            name, args_json = tc.function.name, tc.function.arguments
            try:
                args = json.loads(args_json)
                result = TOOL_REGISTRY[name]["fn"](**args)   # 本地执行
                print(f"  [工具] {name}({args}) -> {result}")
            except Exception as e:                   # 工具失败也要回填，让模型自己决定下一步
                result = f"工具执行失败: {e!r}"
            messages.append({"role": "tool", "tool_call_id": tc.id,
                             "content": str(result)})
        # 循环回到顶部：带着工具结果再问模型（Observation → 下一轮 Thought）
    return "（达到最大轮数，强制停止）"

# ---------- 多轮对话外壳：维护历史 + 滑动窗口截断（第 3.4/3.5 件事） ----------
def main():
    history: deque[dict] = deque(maxlen=24)          # 滑动窗口，防上下文爆窗
    print("输入 exit 退出。试试：成都和北京哪个更热？再帮我算 26+15")
    while True:
        user = input("你: ").strip()
        if user in {"exit", "quit"}:
            break
        history.append({"role": "user", "content": user})
        messages = [{"role": "system", "content": SYSTEM}, *history]
        answer = run_agent(messages)                 # ① 整段对话进 Agent 循环
        history.append({"role": "assistant", "content": answer})   # ② 答案入历史
        print(f"助手: {answer}")

if __name__ == "__main__":
    main()

跑起来的样子：

你: 成都和北京哪个更热？再帮我算 26+15
  [工具] get_weather({'city': '成都'}) -> 26℃ 多云
  [工具] get_weather({'city': '北京'}) -> 15℃ 晴
  [工具] add({'a': '26', 'b': '15'}) -> 41.0
助手: 成都更热（26℃ vs 15℃）；26+15=41。

注意模型自主决定了调用顺序和次数（两查天气 + 一算数，一次往返全并发请求）——这就是 Agent 与 Workflow 的分界（第 1 节的定义）。

5. 框架全景：它们各自藏起了哪几行

全部为 2026-09-05 核实的官方信息（出处见文末）：

结论（回到本章开头的判断）：手写过第 4 节的 100 行之后，你的问题从"这个框架怎么用"变成"它把哪几行代码藏哪了"——读框架源码时直接找它的 while 循环、工具注册表、消息组装函数这三样，任何框架 10 分钟定位到骨架。什么时候值得上框架：需要多智能体编排、人工审批节点、状态持久化/恢复、流式中间事件——这些"生产件"自己写费时间；单 Agent 简单场景，100 行裸写往往更好调试。

6. MCP：把工具变成标准化服务

6.1 是什么，为什么重要

MCP（Model Context Protocol，模型上下文协议）= Anthropic 2024-11 开源的工具/资源标准化协议：任何 LLM 应用（Host，如 ZCode、Claude Desktop、Cursor）通过统一协议调用任何 MCP Server 暴露的 tools/resources/prompts——"AI 应用的 USB-C 接口"。协议基于 JSON-RPC 2.0，传输支持 stdio（本地子进程）与 Streamable HTTP（网络服务）。

三条与时俱进的事实（出处见文末）：

1. 规范持续演进：当前最新版本 2026-07-28（spec 站点版本历史：2024-11-05 → 2025-06-18 → 2025-11-25 → 2026-07-28）
2. 2025-12-09 MCP 捐赠给 Linux 基金会旗下 Agentic AI Foundation（AAIF），由中立基金会治理，已获行业广泛支持
3. 官方 Python SDK 已发布 v2（大重构，pip install mcp 默认装 2.x），原生支持 2026-07-28 规范

💡 你每天用的 ZCode 就内置 MCP 客户端（~/.zcode/cli/config.json 里注册 MCP server）——你已经在 MCP 的消费端了，本章教你自己当供给端。

6.2 写一个 MCP Server（官方 v2 SDK，15 行）

uv add "mcp[cli]"

# server.py —— 完整可用的 MCP 服务器（官方 README 示例，v2 写法）
from mcp.server import MCPServer

mcp = MCPServer("Demo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@mcp.resource("greeting://{name}")
def greeting(name: str) -> str:
    """Greet someone by name."""
    return f"Hello, {name}!"

注意你没有写的东西：没有 JSON Schema（a: int, b: int 类型注解就是 schema）、没有请求解析、没有校验、没有协议处理。调试用官方 Inspector：

uv run mcp dev server.py     # 打开浏览器可视化调工具

6.3 写一个 MCP Client

import asyncio
from mcp import Client

async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:   # Streamable HTTP
        result = await client.call_tool("add", {"a": 1, "b": 2})
        print(result.structured_content)      # {'result': 3}

asyncio.run(main())

把 MCP server 的工具动态转成第 4 节 Agent 的 TOOLS schema（MCP 工具的 JSON Schema 协议与 function calling 同构），你就得到了一个"即插即用工具"的 Agent——这正是 ZCode/Claude Code 连接 MCP 的原理。

7. Agent 设计模式与常见坑

1. 必须有 max_rounds：模型可能陷入"调工具→不满意→再调"死循环，循环上限是生产底线（上面代码的 range(1, max_rounds+1)）
2. 工具失败不要抛异常终止循环：把错误信息作为 tool 结果回填，让模型自行决策（重试/换路/告知用户）——这是和普通后端代码最大的思维差异
3. 工具 description 是给模型看的 API 文档：写清楚"什么时候用我"，模型选错工具 90% 是描述烂
4. 工具结果要瘦身：网页全文、大 JSON 直接入 messages 会瞬间吃爆上下文（第 3.4 节）
5. Human-in-the-loop：危险操作（删数据、发邮件）在工具执行前插入人工确认节点——Anthropic 工程博客对此有专门论述（见文末）
6. 从 Workflow 开始：Anthropic 的建议——能预先编排就别上自主 Agent，简单优先（simple, composable patterns）

8. 自测清单与练习

• [ ] Workflow 和 Agent 的分界线是什么（一句话）？
• [ ] ReAct 论文的年份、机构、三个循环要素？
• [ ] 五件核心事闭卷说出来；每件对应你 100 行代码里的哪几行？
• [ ] 上下文爆窗的三种策略各自的代价？
• [ ] LangChain 1.0 的 Agent 建立在什么之上？create_agent 替你写了哪个函数？
• [ ] MCP 的三种原语（tools/resources/prompts）？两种传输方式？最新规范版本号？
• [ ] 为什么工具执行失败要回填而不是 raise？

练习 1：给 100 行 Agent 加第 3 个工具 read_file(path)（限白名单目录），让模型总结一个 markdown 文件。 练习 2：把 history 从 deque 换成 SQLite 表（第 3.5 节结构），实现跨进程重启的会话恢复。 练习 3：实现 3.4 节的"摘要压缩"：窗口超 20 条时用便宜模型（deepseek-v4-flash）压缩最老的 10 条。 练习 4：把 6.2 的 MCP server 跑起来，用 Inspector 调用 add；再写脚本让 100 行 Agent 通过 MCP client 调用这个工具（提示：Client 的 list_tools() 拿 schema 转 TOOLS）。

9. 参考与出处

以下全部为官方一手来源（访问日期：2026-09-05）：

⬅️ 返回目录 ｜ ➡️ 下一章：06-数据科学与ML