手写AI Agent系统:从零构建可调试的执行引擎、状态中枢与路由大脑 1. 这门课到底在教什么不是写AI是在重建你和代码的关系“零基础手写AI Agent系统课从‘调包侠’到‘造轮子的人’”——这个标题里藏着三个被多数人忽略的关键词零基础、手写、Agent系统。它不是又一门教你用LangChain搭个聊天机器人、再加个RAG插件就结课的速成班也不是面向已有工程经验者讲LLM推理优化或分布式调度的高阶课。它瞄准的是一个真实存在的断层带大量刚学完Python基础、能跑通HuggingFace示例代码、却连__init__.py文件为什么存在都说不清的学习者。他们卡在“知道API怎么调但不知道API为什么这样设计”的临界点上。我带过不少这类学员最典型的反馈是“我照着文档把Agent跑起来了可一旦模型返回格式错了一位、工具调用多嵌套了一层、或者用户突然问了个没预设路径的问题整个流程就崩了连日志都看不懂从哪开始查。”这门课的底层逻辑是把AI Agent拆解成一套可触摸、可调试、可打断的最小可运行单元链。它不从“智能体应该有记忆、规划、工具调用”这些抽象概念出发而是从一行while True:循环开始让你亲手写出第一个永远不退出的交互主循环从一个空字典{}开始逐步给它加上last_message、tool_calls、execution_history字段理解状态为何必须显式管理从手动拼接一段system_prompt user_input字符串开始体会提示词不是魔法咒语而是需要被结构化、版本化、可回溯的程序输入。所谓“手写”不是拒绝使用现有库而是先亲手实现SimpleRouter、MockToolExecutor、InMemoryStateStore这三个核心组件再对比LangChain的RouterChain、ToolExecutor、ConversationBufferMemory看它们在哪些边界做了取舍——比如LangChain默认把工具调用结果直接塞进消息历史而手写版会让你明确看到“执行结果→解析→格式校验→注入上下文”这四步缺一不可。适合谁学第一类是自学半年以上、能独立完成Kaggle入门项目但总感觉“代码像借来的”、不敢改一行源码的实践者第二类是转行进入AI应用层、被要求快速交付POC却常因底层逻辑模糊导致返工的工程师第三类是高校教学一线教师想为学生设计真正能建立系统思维的实验课。它不承诺“学完就能年薪百万”但能保证当你第三次手动重写parse_tool_call函数时你会突然明白为什么OpenAI的Function Calling规范强制要求name字段必须是字符串而非对象——因为序列化与反序列化之间隔着一层人类容易忽略的类型契约。2. 为什么必须“手写”拆解Agent系统的三层物理结构很多人误以为Agent就是“大模型几个工具”就像组装乐高。但实际落地时90%的故障不出现在模型侧而出现在连接层。这门课的手写过程本质是把Agent系统按物理职责划分为三层并逐层构建2.1 第一层执行引擎Execution Engine——让AI“动起来”的肌肉这是最易被忽视的底层。所谓“让大模型思考”背后是一整套同步/异步调度、超时控制、重试策略、错误隔离的机制。课程不直接用asyncio.gather()而是从最朴素的for tool in tools:循环开始手动实现工具调用的串行阻塞执行理解为什么不能并行调用有状态工具执行结果的JSON Schema校验用jsonschema.validate()而非try...except json.loads()超时熔断当某个工具执行超过3秒主动终止并返回{error: timeout}而非让整个Agent挂起我曾让学员对比两种写法一种是直接response llm.invoke(prompt)另一种是封装成execute_step(prompt, timeout5.0, max_retries2)。后者看似多此一举但当某次API返回503 Service Unavailable时前者会直接抛出未捕获异常导致进程退出后者则自动重试并记录retry_count1到状态中。这就是“手写”带来的确定性——你清楚每一毫秒CPU在做什么而不是依赖框架的黑盒兜底。2.2 第二层状态中枢State Hub——Agent的“短期记忆”Agent没有真正的记忆只有状态快照。课程用一个60行的class AgentState类替代所有现成Memory组件强制暴露三个关键设计决策状态不可变性每次更新必须new_state old_state.copy().update(...)避免隐式引用导致的脏数据字段强约束messages: List[Dict[str, Any]]而非messages: Any用pydantic.BaseModel定义每个message必须含role、content、timestamp版本追踪每个state自带version: int和created_at: datetime当发现state.version expected_version时主动拒绝执行防止并发修改冲突实操中学员常犯的错误是把用户原始输入直接存入messages结果当用户发来scriptalert(1)/script时前端渲染直接XSS。手写状态层会让你在add_message()方法里插入html.escape(content)这种防御性编程意识是调包永远学不会的肌肉记忆。2.3 第三层路由大脑Routing Brain——决定“下一步做什么”的逻辑这才是真正体现“智能”的部分却常被简化为if 天气 in query: call_weather_tool()。课程要求手写一个支持多级条件的RuleBasedRouterclass RuleBasedRouter: def __init__(self): self.rules [ (lambda s: 订餐 in s.last_user_message and 外卖 not in s.last_user_message, lambda s: s.set_next_action(suggest_restaurant)), (lambda s: len(s.tool_calls) 0 and s.tool_calls[-1][status] failed, lambda s: s.set_next_action(retry_last_tool)) ] def route(self, state: AgentState) - str: for condition, action in self.rules: if condition(state): action(state) return state.next_action return default_think重点不在代码本身而在于迫使你思考规则顺序是否重要失败重试时要不要清空之前的tool_calls当多个规则同时匹配该用first-match还是best-match这些细节在LangChain的RouterChain里被封装成priority参数但手写过程让你直面设计权衡——比如把重试规则放在最后是因为我们认定“工具失败”是异常路径不应干扰正常业务流。提示手写不是为了替代框架而是为了获得“框架选择权”。当你亲手实现过内存泄漏的InMemoryStateStore下次选Redis还是PostgreSQL做持久化时你会自然关注连接池大小、序列化开销、事务隔离级别而不是只看文档里“支持分布式”三个字。3. 核心模块手写实录从0到1构建可调试Agent课程的核心实操环节是用不到300行纯Python代码构建一个具备完整交互闭环的Agent。它不追求功能炫酷但每个环节都预留了调试入口。以下是最关键的三个模块实现细节附带我在教学中发现的高频卡点。3.1 消息编排器Message Orchestrator提示词不是文本是协议很多学员以为提示词就是“把需求写得更清楚”但手写过程揭示其本质是人机通信协议。课程定义MessageOrchestrator类强制分离三类消息System Message固定模板含角色定义、能力边界、安全守则如“禁止生成违法信息”History Messages从state.messages截取最近5条但需过滤掉role tool的中间结果避免污染上下文User Message对原始输入做标准化处理——移除首尾空格、替换连续换行为单个\n、截断超长文本2000字符最关键的一步是消息注入时机控制。课程不采用“把所有历史拼一起扔给模型”的粗暴方式而是实现inject_context()方法def inject_context(self, state: AgentState) - List[Dict]: messages [] # 1. 系统消息永远在最前 messages.append({role: system, content: self.system_prompt}) # 2. 历史消息按时间倒序但跳过tool类型 for msg in reversed(state.messages[-5:]): if msg[role] ! tool: messages.append(msg) # 3. 用户新消息在最后 messages.append({role: user, content: state.last_user_message}) return messages这个设计解决了真实场景中的经典问题当用户连续追问“刚才说的餐厅人均多少”模型需要关联前文但如果把tool类型的消息如{role: tool, name: get_restaurant_info, content: {price_range: ¥¥}}也塞进上下文模型可能误以为这是用户提问。手写过程让你看清提示词工程的本质是构造一个能让模型稳定复现推理路径的输入空间。3.2 工具执行沙箱Tool Execution Sandbox让外部调用变得可预测学员最常崩溃的环节是工具调用失败。课程不直接用requests.get()而是构建ToolSandboxclass ToolSandbox: def __init__(self, tools: Dict[str, Callable]): self.tools tools self.execution_log [] # 记录每次调用的输入/输出/耗时 def execute(self, tool_name: str, args: Dict) - Dict: start_time time.time() try: if tool_name not in self.tools: raise ValueError(fTool {tool_name} not registered) result self.tools[tool_name](**args) duration time.time() - start_time self.execution_log.append({ tool: tool_name, input: args, output: result, duration: duration, status: success }) return {result: result, status: success} except Exception as e: duration time.time() - start_time self.execution_log.append({ tool: tool_name, input: args, error: str(e), duration: duration, status: error }) return {error: str(e), status: error}这个沙箱带来三个确定性可追溯execution_log直接暴露哪次调用超时、哪个参数触发了异常可模拟测试时可注入mock_tools {weather: lambda city: {temp: 25}}无需启动真实API可审计当发现工具返回{temp: 25°C}字符串而模型期望数字时日志里清晰显示input: {city: Beijing}→output: {temp: 25°C}问题定位从小时级缩短到分钟级注意沙箱必须显式记录args而非*args因为真实工具常有api_key等敏感参数手写过程强制你思考“哪些参数该进日志哪些该打码”这是生产环境必备的安全意识。3.3 状态演化器State EvolverAgent的“成长档案”最终所有模块要汇入一个evolve_state()函数它定义Agent如何从旧状态走向新状态。课程要求该函数必须是纯函数无副作用接收old_state和step_result返回new_statedef evolve_state(old_state: AgentState, step_result: Dict) - AgentState: new_state old_state.copy() new_state.version 1 new_state.updated_at datetime.now() # 根据step_result类型更新不同字段 if step_result.get(type) llm_output: new_state.messages.append({ role: assistant, content: step_result[text], tool_calls: step_result.get(tool_calls, []) }) elif step_result.get(type) tool_result: new_state.messages.append({ role: tool, name: step_result[tool_name], content: json.dumps(step_result[result]) }) return new_state这个设计逼你回答根本问题Agent的“智能”体现在哪里不是模型多强大而是evolve_state能否在llm_output含tool_calls时正确生成tool角色消息能否在tool_result返回后把结果精准注入到对应tool_calls的上下文中。当学员第一次手动实现这个函数并用print(new_state.messages[-3:])观察状态变化时那种“原来如此”的顿悟感是任何调包教程无法给予的。4. 从“能跑”到“稳跑”手写Agent的四大避坑实战清单手写过程最大的价值不是写出完美代码而是踩够足够多的坑。以下是学员在实操中反复验证的四大高频陷阱附带我的现场调试记录和根因分析。4.1 陷阱一消息ID混乱导致上下文错位发生率87%现象Agent在多轮对话中突然“失忆”比如用户问“北京天气如何”Agent调用天气工具后返回“北京今天25度”用户再问“那上海呢”Agent却重复回答北京天气。调试过程我在evolve_state()里加了print(fStep {new_state.version}: {len(new_state.messages)} msgs)发现版本号跳跃增长1→3→5说明某些步骤被跳过。进一步检查MessageOrchestrator.inject_context()发现它从state.messages[-5:]取历史但当tool消息被错误地加入messages时列表长度暴增导致取到的“最近5条”包含大量无关的tool消息挤掉了真正的用户提问。根因tool消息不该直接append到messages而应存入独立的tool_results字段inject_context()只从messages取user/assistant消息。手写强制你设计消息分类存储而调包框架常把所有消息混存。解决方案重构AgentState增加tool_results: List[Dict]字段并在evolve_state()中严格区分消息类型存储。4.2 陷阱二工具参数类型失配引发静默失败发生率73%现象天气工具明明注册了get_weather(city: str)但用户说“查深圳天气”Agent却返回{error: city parameter missing}。调试过程在ToolSandbox.execute()里打印args发现传入的是{city: [Shenzhen]}列表而非字符串。追查parse_tool_call()函数发现它用正则提取city: Shenzhen时因用户输入含中文引号“深圳”正则rcity:\s*([^]*)匹配失败退而求其次返回空字典最终**{}调用导致参数缺失。根因JSON解析不能依赖正则必须用标准json.loads()。但用户输入非标准JSON时如中文引号、尾随逗号json.loads()会报错。手写过程逼你实现容错解析先尝试json.loads()失败则用ast.literal_eval()再失败才用正则兜底。解决方案重写parse_tool_call()增加三层解析策略并在日志中记录每次解析使用的策略便于问题溯源。4.3 陷阱三状态突变导致无限循环发生率61%现象Agent陷入死循环不断调用同一个工具如反复查询天气却不返回结果。调试过程在RuleBasedRouter.route()里加print(fRouting with state version {state.version})发现版本号卡在某个值不动。检查evolve_state()发现当tool_result含特殊字符如{temp: 25°C}中的°时json.dumps()生成的字符串含\u00b0而MessageOrchestrator在拼接消息时未做Unicode处理导致后续json.loads()失败state未更新version不变路由逻辑永远看到旧状态。根因状态序列化/反序列化链路中任意一环的编码不一致都会导致状态“冻结”。手写过程让你在每处json.dumps()后立即跟json.loads()验证确保可逆。解决方案在AgentState的to_dict()方法中强制json.dumps(..., ensure_asciiFalse)并在from_dict()中指定encodingutf-8。4.4 陷阱四超时设置反模式引发雪崩发生率52%现象单个工具超时后整个Agent服务响应延迟飙升甚至OOM。调试过程用psutil.Process().memory_info().rss监控内存发现每次超时后内存增长2MB且不释放。检查ToolSandbox.execute()发现try...except块内创建了大型临时对象如pd.DataFrame但异常时未显式delPython GC未及时回收。根因异步超时如asyncio.wait_for()会中断协程但已分配的内存不会自动清理。手写过程逼你实现资源清理钩子在execute()开头记录初始内存在finally块中计算增量并告警。解决方案引入contextlib.ExitStack管理临时资源并在超时异常中显式调用gc.collect()。陷阱类型触发条件调试关键点手写带来的认知升级消息ID混乱多轮对话工具调用混合state.messages长度与内容一致性消息类型必须物理隔离不能仅靠role字段逻辑区分参数类型失配用户输入含非标准符号args字典的实际内容与预期类型工具接口契约比HTTP状态码更重要必须在入口处强校验状态突变失效Unicode字符JSON序列化state.version是否递增状态演化必须是原子操作任何环节失败都应rollback而非静默超时雪崩高并发内存密集型工具单次执行前后内存占用差值超时不是终点而是资源清理的起点需定义明确的清理契约5. 手写之后如何把“轮子”装回真实项目手写完成不等于结束而是真正开始。很多学员问“我写出了自己的Agent然后呢”我的答案很实在把它当成一个可插拔的诊断探针而非生产组件。以下是三个经过验证的落地路径。5.1 路径一作为现有框架的“健康检查仪”把你的手写Agent部署为独立服务接入LangChain或LlamaIndex项目。当线上Agent出现诡异行为如突然不调用工具、回复变短不急着改提示词而是将相同用户输入、相同工具配置、相同模型参数同步发送给手写Agent和生产Agent对比两者的execution_log如果手写版能正确解析tool_calls而生产版不能说明是框架的OutputParser有bug如果两者都失败问题在模型或提示词我带的一个电商项目就用这招发现LangChain的JsonOutputParser在处理嵌套JSON时会丢失字段而手写版的parse_tool_call()因强制json.loads()校验提前暴露了问题5.2 路径二定制化工具链的“最小验证集”企业常有私有API如内部CRM系统调包框架的通用工具封装无法满足安全审计要求。此时把手写Agent的ToolSandbox模块抽出来重写execute()方法集成公司统一认证SDK如OAuth2.0 Bearer Token注入在execution_log中自动添加审计字段request_id,operator_id用evolve_state()的纯函数特性实现工具调用的离线回放测试 某金融客户用此方案将新工具上线测试周期从3天缩短到2小时因为所有测试用例都基于state快照无需真实调用生产API。5.3 路径三教学演示的“透明黑板”给新人培训时不再展示“一键部署”的幻灯片而是打开你的手写代码实时修改RuleBasedRouter的规则顺序演示为什么把error_recovery规则放最后在MessageOrchestrator里注释掉filter tool messages逻辑让学生亲眼看到上下文污染如何让模型胡言乱语用print(state.messages)逐行展示状态演化比任何架构图都直观 有位高校导师反馈学生第一次看到state.version从1跳到5时集体鼓掌——因为他们终于理解了“状态”不是抽象概念而是可触摸的整数。最后分享一个小技巧每次手写一个新模块立刻用git tag v0.1-message-orchestrator打标签。半年后回看你会惊讶于自己如何从纠结role字段命名进化到设计跨Agent的状态同步协议。这种成长轨迹是任何调包教程无法提供的个人技术年鉴。我在实际带教中发现真正完成全部手写模块的学员三个月后在技术面试中被问到“如何设计一个可扩展的Agent架构”时几乎没人再泛泛而谈“用微服务拆分”而是会掏出手机展示自己写的AgentState类指着version和updated_at字段说“这是我的乐观锁也是我的事件溯源起点。”——这种笃定不是来自对框架的熟悉而是源于亲手锻造过每一个零件的底气。