AI Agent 开发（30）：端到端数据分析实战

第 29 篇讲部署，Agent 从脚本变成服务，系列的工具箱到那一篇就齐了。但工具箱不等于项目：29 篇文章各自解决一个问题，还没有一次在同一个程序里合体。本篇是《AI Agent 开发》系列第 30 篇，收官篇，不复述新知识——把全系列的积木拼成一个真实可用的项目，每用到一个前文概念都点名出处。

项目设定：一个对接 SQLite 的销售数据分析 Agent。用户用自然语言提问，"上季度哪个大区增长最快"，Agent 自己查库、算数、给出有数据支撑的结论；支持多轮追问，"再按渠道拆一下"必须听得懂；记得住用户偏好，换个会话也不忘；涉及数据修正的写操作，一律人工审批后执行。全文按构建顺序推进：工具层 → Agent 核心 → 状态与记忆 → 写操作审批 → 评测与上线，最后给一张全系列知识地图。代码基于 langgraph 1.2.x（本篇核对到 1.2.16）与 langchain-core 1.6.x。

1、需求与总架构

需求收敛成四句话：问得自然，自然语言进；答得扎实，结论必须挂着 SQL 与数据行；聊得下去，多轮追问成立；改得安全，写操作过人工审批。四句话各自对应一组前文积木，这是收官篇最重要的事实：架构不是重新设计的，是从 29 篇里长出来的。总架构如下：

自然语言提问："上季度哪个大区增长最快？"
        │  请求携带 thread_id（会话）与 user_id（身份）
        ▼
FastAPI 服务外壳（第 29 篇）
 └─ create_agent（第 10 篇，底座是第 6～9 篇那张图）
     ├─ response_format：AnalysisReport 结构化报告（第 4 篇）
     ├─ dynamic_prompt：检索偏好注入提示词（第 23 篇）
     │    └─ Store[("prefs", user_id)] 跨会话记忆
     ├─ HumanInTheLoopMiddleware：写操作审批（第 11、28 篇）
     │    └─ execute_fix → interrupt → 人工批准 → Command(resume)
     ├─ checkpointer：多轮追问与断点存档（第 12 篇）
     └─ 工具两路
          ├─ MCPAdapter（第 21 篇）──stdio──▶ sales-db 只读 Server（第 20 篇）
          │    get_schema / describe_table / query，护栏见第 22 篇
          └─ save_preference 与 execute_fix（进程内工具，第 23、28 篇）
上线守门：10 条回归问答 + LangSmith 评测（第 27 篇）

解读：请求自上而下带两个 ID——thread_id 标会话、user_id 标身份，第 12 篇的会话模型与第 23 篇的 namespace 都靠它们路由。主体是第 10 篇的 create_agent，它底下那张图正是第 6～9 篇手搭过的模型节点、工具节点与条件回边。工具分两路：只读查询走 MCP Server，独立进程、跨应用复用（第 19～21 篇）；偏好写入与数据修正留在进程内，前者要拿 Store，后者要可审计。两个持久层各管一段：checkpointer 管会话内与审批断点，Store 管跨会话偏好。下面五节按图施工，每节只加几行装配。

2、SQL 工具层

第一步把数据能力做成独立进程：名为 sales-db 的 MCP Server。为什么不让工具直接住在 Agent 进程里？第 19 篇给过答案：进了 Server，Claude Desktop、同事的 Agent 与本项目共用同一套工具，工具的修改变成服务的发布。写法沿用第 20 篇核实的 SDK v2 类名 MCPServer，细节看官方 MCP Python SDK 仓库。

示例数据两张表：orders 销售订单（region 大区、channel 渠道、amount 金额、created_at 下单日）与 regions 大区维表。第一件工具与第 20 篇的差异在描述层——第 22 篇讲过，描述就是写给模型读的 API 文档，这里干脆把业务口径做成一个工具：

import sqlite3
from contextlib import closing

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("sales-db")  # v2 类名，第 20 篇核实

SCHEMA_DOC = {
    "orders": "销售订单。列：region 大区、channel 渠道、"
              "amount 金额（元）、created_at 下单日。",
    "regions": "大区维表。列：code、name、manager。",
}

def _connect() -> sqlite3.Connection:
    conn = sqlite3.connect("file:sales.db?mode=ro", uri=True)  # 物理只读
    conn.row_factory = sqlite3.Row
    return conn

@mcp.tool()
def get_schema() -> dict:
    """各表业务口径与列说明，写查询前必读。"""
    return {"tables": SCHEMA_DOC}

解读：get_schema 返回的是业务口径而非裸表结构——amount 的单位、channel 的语义，模型不读它就只能猜，而猜口径是数据分析的第一大事故源。它还顺手吞并了第 20 篇的 list_tables：表清单就是字典的键，第 22 篇说过，同构工具合并，误用率反而下降。describe_table 原样照搬第 20 篇（PRAGMA 查列信息，表名先过 isalnum() 白名单），此处不再贴。

查询工具的四层护栏同样从第 20 篇整段搬来，改动只有库名：

import re

from mcp.server.mcpserver.exceptions import ToolError

ALLOWED = re.compile(r"(?is)^\s*(select|with)\b")
BANNED = re.compile(r"(?i)\b(insert|update|delete|drop|alter|create)\b")

@mcp.tool()
def query(sql: str, limit: int = 50) -> list[dict]:
    """只读查询：仅接受单条 SELECT/WITH，结果强制 LIMIT。"""
    sql = sql.strip().rstrip(";")
    if ";" in sql:
        raise ToolError("拒绝多语句执行")
    if not ALLOWED.match(sql):
        raise ToolError("仅允许 SELECT/WITH 开头")
    if BANNED.search(sql):
        raise ToolError("检测到写操作关键字")
    if not re.search(r"(?i)\blimit\s+\d+\s*$", sql):
        sql += f" LIMIT {min(max(limit, 1), 200)}"  # 上限钳在 200
    with closing(_connect()) as conn:
        return [dict(r) for r in conn.execute(sql).fetchall()]

if __name__ == "__main__":
    mcp.run()  # stdio 传输，Host 是第三步的 MCPAdapter

解读：连接层 mode=ro 物理只读，UPDATE 在连接层就报错；语句层白名单只放行 SELECT 与 WITH、黑名单拦写动词；结果层强制 LIMIT 200，防一句"查全表"撑爆上下文。三层各防一类事故，这个项目最贵的安全资产就是这段二十行的旧代码。

第 20 篇的规矩在收官篇原样成立：LLM 生成的 SQL 永远不该被默认信任——只读护栏不是 Agent 的功能，是 Agent 的地基。

3、Agent 核心组装

第二步组装 Agent 主体，三件积木：create_agent 出图（第 10 篇）、MCPAdapter 接工具（第 21 篇）、response_format 钉死报告形态（第 4 篇结构化输出思想的 Agent 版）。先定义报告 schema：

from pydantic import BaseModel, Field

class AnalysisReport(BaseModel):
    """数据分析报告的固定结构。"""
    question: str = Field(description="用户的问题")
    answer: str = Field(description="结论：先答案，后数据依据")
    sql_used: list[str] = Field(description="本次执行过的 SQL")
    rows: list[dict] = Field(description="支撑结论的关键数据行")
    caveats: list[str] = Field(
        default_factory=list, description="数据口径的局限，如时间范围")

解读：第 4 篇的 with_structured_output 解决"单次调用吐 JSON"；Agent 场景要的是"干完工具的活再按 schema 交作业"，所以用第 10 篇的 response_format。报告的形态由 schema 钉死：answer 先结论后依据；rows 让下游拿原始数行直接画图表；caveats 强迫模型交代口径局限——分析报告最常见的事故不是算错数，是把"只统计了线上渠道"这种前提藏起来。

装配与首轮调用如下，CONFIG 与第 21 篇接本地 Server 的配置同款，用法细节见 Agents 官方文档与 LangChain MCP 文档：

import asyncio

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.mcp import MCPAdapter

SYSTEM = ("你是销售数据分析师。先调 get_schema 了解口径，"
          "再写只读 SQL 验证，结论必须有数据支撑。")

CONFIG = {"mcpServers": {"sales-db": {
    "command": "uv",
    "args": ["--directory", "E:/agent/sales-db", "run", "server.py"],
}}}

async def main():
    async with MCPAdapter(CONFIG) as adapter:  # 第 21 篇的接入姿势
        tools = await adapter.list_tools()     # 发现 3 个工具
        agent = create_agent(
            init_chat_model("openai:gpt-5"),
            tools=tools,
            system_prompt=SYSTEM,
            response_format=AnalysisReport,    # 第 4 篇的结构化输出
        )
        result = await agent.ainvoke({"messages": [
            {"role": "user", "content": "上季度哪个大区销售额最高？"}]})
        print(result["structured_response"].answer)

asyncio.run(main())

解读三件事。其一，list_tools() 返回的就是标准 LangChain 工具，与进程内工具合并成一份列表，模型不感知进程边界——第 21 篇的核心结论。其二，全程 async：MCP 工具只有异步实现，invoke 要换成 await ainvoke（第 13 篇）。其三，stdio 的上下文要常开（第 21 篇第 5 节），Agent 的整个生命周期活在 async with 里，第六步把它开进 FastAPI 的 lifespan。跑起来问一句"上季度哪个大区销售额最高"，模型自己走 get_schema → query 的路径，校验过的报告落在 result["structured_response"]，.model_dump_json() 直接给前端。

4、状态与偏好记忆

第三步让 Agent 记得住，两个维度：会话内的多轮追问、跨会话的偏好。checkpointer 管会话内，Store 管跨会话——第 12、23 篇的分工在此合体。偏好写入用工具式（第 23 篇 4.1 的结论：对话型 Agent 优先让模型自己决定记什么）：

import uuid
from dataclasses import dataclass

from langchain.tools import ToolRuntime, tool
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()  # 生产换 PostgresStore（第 23 篇）

@dataclass
class Context:
    user_id: str  # 身份随 context 走，不让模型当参数传

@tool
def save_preference(preference: str,
                    runtime: ToolRuntime[Context]) -> str:
    """记住用户长期有效的分析偏好，如"金额一律用万元"。"""
    uid = runtime.context.user_id
    runtime.store.put(("prefs", uid), str(uuid.uuid4()),
                      {"text": preference})
    return f"已记住：{preference}"

解读：与第 23 篇的 save_memory 同构，差异在取 Store 的姿势——第 23 篇在自建图里用 get_store()，create_agent 的工具签名里直接注入 runtime: ToolRuntime[Context]，Store 与用户身份一次到手（官方长期记忆文档的同款写法）。身份走 context，不走模型参数：user_id 若由模型当参数传，就是一个可被诱导、可被冒充的自由变量。checkpointer 选 AsyncSqliteSaver——第 12 篇 SQLite 版的异步实现，整条调用链是 async，持久层也得跟得上。

注入侧有个第 10 篇埋过的边界：system_prompt 是静态的，按用户动态拼提示词要用中间件。dynamic_prompt 装饰器把偏好检索接进每次模型调用：

from langchain.agents.middleware import dynamic_prompt

@dynamic_prompt
def prompt_with_prefs(request) -> str:
    uid = request.runtime.context.user_id  # 身份从 context 取
    items = request.runtime.store.search(("prefs", uid), limit=5)
    prefs = "\n".join(f"- {i.value['text']}" for i in items)
    return SYSTEM + (f"\n\n该用户的分析偏好：\n{prefs}" if prefs else "")

解读：钩子函数收 request，从 request.runtime 拿 context 与 store——每次模型调用前现场检索、现场拼接，偏好更新下一轮即生效；namespace 按 ("prefs", user_id) 分目录，多用户天然隔离（第 23 篇）。装配时，第 23 篇在 compile 会师的两个持久层，这里换成了 create_agent 的两个参数：

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

async with MCPAdapter(CONFIG) as adapter, \
        AsyncSqliteSaver.from_conn_string("checkpoints.db") as cp:
    await cp.setup()  # 首次建表，之后可省
    tools = await adapter.list_tools()
    agent = create_agent(
        init_chat_model("openai:gpt-5"),
        tools=[*tools, save_preference],
        middleware=[prompt_with_prefs],
        response_format=AnalysisReport,
        checkpointer=cp, store=store, context_schema=Context,
    )
    cfg = {"configurable": {"thread_id": "t-1"}}
    ctx = Context(user_id="u-42")
    await agent.ainvoke({"messages": [
        {"role": "user", "content": "金额一律用万元，帮我记住。"}]},
        cfg, context=ctx)
    r = await agent.ainvoke({"messages": [
        {"role": "user", "content": "那上月各渠道的数字呢？"}]},
        cfg, context=ctx)
    print(r["structured_response"].answer)  # 数字以万元呈现

解读：checkpointer 挂上后，两次 ainvoke 之间自动续历史——第二轮那个"那"字指代的是第一轮的话题，模型答得上来靠的是 checkpoint，不是运气。context=ctx 每次都要带：它不进 checkpoint，是每次调用的运行时身份。把 thread_id 换成全新的再问一句，历史归零、偏好仍在——第 23 篇第 5 节验证过的跨线程闭环，这里一行没改地兑现。

5、写操作审批闸门

第四步处理项目里唯一的危险动作：数据修正。"把订单 1001 的渠道改对"是分析场景的真实需求，UPDATE 不可逆，第 11 篇的结论直接适用。写工具刻意不进 MCP Server——读走 Server，可复用、无状态；写留在进程内，可审计、入口唯一：

import json
import re
import sqlite3
from contextlib import closing

from langchain.tools import tool

WRITE_OK = re.compile(r"(?is)^\s*(update|delete)\b.*\bwhere\b")
AUDIT_LOG = "audit.log"  # 第 28 篇：可审计是写操作的硬前提

@tool
def execute_fix(sql: str) -> str:
    """执行数据修正（UPDATE/DELETE），必须带 WHERE，需人工审批。"""
    sql = sql.strip().rstrip(";")
    if ";" in sql or not WRITE_OK.match(sql):
        return json.dumps({"error": "仅允许带 WHERE 的单条 UPDATE/DELETE"},
                          ensure_ascii=False)
    with closing(sqlite3.connect("sales.db")) as conn:  # 独立可写连接
        cur = conn.execute(sql)
        conn.commit()
        with open(AUDIT_LOG, "a", encoding="utf-8") as f:
            f.write(f"{cur.rowcount} rows | {sql}\n")
    return f"已修正 {cur.rowcount} 行"

解读：护栏比第 20 篇多一层正则——UPDATE/DELETE 必须带 WHERE，无差别改表在进库前就被拦下；执行结果追加审计日志，改了几行、语句原文全留档。第 28 篇的三件套"最小权限、审批、审计"在这里集齐两件，第三件是下面的闸门，用第 11 篇 6.2 的标准中间件：

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.types import Command

policy = {t.name: False for t in [*tools, save_preference]}
policy["execute_fix"] = {  # 写操作必须人工点头
    "allowed_decisions": ["approve", "edit", "reject"]}

agent = create_agent(
    init_chat_model("openai:gpt-5"),
    tools=[*tools, save_preference, execute_fix],
    middleware=[prompt_with_prefs,
                HumanInTheLoopMiddleware(interrupt_on=policy)],
    response_format=AnalysisReport,
    checkpointer=cp, store=store, context_schema=Context,
)

cfg = {"configurable": {"thread_id": "t-2"}}
r1 = await agent.ainvoke({"messages": [
    {"role": "user", "content": "把订单 1001 的渠道改成线上。"}]},
    cfg, context=Context(user_id="u-42"))
print(r1["__interrupt__"][0].value)  # 申请单：工具名 + SQL

r2 = await agent.ainvoke(  # 审批人点了"批准"
    Command(resume={"decisions": [{"type": "approve"}]}),
    cfg, context=Context(user_id="u-42"))
print(r2["structured_response"].answer)

解读：策略表覆盖全部工具——MCP 查询与 save_preference 直过，execute_fix 放开 approve、edit、reject 三种决策，语义见第 11 篇 6.2 的四种决策表。第一段调用带着 __interrupt__ 正常返回，申请单里是待审工具与参数；人工批准后 Command(resume=...) 送回决定，工具节点从头重跑、执行修正。恢复调用也要带 context——重跑的节点里钩子还要读 user_id，漏了就是当场 KeyError。以下代码都写在第三步的 async with 块内，cp 与 tools 沿用。

第 11 篇的话在收官篇值得再说一遍：人工介入不是对 Agent 的不信任，而是对不可逆动作的工程保险——闸门只装在最贵的那一步。

6、评测与上线

第五步，代码写完不等于能上线。第 27 篇的方法论：改动之前先有回归集，让 10 条问答替你踩一遍坑。

6.1、回归评测集

回归集与本地跑法如下，每条用独立 thread_id，用例之间互不污染：

CASES = [  # 10 条回归问答：问题 + SQL 里必须出现的关键词
    ("各渠道上月销售额合计", "channel"),
    ("上季度哪个大区增长最快", "region"),
    ("全年销售额最高的 5 天", "ORDER BY"),
    ("线上渠道的订单数", "channel"),
    ("大区销售额排名", "ORDER BY"),
    ("3 月销售额环比 2 月", None),
    ("哪个大区客单价最高", "amount"),
    ("销售额月度趋势", "strftime"),
    ("订单 1001 的详情", "1001"),
    ("华东大区线上渠道占比", "region"),
]

async def run_eval(agent, ctx):
    passed = 0
    for i, (question, keyword) in enumerate(CASES):
        cfg = {"configurable": {"thread_id": f"eval-{i}"}}
        r = await agent.ainvoke({"messages": [
            {"role": "user", "content": question}]}, cfg, context=ctx)
        report = r["structured_response"]
        ok = bool(report.sql_used) and (keyword is None or any(
            keyword.lower() in s.lower() for s in report.sql_used))
        passed += ok
        print(("PASS " if ok else "FAIL ") + question)
    print(f"{passed}/{len(CASES)}")

解读：断言用"SQL 里必须出现关键词"这种笨办法——确定性、零额外成本、毫秒级跑完；它抓的是结构性事故：没查库就作答、GROUP BY 用错维度。结论对错的把关升级路是 LLM-as-judge（第 27 篇）：把 answer 与 rows 喂给评审模型评分；再把同一组问答推成 LangSmith 数据集，CI 里跑 evaluate，每次改动留 trace，用法见 LangSmith 文档。评测集是 Agent 的单元测试，它应当先于新功能存在。

6.2、FastAPI 部署骨架

外壳沿用第 29 篇自托管路线的骨架，装配决策全部在前文出现过：

from contextlib import asynccontextmanager

from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with MCPAdapter(CONFIG) as adapter:  # stdio 常开（第 21 篇）
        tools = await adapter.list_tools()
        async with AsyncSqliteSaver.from_conn_string(
                "checkpoints.db") as cp:
            app.state.agent = build_agent(tools, cp)  # 前四步的装配
            yield

app = FastAPI(lifespan=lifespan)

@app.post("/chat")  # body：{thread_id, user_id, message}
async def chat(body: dict):
    cfg = {"configurable": {"thread_id": body["thread_id"]}}
    ctx = Context(user_id=body["user_id"])
    result = await app.state.agent.ainvoke(
        {"messages": [{"role": "user", "content": body["message"]}]},
        cfg, context=ctx)
    if "__interrupt__" in result:  # 写操作挂起，等人工审批
        return {"status": "pending_approval",
                "requests": [i.value for i in result["__interrupt__"]]}
    return {"status": "done",
            "report": result["structured_response"].model_dump()}

解读：其一，MCPAdapter 上下文开在 lifespan——服务活多久 stdio 子进程活多久，正是第 21 篇"长驻服务常开上下文"的形态，build_agent 就是前四步装配的封装。其二，thread_id 与 user_id 由调用方传入：图与服务全单例，会话靠 ID 隔离（第 12 篇）。其三，__interrupt__ 映射成 pending_approval 响应，前端渲染审批卡片，人工提交后再调一次带上 Command——两次 HTTP 之间隔着任意长的人工时间，状态全在 checkpointer 里。真上生产，第 29 篇的清单照走：Docker 化、checkpointer 与 Store 换 Postgres 后端、健康检查与灰度回滚。

7、全系列知识地图

复盘每一块积木的出处，30 篇到本项目模块的映射如下：

模块	篇目	本项目落点
全景与基础	01～05	模型与工具调用底座、报告 schema
LangGraph 核心	06～09	create_agent 底下那张图的全部原理
预构建与协同	10～11	create_agent 主体、interrupt 审批
持久化与流式	12～13	AsyncSqliteSaver 多轮、全异步链路
RAG	14～18	未启用，留作扩展
MCP 与工具	19～22	只读 Server、Adapter、护栏三件套
进阶架构	23～26	Store 偏好记忆，其余未启用
生产化	27～29	回归评测、审计、FastAPI 外壳

诚实交代两处留白。RAG（14～18 篇）这个项目一行没用：两张表的口径一个 get_schema 就装得下，检索是杀鸡用牛刀。多 Agent（24 篇）同理：单循环加五件工具，还没到需要 Supervisor 的复杂度——第 25 篇横评教的选型判断，第一步是不用。没用到不是缺憾，是选型；扩展位也留得清楚：历次分析报告入库做语义检索（第 15 篇），或把取数与成文拆成子代理（第 24 篇）。

8、总结

30 篇到此收官，回头看有三条主线。

第一条控制权。第 6 篇的 StateGraph、第 9 篇的手写 ReAct、第 10 篇的 create_agent，系列在造轮子与用轮子之间反复往返，只为一个目的：知道引擎里每个循环长什么样，出问题才知道下沉到哪一层去改。本篇每一步装配都能拆开看内部，就是这条主线换来的底气。

第二条可靠性。护栏（20、22 篇）、审批（11 篇）、记忆（12、23 篇）、评测（27 篇）、审计（28 篇）——Agent 从 demo 到产品的距离，几乎全部在模型之外。这个项目里模型真正做的决定只有一件：下一步调哪件工具；其余的每一层都是工程。

第三条生态。MCP（19～21 篇）让工具成为跨进程、跨应用的服务；框架横评（25 篇）让选型有据可依；部署（29 篇）让服务真正落地。LangChain 只是系列的教学载体——图、工具、记忆、审批这些积木的形状，在任何框架里都成立。

练手方向：把 orders 换成你自己的数据域——工单、日志、CRM——10 条回归问答换成你的业务问题，这个项目就是你的第一个生产 Agent；有余力的把历史报告入库做成 RAG 检索，补上 14～18 篇那块留白。

系列的方法论浓缩成一句话：先让 Agent 可控，再让它可靠，最后让它可迁移——三样都齐了，才谈得上智能。

对于跟完 30 篇的读者，建议下一步做三件事：通读 create_agent 源码，验证第 9 篇的积木如何拼成第 10 篇的产品；拿本项目当底座做一次端到端的换域改造；持续跟进 MCP 规范与模型迭代——文章会过时，方法不会。感谢 30 篇同行，下个系列见。