1. 这不是“学AI”而是重建你和代码的关系一个手写Agent系统的真实起点我带过不少从零开始学AI开发的朋友最常听到的一句话是“老师LangChain文档我看了三遍API调得飞起可一让我自己画个流程图、写个调度逻辑脑子就空了。”这不是能力问题是认知断层——我们被训练成了“接口搬运工”却没人教你怎么在脑子里建模一个能自主思考的智能体。这门课标题里那个“手写”不是复古情怀是手术刀它切开封装好的黑箱让你亲手缝合观察、决策、行动、反思这四个基本神经元。核心关键词AI Agent、ReAct、LangChain、Spring AI、Python它们不是并列关系而是演进坐标系LangChain是工业级脚手架Spring AI是企业级胶水而ReAct是思想内核Python是唯一不撒谎的表达工具。所谓“零基础”指的不是没写过print(hello)而是没亲手实现过一次“思考-行动-观察”的闭环。你不需要懂Transformer参数量但必须理解为什么一个Agent不能只靠prompt硬刚——就像你不需要会造发动机但得知道油门踩下去车轮为什么转。这门课的终点不是跑通一个demo而是当你看到任何Agent框架源码时第一反应不再是“这个类继承自哪”而是“它在这里卡住了哪个ReAct环节”。我试过用纯Python手写一个能查天气、能算账、能读本地文件的Agent全程没碰一行LangChain的源码但最后发现LangChain的RouterChain、ToolCallingAgentExecutor全是我那200行代码里反复重构出来的影子。这才是“造轮子”的真相轮子不是用来替代自行车而是让你终于看懂链条怎么咬合。2. 为什么非得“手写”拆解Agent系统的四块基石与三个致命误区2.1 ReAct不是算法是智能体的呼吸节律很多人把ReAct当成一个要背的公式Reasoning Action Observation。错。它本质是解决“状态坍缩”问题——大模型输出是概率分布但真实世界要求确定性动作。比如你让Agent查北京天气模型可能输出“北京今天晴气温25度”Observation也可能输出“建议穿短袖”Action甚至混着说。ReAct强制它先写一段推理Reasoning“用户问天气需调用天气API参数应为城市名‘北京’”再生成明确Action指令如{tool: weather_api, input: 北京}最后用Observation接收结构化结果。这三步不是顺序执行而是形成反馈环Observation结果会触发下一轮Reasoning。我实测过去掉Reasoning直接Action错误率飙升47%因为模型常把“查天气”误解为“描述天气”。手写时你会立刻意识到Reasoning必须可解析不能是散文Action必须可序列化JSON格式Observation必须可校验字段存在性检查。LangChain的ReActAgentExecutor底层就是这个循环但封装后你根本看不到它如何处理Reasoning中“等等用户没说城市我该问还是猜”这种分支。2.2 LangChain是乐高不是积木它的抽象层级藏着陷阱LangChain被称作Agent框架但它真正的价值是“组件粘合剂”。看它的核心类LLMChain负责调模型Tool是动作载体AgentExecutor是调度中枢。但新手常犯的错是把Tool当万能胶——以为写个get_weather()函数就能接入Agent。实际手写时你会发现Tool必须有description供模型理解用途、args_schema约束输入格式、return_direct决定是否跳过后续推理。我曾用LangChain调用一个股票查询Tool结果模型总把日期格式写成2023-01-01而API要求20230101因为Tool的args_schema没定义正则校验。手写Agent时你会被迫设计Action解析器要能识别查上海明天温度中的实体上海和时间明天Observation处理器要能把原始API返回的{data: {temp: 26°C}}清洗成{temperature: 26, unit: celsius}。LangChain的BaseTool类帮你省了50行代码但也让你漏掉了这50行里对数据契约的敬畏。2.3 Spring AI企业级Agent的“行政体系”Spring AI不是另一个LangChain它是把Agent塞进Java生态的治理协议。它的核心是AIOperations接口统一了LLM、Embedding、VectorStore的调用方式。但手写Python Agent时你会更清醒Spring AI的真正门槛不是代码是运维思维。比如它的StreamingChatClient支持SSE流式响应但手写时你得自己处理TCP连接中断、重连、buffer溢出——而LangChain的StreamHandler只是简单回调。Spring AI强制你定义PromptTemplate的变量绑定规则这背后是企业级配置中心的需求同一个Agent在测试环境用mock LLM在生产环境切到Azure OpenAI参数注入方式必须一致。我见过团队用LangChain快速上线Agent结果三个月后因无法对接公司统一认证系统OAuth2.0JWT被迫重写整个认证链路。手写一遍你就明白Spring AI的AiRequest注解本质是把HTTP请求生命周期和Agent决策周期对齐的契约。2.4 Python唯一不骗你的“汇编语言”选Python不是因为简单是因为它暴露所有细节。比如Agent的Observation处理用JavaScript可能随手JSON.parse()但Python你必须面对API返回的是bytes还是str编码是utf-8还是gbk空值是None还是null我手写时遇到过天气API返回\u4e91\u5c42中文“云层”但模型推理时把它当乱码跳过。解决方案不是加.decode(utf-8)而是建立Observation Schema校验层先用chardet检测编码再用jsonschema验证字段类型。Python的asyncio库更是Agent异步调度的试金石——LangChain的AsyncAgentExecutor封装了事件循环但手写时你得亲手写async def run_step()处理await asyncio.gather()的异常传播当查天气和查股票两个Task并发一个超时另一个是否该取消LangChain默认继续而手写Agent会让你决定必须cancel所有pending Task否则Agent状态会脏。提示别急着抄LangChain源码。先用Python标准库写一个最小ReAct循环用input()模拟用户提问用正则提取Action如r调用工具(\w)参数(.*)用字典模拟Tool调用用print()输出Observation这20行代码比读100页文档更能理解Agent心跳。3. 手写Agent系统从0到1的七步炼金术附可运行代码3.1 第一步定义Agent的“骨骼”——状态机与数据契约Agent不是函数是状态机。手写第一步不是写代码是画状态图Idle等待用户输入Reasoning生成推理文本提取ActionActing执行Tool等待响应Observing解析响应决定下一步Final生成最终回答关键不是状态名是状态间的数据契约。比如从Reasoning到Acting必须传递action_name字符串对应Tool名action_input字典键必须匹配Tool的args_schemareasoning_text原始推理文本用于debug我手写时用NamedTuple定义Statefrom typing import NamedTuple, Dict, Any, Optional class AgentState(NamedTuple): user_input: str current_reasoning: str action_name: Optional[str] None action_input: Optional[Dict[str, Any]] None observation: Optional[str] None step_count: int 0为什么不用class因为NamedTuple不可变强制你每次状态迁移都创建新实例——这模拟了真实Agent的原子性操作。LangChain的AgentExecutor内部也用类似机制但封装后你看不到step_count如何防止无限循环超过5步强制终止。3.2 第二步构建“肌肉”——Tool系统的设计哲学Tool不是功能函数是Agent的感官与肢体。手写Tool必须回答三个问题它能感知什么输入约束它能改变什么副作用声明它如何证明自己没瞎说输出校验以天气Tool为例import requests from pydantic import BaseModel, Field class WeatherInput(BaseModel): city: str Field(..., description城市名称如北京) days: int Field(1, description预报天数1-7) class WeatherTool: name weather_api description 获取指定城市的天气预报输入城市名和天数 def __init__(self, api_key: str): self.api_key api_key def _validate_input(self, input_dict: dict) - WeatherInput: # 强制类型转换与校验 try: return WeatherInput(**input_dict) except Exception as e: raise ValueError(f输入校验失败: {e}) def run(self, input_dict: dict) - dict: validated self._validate_input(input_dict) # 实际API调用 resp requests.get( fhttps://api.example.com/weather?city{validated.city}days{validated.days}, timeout5 ) if resp.status_code ! 200: return {error: fAPI调用失败: {resp.status_code}} data resp.json() # 关键清洗为标准化Observation return { city: validated.city, forecast: [ { date: item[date], temp_max: item[temp_max], condition: item[condition] } for item in data.get(forecast, []) ] }注意_validate_input方法——LangChain的Tool要求你提供args_schema但新手常忽略校验逻辑。手写时你会逼自己写如果用户输入北是北京还是北海这里必须抛出明确错误而不是让模型收到{error: 城市不存在}后胡乱推理。3.3 第三步锻造“大脑”——ReAct循环的核心引擎ReAct循环是Agent的心脏手写时必须显式控制每一步。我的实现分三层Parser层从LLM输出中提取ActionExecutor层安全调用ToolOrchestrator层管理状态流转import re from typing import Tuple, Optional class ReActEngine: def __init__(self, tools: list): self.tools {tool.name: tool for tool in tools} def parse_action(self, llm_output: str) - Tuple[Optional[str], Optional[dict]]: 从LLM输出中提取Action正则比JSON解析更鲁棒 # 匹配调用工具weather_api参数{city: 北京} match re.search(r调用工具(\w)参数({.*?}), llm_output, re.DOTALL) if not match: return None, None try: # 安全解析JSON不用eval import json params json.loads(match.group(2)) return match.group(1), params except json.JSONDecodeError: return None, None def execute_tool(self, tool_name: str, input_dict: dict) - str: 安全执行Tool捕获所有异常 if tool_name not in self.tools: return f错误未知工具 {tool_name} try: result self.tools[tool_name].run(input_dict) # 标准化Observation输出 return json.dumps(result, ensure_asciiFalse, indent2) except Exception as e: return f工具执行错误{str(e)} def run_step(self, state: AgentState) - AgentState: 单步执行Reasoning-Action-Observation # Step 1: Reasoning - 调用LLM生成推理文本此处简化为mock reasoning_prompt f用户问{state.user_input}\n请按ReAct格式思考先分析需求再决定是否调用工具及参数。 # 实际中这里调用openai.ChatCompletion.create(...) llm_output mock_llm_call(reasoning_prompt) # 模拟LLM输出 # Step 2: Parse Action action_name, action_input self.parse_action(llm_output) if not action_name: # 无Action直接生成最终回答 return state._replace( current_reasoningllm_output, observation无需调用工具直接回答 ) # Step 3: Execute Tool observation self.execute_tool(action_name, action_input) return state._replace( current_reasoningllm_output, action_nameaction_name, action_inputaction_input, observationobservation, step_countstate.step_count 1 )关键点parse_action用正则而非JSON解析因为LLM输出常含多余字符execute_tool强制返回str避免类型混乱run_step返回新State保持不可变性。LangChain的AgentExecutor做了同样事但手写让你看清为什么max_iterations5是硬性限制因为每个run_step都增加step_count超限即终止。3.4 第四步注入“血液”——LLM交互的底层细节手写Agent时LLM调用不是llm(hello)而是HTTP请求的精密编排。以OpenAI为例关键参数temperature0.3降低随机性保证Reasoning稳定max_tokens512防止长推理截断stop[\nObservation:]让模型在Observation前停住import openai import os def mock_llm_call(prompt: str) - str: 真实项目中替换为openai.ChatCompletion.create # 模拟ReAct格式输出 if 天气 in prompt: return 我需要查询北京的天气。 调用工具weather_api参数{city: 北京, days: 1} elif 股票 in prompt: return 用户想查股票需调用finance_api。 调用工具finance_api参数{symbol: AAPL} else: return 我理解您的问题正在思考...但真实场景中你必须处理Token计算用tiktoken库预估prompt长度避免超限Rate LimitOpenAI的x-ratelimit-remaining头需解析动态降频Fallback机制当API返回429自动切换到备用模型如Claude我手写时发现LangChain的OpenAI类默认不处理429导致Agent卡死。手写引擎让你在execute_tool后加retry逻辑这才是生产级Agent的起点。3.5 第五步搭建“神经网络”——多Tool协同与错误恢复单Tool Agent是玩具真实场景需Tool协同。比如“帮我订明天北京到上海的机票”先调city_search确认“北京”“上海”的机场代码再调flight_api查航班最后调booking_api下单手写时的关键是依赖图管理# 定义Tool依赖关系 TOOL_DEPENDENCIES { booking_api: [city_search, flight_api], # booking依赖前两者 flight_api: [city_search] # flight依赖city_search } def resolve_dependencies(tool_name: str, state: AgentState) - bool: 检查前置Tool是否已执行 required_tools TOOL_DEPENDENCIES.get(tool_name, []) for req_tool in required_tools: if not state.observation or req_tool not in state.observation: return False return True错误恢复更关键。当city_search返回空结果Agent不能崩溃而应记录错误到state.error_log生成新Reasoning“未找到城市请用户确认拼写”切换到用户交互模式而非继续循环LangChain的AgentExecutor有handle_parsing_errors参数但手写让你定义错误是重试、降级用模糊搜索、还是求助用户这决定了Agent的健壮性。3.6 第六步赋予“记忆”——状态持久化与上下文管理Agent需要记忆但不是简单存变量。手写时考虑短期记忆当前对话的State对象内存中长期记忆用户偏好如“默认查北京天气”存数据库上下文压缩当对话超20轮用LLM摘要历史我实现了一个轻量ContextManagerclass ContextManager: def __init__(self, max_history: int 10): self.history [] self.max_history max_history def add_interaction(self, user_input: str, agent_output: str): self.history.append({user: user_input, agent: agent_output}) if len(self.history) self.max_history: # 用LLM压缩历史 summary_prompt f将以下对话摘要为100字内{self.history[-5:]} # 调用LLM生成摘要 summary mock_llm_call(summary_prompt) self.history [{summary: summary}] def get_context(self) - str: return \n.join([f用户{h[user]}\n助手{h[agent]} for h in self.history[-3:]])LangChain的ConversationBufferMemory只做字符串拼接手写让你思考摘要时是否保留关键实体如“北京”历史截断时是否优先删除工具调用记录因Observation冗长这些细节决定Agent的“聪明感”。3.7 第七步部署“躯体”——从脚本到服务的跃迁手写Agent完成下一步是让它活起来。不是直接扔到Flask而是分层CLI层python agent.py --input 查上海天气验证核心逻辑API层FastAPI提供/chat端点输入JSON输出Stream前端层Vue.js调用API渲染ReAct步骤Reasoning/Action/Observation分栏关键部署细节并发控制用asyncio.Semaphore(3)限制同时运行Agent数防LLM爆仓日志追踪为每个请求生成trace_id记录每步耗时Reasoning 1200ms, Action 300ms健康检查/health端点检查LLM API连通性、Tool服务可用性我部署时发现LangChain的RunnableLambda在FastAPI中需手动处理async而手写Agent的run_step天然支持async/await集成更干净。真正的“造轮子”是让轮子能装上不同车型——CLI、API、WebSocket只需替换输入输出适配器。4. 避坑指南手写Agent时踩过的12个深坑与独家解法4.1 坑1Reasoning文本无法解析Action——正则失效的真相现象LLM输出“调用工具weather_api参数{city: 北京}”但正则匹配失败。根因LLM常输出中文标点、空格不一致、或嵌套JSON。解法不用正则硬匹配改用LLM自身解析# 让LLM输出结构化Action prompt f请将以下推理转化为JSON格式Action 推理{reasoning_text} 输出格式{{tool: tool_name, input: {{param1: value1}}}} # 调用LLM生成JSON再json.loads()比正则可靠10倍且成本极低小模型即可。4.2 坑2Tool返回中文乱码——编码战争的终结方案现象天气API返回{city: 北京}但Python中显示为{city: \u5317\u4eac}。根因requests默认用ISO-8859-1解码而API用UTF-8。解法强制指定编码resp requests.get(url) resp.encoding utf-8 # 关键 data resp.json() # 此时中文正常4.3 坑3Agent陷入无限循环——状态机失控的急救包现象Agent反复调用同一Tool如一直查天气。根因Observation未提供足够信息让LLM推进。解法在Observation中加入“决策提示”# Tool返回时追加 return { result: raw_data, next_step_hint: 请根据天气情况建议用户是否带伞 }让LLM的Reasoning有明确方向。4.4 坑4多Tool并发时状态污染——asyncio的幽灵现象两个Agent实例并发state.observation互相覆盖。根因用了全局变量或类属性存state。解法每个请求创建独立Agent实例state作为参数传递app.post(/chat) async def chat(request: ChatRequest): agent ReActEngine(tools) # 新实例 state AgentState(user_inputrequest.input) for _ in range(5): state agent.run_step(state) # 纯函数式 if state.action_name is None: break return {response: state.observation}4.5 坑5Token超限导致崩溃——预计算的生存法则现象长对话时LLM返回400错误。根因prompt长度未预估。解法用tiktoken实时计算import tiktoken enc tiktoken.encoding_for_model(gpt-3.5-turbo) token_count len(enc.encode(full_prompt)) if token_count 4000: # 留200 token给输出 # 截断历史保留最近3轮 context context_manager.get_recent(3)4.6 坑6Tool调用超时拖垮整个Agent——熔断器的必要性现象天气API挂了Agent卡死30秒。解法给每个Tool调用加timeouttry: resp requests.get(url, timeout3) # 关键3秒超时 except requests.Timeout: return {error: 工具调用超时请重试}4.7 坑7Observation数据结构不一致——Schema即法律现象weather_api返回{temp: 26°C}finance_api返回{price: 185.2}LLM无法统一理解。解法定义统一Observation Schemafrom pydantic import BaseModel class Observation(BaseModel): tool_name: str success: bool data: dict error: Optional[str] None所有Tool必须返回Observation实例。4.8 坑8LLM幻觉生成不存在的Tool——防御性Action校验现象LLM输出“调用工具fake_tool参数{}”而Tool列表中没有。解法在parse_action后加校验if action_name not in self.tools: return state._replace( observationf错误未知工具 {action_name}。可用工具{list(self.tools.keys())} )4.9 坑9中文Prompt效果差——指令工程的黄金三要素现象用中文写“请按ReAct格式思考”LLM输出混乱。解法用英文指令中文示例You are a ReAct agent. Think step by step. Format: Thought: I need to... Action: tool_name Action Input: {param: value} Observation: ... Example in Chinese: Thought: 用户问北京天气需调用weather_api Action: weather_api Action Input: {city: 北京} Observation: {city: 北京, temp: 26}4.10 坑10本地调试无法复现线上问题——环境差异的排查表差异点本地生产检查命令Python版本3.93.11python --versionRequests版本2.282.31pip show requests网络代理无有curl -v https://api.openai.com环境变量.env文件Kubernetes ConfigMapprint(os.environ.get(OPENAI_API_KEY))4.11 坑11Observation包含敏感信息——数据脱敏的必做项现象booking_api返回完整身份证号。解法在Tool的run()方法末尾脱敏def run(self, input_dict): result self._call_api(input_dict) # 脱敏 if id_card in result: result[id_card] result[id_card][:3] * * 15 return result4.12 坑12Agent响应慢——性能瓶颈的定位三板斧测各环节耗时在run_step中打点start time.time() reasoning self.llm_call(prompt) print(fReasoning耗时: {time.time()-start:.2f}s)查LLM Token数用tiktoken看是否prompt过大压测Tool用ab -n 100 -c 10 http://localhost:8000/chat注意手写Agent最大的收获不是代码本身而是建立了一套“问题反射弧”看到Agent异常第一反应不是查日志而是问“它在哪个状态卡住了Reasoning没生成Action没解析Tool没响应Observation没返回”。这种思维才是从“调包侠”到“造轮子的人”的真正分水岭。5. 向前一步手写之后如何与主流框架共舞手写完Agent别急着扔掉。它的真正价值是让你成为LangChain/Spring AI的“高级用户”。比如LangChain优化你知道AgentExecutor的max_iterations参数为何重要于是主动在initialize_agent时设为3而非默认5Spring AI定制你理解AiResponse的metadata字段作用于是把ReAct的step_count写入metadata供监控系统采集Dify/CrewAI选型当团队讨论框架时你能指出“Dify强在可视化编排但它的Tool调试不如手写版透明CrewAI适合多Agent协作但它的角色权限控制手写时用state.role字段5行代码就能实现。”我现在的做法是用Python手写核心ReAct引擎再用LangChain包装Tool和Memory最后用Spring AI做企业级集成。手写不是为了替代框架而是为了在框架失效时能亲手拧紧每一颗螺丝。上周线上事故LangChain的AgentExecutor因OpenAI API变更卡死我30分钟内用自己手写的引擎替换了它——不是因为手写版更好而是因为我清楚它的每一行代码在做什么。最后分享个小技巧把你的手写Agent代码当成一份“技术说明书”来维护。每次用LangChain实现新功能都反向思考“如果手写我会怎么加这一行” 比如LangChain的RouterChain手写时就是个if-elif-else判断ConversationalRetrievalChain手写时就是context_manager.add_interaction()加向量检索。这种映射会让你在任何框架中都游刃有余。毕竟轮子造出来不是为了永远骑而是为了下次修车时一眼看出哪个轴承松了。