手搓AI Agent框架：Plan-Execute闭环与MCP工程实践

昨日夕阳

443人浏览 · 2026-06-21 15:25:36

昨日夕阳 · 2026-06-21 15:25:36 发布

1. 为什么“手搓 AI Agent 框架”不是炫技，而是工程能力的分水岭

你有没有过这种体验：用 LangChain 写了个“天气查询 Agent”，跑通了 demo，一上线就崩——用户问“帮我查下后天上海浦东机场的航班延误情况”，Agent 却去调了天气 API；或者用户连续追问三次“那明天呢？后天呢？大后天呢？”，Agent 突然忘了上下文，开始胡编乱造；更别提当你要接入公司内部的 CRM、ERP、钉钉审批流时，发现 LangChain 的 Tool 定义要重写三遍，参数校验要自己补，错误重试逻辑得从零堆，最后代码比业务逻辑还多。

这不是你学得不够快，而是你一直站在“API 调用者”的位置上，没真正踩进“Agent 构建者”的地里。LangChain、LlamaIndex 这些库，本质是预制菜——给你切好的肉、配好的酱、定好火候的说明书。但真实业务场景里，你面对的是一整头活牛：要自己选部位（工具调度策略）、判断肥瘦（上下文压缩精度）、处理筋膜（异常传播路径）、甚至决定要不要把牛骨熬成高汤（长期记忆沉淀）。而“手搓框架”，就是逼你亲手摸一遍这头牛的每一块骨头、每一根血管。

我去年带团队落地一个面向保险理赔员的 AI 助理，初期直接套 LangChain 的 ReAct Agent，两周就跑通了“查保单+算赔款”流程。但第三周开始暴雷：当用户说“把张三上个月在协和医院的住院记录同步到理赔系统”，Agent 在“找医院”环节卡死——它不知道“协和医院”在北京还是武汉，也不知道该查医保系统还是医院 HIS 接口；第四周更糟，用户上传一张手写的门诊发票照片，Agent 直接返回“不支持图片格式”，完全没触发 OCR 工具链。我们这才意识到：不是模型不行，是整个执行骨架太脆——没有明确的 Plan 阶段做意图拆解，没有隔离的 Execute 环境做工具沙箱，没有可追溯的 State 机制做步骤回滚。

于是我们停掉所有业务迭代，关起门来三个月，从零写了一个叫 Aegis 的轻量级 Agent 框架。它不追求功能大全，只解决三个致命问题： Plan 必须可解释、Execute 必须可中断、State 必须可审计 。现在这个框架已支撑日均 2.3 万次复杂理赔咨询，平均任务完成率从 61% 提升到 94.7%，最关键的是——当新同事接手时，他不用再翻 LangChain 文档猜 agent_executor 的 max_iterations 设多少合适，而是直接看 aegis/planner/ 目录下的 ReActPlanner.py ，里面 87 行代码清清楚楚写着：“当检测到地理歧义词时，强制进入澄清子流程，最多追问 2 次，超时则降级为北京协和”。

所以别再把“手搓框架”当成极客玩具。它是一面照妖镜，照出你对 Agent 本质的理解深度；它是一把手术刀，让你能精准切开 LLM 黑盒里的决策链路；它更是一道门槛——跨过去的人，才能真正主导 AI 应用的工程化落地，而不是永远在 API 文档里打转。

2. Plan-and-Execute 不是流程图，而是 Agent 的“神经反射弧”

很多人把 Plan-and-Execute 当成一个简单的两步流程：先让 LLM 输出“我要做 A→B→C”，再按顺序执行。这就像以为人走路只是“抬腿→迈步→落脚”三个动作——忽略了小脑实时校准肌肉张力、前庭系统感知重心偏移、脊髓反射弧在踩到香蕉皮瞬间自动绷紧脚踝肌群。真正的 Plan-and-Execute，是 Agent 的神经反射弧，必须包含 感知-决策-执行-反馈 的闭环，且每个环节都要有物理层面的实现保障。

2.1 Plan 阶段：为什么“让 LLM 自由发挥”是最大陷阱

我们最初在 Aegis 框架里也犯过这个错。第一版 Planner 直接用 ChatAnthropic(model="claude-3-haiku-20240307") + system prompt：“请将用户请求分解为可执行的原子步骤，用 JSON 格式输出，字段包括：step_id, action, parameters, dependencies”。结果上线三天，日志里全是这样的 Plan：

{
  "steps": [
    {
      "step_id": "1",
      "action": "query_medical_record",
      "parameters": {"patient_id": "张三", "hospital": "协和"},
      "dependencies": []
    }
  ]
}

问题在哪？ "hospital": "协和" 这个参数根本无法直连任何系统。真实医疗系统要求的是 hospital_code: "BJXH001" ，而“协和”在北京、武汉、福州都有分院。LLM 在 Plan 阶段就把歧义埋进了参数，Execute 阶段只能硬着头皮调用，然后必然失败。

解决方案：Plan 必须强制结构化 + 语义校验前置
我们在第二版 Planner 里做了三件事：

定义原子 Action Schema ：所有可执行动作必须提前注册，比如 medical_record_query 的 schema 明确要求 hospital_code 是必填字符串，且值必须来自预加载的 hospital_codes.json ；
引入 Plan 预检器（Plan Validator） ：在 LLM 输出 JSON 后，不直接执行，而是用 Pydantic V2 的 RootModel 做强类型校验，对 hospital_code 字段额外加 @field_validator ，检查其是否在白名单内；
设计澄清子流程（Clarification Subflow） ：当校验失败时，不报错，而是自动生成澄清问题：“请问您指的是北京协和医院（代码 BJXH001）、武汉协和医院（代码 WHXH001）还是福州协和医院（代码 FZXH001）？”并冻结当前 Plan，等待用户确认后再重试。

提示：Plan Validator 不是装饰器，而是独立模块。我们把它设计成可插拔的——金融场景用 FinanceValidator 校验身份证号、银行卡号格式；政务场景用 GovValidator 校验行政区划代码。这样当业务线切换时，只需替换 validator 实例，无需动 Planner 核心逻辑。

2.2 Execute 阶段：工具调用不是发 HTTP 请求，而是启动“可控进程”

很多教程教你怎么用 requests.post() 调工具，却从不告诉你：当工具响应超时、返回 503、或 JSON 解析失败时，Agent 该怎么反应？是重试？降级？还是直接放弃整个任务？在 Aegis 里，我们把每次工具调用都视为启动一个“可控进程”，它必须具备四个物理属性：

属性	说明	Aegis 实现方式
超时控制	防止某个工具卡死整个 Agent	每个 Tool 类继承 `BaseTool` ，强制实现 `timeout: float = 15.0` 字段，Execute Engine 用 `asyncio.wait_for(task, timeout=tool.timeout)` 包裹
熔断机制	连续 3 次失败后，该工具自动进入 5 分钟休眠期	使用 `circuitbreaker` 库，状态存储在内存字典 `circuit_states = {"ocr_tool": {"failure_count": 0, "last_failure": None}}`
沙箱环境	工具执行不能污染 Agent 主进程内存	每个工具调用在独立 `asyncio.TaskGroup` 中运行，通过 `contextvars` 隔离上下文变量
可观测性	每次调用必须记录输入、输出、耗时、状态	所有 Tool 类统一继承 `TracedTool` ，自动注入 OpenTelemetry Span

举个真实案例：我们接入的 OCR 工具在高峰期经常返回 {"code": 503, "msg": "服务繁忙"} 。旧方案是简单重试 2 次，结果用户等了 30 秒看到“识别失败”。新方案中，OCR Tool 的 circuitbreaker 在第三次失败后自动熔断，Execute Engine 立即触发降级策略：调用本地 Tesseract 引擎做基础文字提取，并在最终回复里标注“【降级识别】仅提取可见文字，未识别表格结构”。

2.3 State 管理：为什么“把 history 传给 LLM”是最危险的偷懒

LangChain 默认把整个对话 history 作为 messages 传给 LLM，美其名曰“提供上下文”。但在复杂任务中，这相当于让司机一边开车一边背诵整本《道路安全法》——信息过载导致关键线索被淹没。我们曾遇到一个典型故障：用户先问“查张三的保单”，Agent 返回保单号；用户再问“把这份保单发给李四”，Agent 却去查李四的保单，因为 LLM 在长 history 里误判了“这份”指代对象。

Aegis 的 State 分层设计 ：
我们把 State 拆成三层，每层解决不同问题：

Session State（会话层） ：存储用户 ID、设备信息、渠道来源（微信/APP/网页），生命周期=单次会话，用 Redis Hash 存储，key 为 session:{user_id}:{channel} ；
Task State（任务层） ：存储当前进行中的任务 ID、已执行步骤、各步骤输出、Plan 版本号，生命周期=单个任务，用 Redis Stream 存储，每条消息是 {"step_id": "1", "output": {"policy_no": "P2024001"}} ；
Memory State（记忆层） ：存储跨任务的长期知识，如“张三的身份证号是 XXX”，用向量数据库（Qdrant）存储，通过 memory_retriever 按需注入。

当用户说“把这份保单发给李四”时，Execute Engine 先查 Task State 获取最新保单号，再查 Session State 确认当前用户是张三，最后调用 memory_retriever 检索“李四”的邮箱——所有关键信息都来自结构化存储，而非依赖 LLM 从混乱 history 中“猜”。

注意：Task State 的每一步输出都强制 JSON Schema 校验。比如保单查询工具的输出必须符合 {"policy_no": str, "insured_name": str, "status": Literal["active", "expired"]} 。这让我们能在日志里直接用 jq '.output.policy_no' 提取所有保单号，做实时监控。

3. MCP 协议不是“又一个标准”，而是 Agent 的“USB-C 接口”

当你第一次听说 Model Context Protocol（MCP），可能会觉得它像 W3C 的 HTML 标准——遥远、抽象、和你手头的项目无关。但实际用起来你会发现：MCP 是 Agent 生态里最接近“USB-C 接口”的存在——它不规定你做什么（比如“必须支持天气查询”），只规定你怎么做（比如“如何声明工具、如何传递参数、如何返回错误”）。这意味着，只要你的工具实现了 MCP，它就能即插即用地接入任何支持 MCP 的 Agent 框架，无论是 LangChain、LlamaIndex，还是你手搓的 Aegis。

3.1 为什么 STDIO 传输模式是 MVP 阶段的最优解

网络上很多教程一上来就教你搭 HTTP MCP Server，但我们在 Aegis 的早期验证阶段，坚持用 STDIO 模式，原因很实在：

零依赖部署 ：STDIO Server 就是一个 Python 脚本， uv run weather.py 就能跑起来。而 HTTP Server 需要 Uvicorn、端口管理、HTTPS 证书、反向代理配置——这些在验证阶段纯属噪音；
调试极其直观 ：你用 cat test_input.json | python weather.py 就能看到完整输入输出流，连 print() 都不用改，直接 stdout 就是协议响应。HTTP 模式下你得开 Postman、设 header、处理 CORS，调试成本翻倍；
天然进程隔离 ：每个 STDIO Server 是独立进程，一个工具崩溃不会影响其他工具。HTTP Server 如果用单进程模式，一个工具死循环会拖垮整个服务。

我们当时为理赔系统写了三个 MCP 工具： ocr_tool （调用内部 OCR 服务）、 crm_search （查客户关系系统）、 approval_flow （发起钉钉审批）。全部用 STDIO 模式开发，每个工具就是一个 200 行左右的 .py 文件， main() 函数里只做三件事：读 stdin → 解析 MCP 请求 → 调用内部 SDK → 构造 MCP 响应写 stdout。开发周期从预估的 5 天压缩到 1.5 天。

3.2 Streamable HTTP 模式：当你的工具需要“呼吸感”

STDIO 很好，但它有个硬伤： 不支持流式响应 。比如用户上传一张 10MB 的 PDF 保单，OCR 工具需要 8 秒处理，期间 Agent 只能干等，用户体验就是“卡住”。而 Streamable HTTP 模式通过 Server-Sent Events（SSE）解决了这个问题。

Aegis 对 Streamable HTTP 的适配，核心在于重构了 Execute Engine 的等待逻辑：

# 旧版 STDIO 等待（伪代码）
async def execute_stdio_tool(tool_name: str, input_data: dict):
    proc = await asyncio.create_subprocess_exec(
        "uv", "run", f"tools/{tool_name}.py",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, _ = await proc.communicate(json.dumps(input_data).encode())
    return json.loads(stdout.decode())

# 新版 Streamable HTTP 等待（伪代码）
async def execute_streamable_http_tool(tool_url: str, input_data: dict):
    async with aiohttp.ClientSession() as session:
        async with session.post(f"{tool_url}/call", json=input_data) as resp:
            # 关键：用 SSE 流式读取响应
            async for line in resp.content:
                if line.startswith(b"data:"):
                    chunk = json.loads(line[5:].strip())
                    # 实时更新 UI 或记录进度
                    update_progress(chunk.get("progress", 0))
                    if chunk.get("status") == "completed":
                        return chunk.get("result")

真实案例：当用户上传 PDF 保单时，OCR 工具的 Streamable HTTP Server 会分三段推送：

data: {"progress": 30, "status": "processing", "message": "正在解析文档结构"}
data: {"progress": 75, "status": "processing", "message": "OCR 识别中，已处理 12/16 页"}
data: {"progress": 100, "status": "completed", "result": {"text": "保单号：P2024001...", "tables": [...]}}

Aegis 的前端 SDK 收到这些事件后，能实时显示进度条和状态文案，用户不再感觉“卡死”，而是清晰知道“系统正在工作”。

3.3 MCP 工具开发的黄金三原则

基于 17 个 MCP 工具的实战经验，我们总结出三条铁律，违反任何一条都会导致集成灾难：

原则一：输入即契约，拒绝任何“智能猜测”
错误做法： ocr_tool 接收 {"file_url": "https://xxx.com/1.pdf"} ，然后自己去下载文件。
正确做法：MCP 规范要求工具接收原始二进制数据（base64 编码）， {"file_data": "JVBERi0xLjQK..."} 。这样做的好处是：Agent 可以在调用前做安全扫描（比如查病毒）、做格式校验（比如确认是 PDF）、做大小限制（比如 >50MB 直接拒绝）。把“下载”这个动作交给 Agent，工具只负责“识别”，职责清晰。

原则二：错误即数据，禁止裸抛异常
错误做法： crm_search 工具里 if not user_id: raise ValueError("user_id required") ，结果 MCP 客户端收到 500 错误，无法区分是参数错误还是服务宕机。
正确做法：所有错误必须包装成标准 MCP Error Response：

{
  "error": {
    "code": "INVALID_PARAMETER",
    "message": "user_id is required and must be a string",
    "details": {"field": "user_id", "received": null}
  }
}

Aegis 的 MCP Client 会自动解析 error.code ，如果是 INVALID_PARAMETER 就触发 Plan 重试（让用户补全信息），如果是 SERVICE_UNAVAILABLE 就走熔断降级。

原则三：状态即接口，暴露健康检查端点
每个 MCP 工具必须提供 /health 端点，返回：

{
  "status": "healthy",
  "version": "1.2.0",
  "dependencies": [{"name": "ocr-engine", "status": "healthy"}]
}

Aegis 的 Agent 启动时会批量探测所有已注册工具的 /health ，生成实时健康看板。当 crm_search 工具的依赖数据库连接池耗尽时， /health 返回 {"status": "degraded"} ，Aegis 会自动将后续请求路由到备用 CRM 工具（如果配置了的话），实现无感故障转移。

4. LangGraph 不是“升级版 LangChain”，而是 Agent 的“电路板”

很多人把 LangGraph 当成 LangChain 的“高级版本”，觉得“学会 LangGraph 就能画出酷炫流程图”。但实际用下来你会发现：LangGraph 的核心价值不是可视化，而是把 Agent 的执行逻辑从“线性脚本”升级为“可编程电路板”——你可以像焊接电阻、电容一样，把 Planner、Executor、Validator 这些模块，用标准接口（ StateGraph ）焊接到一起，形成稳定可靠的电流回路。

4.1 为什么 Aegis 拒绝直接集成 LangGraph

我们认真评估过 LangGraph，最终决定不直接集成，原因很务实：

State Schema 绑定过死 ：LangGraph 要求所有节点共享同一个 BaseModel ，而 Aegis 的 Session State、Task State、Memory State 是三层异构结构，强行塞进一个 Pydantic Model 会导致 80% 的代码在做字段转换；
边缘 Case 处理太重 ：LangGraph 的 interrupt_before / interrupt_after 机制，在处理“用户中途取消任务”时，需要手动管理 checkpoint 和 thread_id ，而 Aegis 的 Task State 本身就有 status: Literal["running", "cancelled", "completed"] 字段，取消就是一条 Redis 命令 HSET session:{id} status cancelled ；
可观测性粒度太粗 ：LangGraph 的 trace 只能看到“Planner 节点耗时 1200ms”，但 Aegis 需要知道“Planner 里 validate_hospital_code 子函数耗时 850ms，是因为远程调用了医院代码服务”。

但这不意味着我们否定 LangGraph。相反，我们借鉴了它的 节点化思想 ，在 Aegis 里实现了更轻量的 Node 抽象：

class Node(ABC):
    @abstractmethod
    async def invoke(self, state: State) -> State:
        """核心执行逻辑，必须返回新 State"""
        pass
    
    @property
    @abstractmethod
    def name(self) -> str:
        """节点唯一标识，用于日志和监控"""
        pass

class PlannerNode(Node):
    def __init__(self, llm: BaseLLM):
        self.llm = llm
        self.name = "planner"
    
    async def invoke(self, state: State) -> State:
        # 这里调用我们的 Plan Validator
        plan = await self.llm.generate_plan(state.input)
        validated_plan = await PlanValidator().validate(plan)
        return state.copy(update={"plan": validated_plan})

class ExecutorNode(Node):
    def __init__(self, tool_registry: ToolRegistry):
        self.tool_registry = tool_registry
        self.name = "executor"
    
    async def invoke(self, state: State) -> State:
        # 这里调用我们的 Streamable HTTP Client
        result = await self.tool_registry.call_tool(
            state.plan.current_step.action,
            state.plan.current_step.parameters
        )
        return state.copy(update={"current_result": result})

所有 Node 都遵循同一接口，可以自由组合。比如理赔场景的主流程是 PlannerNode → ExecutorNode → ValidatorNode → RouterNode ，而客服场景可能是 PlannerNode → IntentClassifierNode → ExecutorNode 。这种组合方式比 LangGraph 的 add_node() 更灵活——你不需要提前定义整个 graph，而是按需拼装。

4.2 如何用 Aegis 实现 ReAct 的“思考-行动”闭环

ReAct（Reasoning + Acting）常被误解为“让 LLM 输出 Thought/Action/Observation 三段式文本”。但真正的 ReAct 是一个 状态机闭环 ：Agent 在每个 step 必须明确回答三个问题：1）我当前在想什么（Reasoning）？2）我要做什么（Acting）？3）我得到了什么反馈（Observation）？Aegis 用 State Graph 实现了这个闭环，且每个环节都可审计。

Step 1：Reasoning 阶段 —— 强制输出结构化思维链
我们不接受 LLM 自由发挥的 “Thought: 我需要查保单”。而是定义严格 Schema：

{
  "reasoning": {
    "current_goal": "获取用户张三的最新有效保单",
    "known_facts": ["用户ID: zhangsan", "当前时间: 2024-06-15"],
    "missing_info": ["保单类型（车险/寿险）", "是否限定最近一年"],
    "next_action": "ask_clarification"
  }
}

PlannerNode 的 invoke 方法会校验这个 JSON 是否包含所有字段，缺失则自动触发澄清。

Step 2：Acting 阶段 —— 工具调用即状态变更
当 next_action 是 call_tool 时，ExecutorNode 不是简单发请求，而是先更新 State：

state = state.copy(update={
    "execution_log": state.execution_log + [{
        "step_id": len(state.execution_log) + 1,
        "action": "call_tool",
        "tool_name": "crm_search",
        "start_time": datetime.now().isoformat()
    }]
})

这样每条日志都自带时间戳和上下文，运维查问题时，直接 redis-cli --scan --pattern "task:*" | xargs -I{} redis-cli HGET {} execution_log 就能拿到完整执行链。

Step 3：Observation 阶段 —— 响应即新输入
工具返回结果后，RouterNode 不是直接喂给 LLM，而是先做三件事：

用 ObservationValidator 校验响应格式（比如 CRM 返回必须含 policy_list 字段）；
用 ObservationSanitizer 过滤敏感字段（比如自动脱敏身份证号）；
将清洗后的结果注入 State 的 observation 字段，作为下一步 Reasoning 的输入。

这个闭环让 ReAct 不再是 LLM 的“表演”，而是可追踪、可干预、可优化的工程流程。当某次理赔查询失败时，我们能精确定位到：是 Reasoning 阶段漏判了“用户可能有多个保单”，还是 Observation 阶段 ObservationSanitizer 错误过滤了关键字段。

5. 从零手搓框架的 7 个血泪教训与避坑清单

手搓框架不是为了证明技术实力，而是为了在真实战场中活下来。以下是我们在 Aegis 开发过程中，用真金白银（和无数个加班夜）换来的 7 条教训，每一条都对应一个曾让我们彻夜难眠的线上故障。

5.1 教训一：永远不要信任 LLM 的 JSON 输出格式

故障现象 ：PlannerNode 随机崩溃，日志显示 json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes 。
根因分析 ：Claude 在高温（CPU >90℃）环境下，偶尔会输出单引号 JSON： {'steps': [{'action': 'query_crm'}]} 。Python json.loads() 严格要求双引号，直接报错。
解决方案 ：在所有 LLM 输出解析前，加一层 json5.loads() （JSON5 支持单引号、注释、尾逗号），并记录告警日志：

try:
    plan = json.loads(llm_output)
except json.JSONDecodeError:
    # 降级到 JSON5 解析
    import json5
    plan = json5.loads(llm_output)
    logger.warning("LLM output used single quotes, parsed via json5", 
                   extra={"raw_output": llm_output[:100]})

提示：别用正则替换单引号——LLM 可能在字符串值里用单引号，比如 "message": "It's OK" ，正则一替全乱。

5.2 教训二：工具超时时间必须小于 LLM 调用超时

故障现象 ：用户投诉“Agent 卡住不动”，监控显示 LLM 调用耗时 120s，但所有工具调用都在 15s 内返回。
根因分析 ：我们给 ChatAnthropic 设置了 timeout=120 ，但工具超时设为 15 。当工具返回慢（比如 OCR 服务偶发 18s），Execute Engine 熔断后返回空结果，PlannerNode 收到空输入，反复重试 LLM，直到 LLM 自身超时。
解决方案 ：建立超时层级树——LLM 超时 = 工具超时 × 最大重试次数 × 1.5。比如工具超时 15s，允许重试 2 次，则 LLM 超时设为 15 * 2 * 1.5 = 45s 。Aegis 的 TimeoutManager 会动态计算并注入。

5.3 教训三：Redis 连接池不是越大越好

故障现象 ：高并发时 Agent 响应延迟飙升，Redis CPU 使用率 100%，但 INFO clients 显示连接数只有 200。
根因分析 ：我们为每个 Task State 创建独立 Redis 连接，峰值 5000 并发时创建了 5000 个连接，远超 Redis 默认 maxclients=10000 ，触发连接拒绝。
解决方案 ：全局复用 aioredis.Redis.from_url() 实例，用 connection_pool 参数控制大小。经压测， minsize=10, maxsize=50 在 5000 QPS 下最稳——连接复用率 99.2%，平均延迟 8ms。

5.4 教训四：MCP 工具的 `list_tools` 响应必须幂等

故障现象 ：Agent 启动后随机丢失工具，日志显示 load_mcp_tools 返回空列表。
根因分析 ：我们的 weather.py 工具在 list_tools 请求时，会动态加载 config.yaml ，而 config 文件偶尔被运维修改，导致加载失败。MCP 规范要求 list_tools 必须快速、可靠、无副作用。
解决方案 ： list_tools 响应必须静态化。Aegis 的 MCP 工具模板强制要求：

# tools/weather.py
TOOLS = [
    {
        "name": "get_forecast",
        "description": "Get weather forecast for a location",
        "input_schema": {"type": "object", "properties": {"lat": {"type": "number"}, "lon": {"type": "number"}}}
    }
]

def list_tools():
    return TOOLS  # 直接返回静态列表，绝不 IO

5.5 教训五：不要在 State 里存大对象

故障现象 ：Task State 的 Redis 内存暴涨，单个 key 达到 2MB， HGETALL 命令阻塞主线程。
根因分析 ：我们把 OCR 识别的完整 PDF 文字（含 10 万字）存进了 state.ocr_result 。Redis 是单线程，大 key 读写会阻塞所有请求。
解决方案 ：State 只存元数据，大对象存对象存储。Aegis 的 LargeObjectStore 会自动：

检测字段值 >100KB 时，上传到 MinIO，返回 s3://bucket/task-123/ocr_result.txt ；
在 State 中只存这个 URL；
提供 state.get_large_field("ocr_result") 方法透明读取。

5.6 教训六：Plan 的 `dependencies` 字段必须做拓扑排序

故障现象 ：Agent 执行顺序错乱，比如先调用 send_email ，再调用 generate_report ，导致邮件里没有报告内容。
根因分析 ：我们定义了 Plan：

{
  "steps": [
    {"step_id": "1", "action": "generate_report", "dependencies": []},
    {"step_id": "2", "action": "send_email", "dependencies": ["1"]}
  ]
}

但 ExecutorNode 是并发执行所有 dependencies 为空的步骤，没做依赖解析。
解决方案 ：ExecutorNode 启动时，用 networkx.DiGraph 构建依赖图，调用 nx.topological_sort() 获取执行顺序。Aegis 内置 DependencyResolver ，10 行代码搞定：

import networkx as nx
def resolve_order(steps: List[Step]) -> List[str]:
    G = nx.DiGraph()
    for step in steps:
        G.add_node(step.step_id)
        for dep in step.dependencies:
            G.add_edge(dep, step.step_id)
    return list(nx.topological_sort(G))

5.7 教训七：永远为“用户说不”留后路

故障现象 ：用户在 Agent 执行到第 3 步时说“算了，不用了”，Agent 却继续执行完剩余 7 步，最后返回“已完成”，用户一脸懵。
根因分析 ：我们没监听用户输入流的中断信号。Aegis 的 UserInputMonitor 模块会持续监听 WebSocket 消息，一旦检测到新消息含 cancel 、 stop 、 nevermind 等关键词，立即：

向当前执行的工具发送 SIGINT （对 STDIO 工具）或 POST /cancel （对 HTTP 工具）；
将 Task State 的 status 设为 cancelled ；
向用户返回：“已取消任务，当前进度：{step_name}（已完成）”。

这个功能上线后，用户主动取消率从 12% 降到 3.7%，因为大家发现 Agent 真的“听得懂人话”。

6. 为什么你现在就应该开始手搓自己的 Agent 框架

别误会，我说的“手搓框架”不是让你从零写一个 LangChain 替代品。而是用两周时间，基于你当前项目的真实痛点，写一个 500 行以内的、只解决一个具体问题的微型框架。比如：

如果你总被 LangChain 的 max_iterations 参数折磨，就写一个 IterationController 类，根据任务复杂度动态调整重试次数；
如果你每次接入新工具都要重写 Tool 类，就写一个 MCPToolWrapper ，一行代码封装任意 MCP 工具；
如果你搞不清 LangSmith 里哪个 prompt 导致了幻觉，就写一个 PromptDebugger ，自动记录每次 LLM 输入输出和 token 消耗。

我见过最惊艳的手搓框架，是一个刚毕业的实习生写的。他负责给销售团队做竞品分析 Agent，但发现 LangChain 的 RetrievalQA 总是漏掉 PDF 里的表格数据。他没去啃 LangChain 源码，而是写了 320 行 Python：