1. 这门课不是教你怎么“用AI”,而是教你“让AI听你的话”
我带过不下二十个从零开始学AI Agent的学员,最常听到的一句话是:“老师,LangChain文档我看完了,API我也调通了,可一让我自己搭个能真正做事的Agent,比如自动查天气+订会议室+发邮件通知,我就卡在第一步——不知道该从哪下手。”
这不是能力问题,是认知断层。市面上90%的AI课程,本质是“高级调包教程”:告诉你llm.invoke()怎么写、Chain怎么串、Tool怎么注册。但没人告诉你,当invoke()返回一个空字符串时,背后可能是LLM的token截断策略、工具调用的schema校验失败、还是ReAct循环里observation字段被意外清空?更没人告诉你,为什么LangChain的AgentExecutor默认不重试、为什么LangGraph的state更新必须用StateUpdate而不是直接赋值、为什么Spring AI的AiResponse对象里藏着一个你永远用不到却无法绕过的metadata字段。
这门《零基础手写AI Agent系统课》,就是专治这种“调包幻觉”的解药。它不教你怎么快速跑通一个Demo,而是带你亲手把Agent的每一根骨头都拆开、摸透、再重新组装起来。我们不用任何现成的Agent框架——不碰LangChain的create_react_agent,不抄Dify的可视化编排逻辑,不套CrewAI的Role定义模板。我们要做的,是从Python最基础的dict和list开始,一行行写出能完成“搜索→推理→调用→验证→反思”闭环的最小可运行Agent内核。
关键词里的AI Agent,在这里不是名词,是动词;ReAct不是论文标题,是每天要写的三行代码;LangChain不是学习目标,而是你写完自己的Agent后,拿来对比“它为什么比我的设计多了一层抽象”的参照物;Spring AI不是Java生态的替代品,是你理解“不同语言如何表达同一抽象概念”的翻译器;而Python,是我们唯一需要的工具——不是因为它是“最简单”的语言,而是因为它足够透明:没有JVM字节码干扰,没有Rust的所有权系统遮蔽,你能一眼看到__call__方法里到底发生了什么。
这门课适合三类人:
- 刚学完大模型API调用,却不敢接真实需求的开发者——你缺的不是新知识,是把知识焊进肌肉记忆的锤炼过程;
- 被各种Agent框架文档绕晕的产品/运营同学——你不需要写代码,但必须看懂
tool_call和tool_response之间那200毫秒里发生了什么; - 想搞懂“AI Agent到底是不是银弹”的技术决策者——当你亲手实现过ReAct的失败回滚机制,你就知道为什么有些场景必须上LangGraph,而有些场景连
while True:都比框架更可靠。
别急着装环境、配依赖。先问自己一个问题:如果现在让你用纯Python写一个能自主决定“该查天气还是该查航班”的Agent,你的第一行代码会是什么?不是from langchain import ...,而是class AgentState: ...。答案就藏在这门课的第一课里。
2. 为什么“手写”比“调包”更能暴露真实问题
很多人觉得“手写Agent”是复古、是低效、是重复造轮子。我去年帮一家做智能客服的公司做架构评审,他们用LangChain搭的Agent在测试环境准确率92%,上线后跌到63%。运维日志里全是ToolExecutionError,但没人能说清错误是发生在工具调用前(LLM输出格式错)、调用中(HTTP超时)、还是调用后(JSON解析失败)。最后发现,问题出在LangChain默认的Tool类里,args_schema校验把用户输入的中文括号()当成了非法字符——而这个校验逻辑,在官方文档里只有一行小字说明。
这就是“调包”的代价:你永远在框架的抽象层之下,隔着一层看不见的玻璃看世界。而手写,就是亲手砸碎这块玻璃。
我们这门课的手写,不是从零发明轮子,而是精准复刻主流框架的核心契约。比如ReAct模式,它的灵魂不在“Thought/Action/Observation”这三个词,而在以下四个不可妥协的约束:
- 状态不可变性:每次循环必须生成全新state,旧state只能读不能改;
- 动作原子性:一个Action必须对应且仅对应一次外部调用(API/DB/Shell);
- 观测隔离性:Observation内容必须严格来自Action执行结果,不能混入LLM内部状态;
- 终止确定性:Agent必须有明确的
finish信号,且该信号只能由LLM在Thought中显式声明。
LangChain的ReActSingleActionAgent满足前两条,但第三条靠tool_input字段隐式传递,第四条依赖Final Answer:硬编码前缀——这导致当LLM输出I'm done.时,Agent直接崩溃。而我们手写的版本,会强制要求LLM返回结构化JSON:
{ "thought": "我已获取到所有必要信息", "action": "FINISH", "action_input": {"result": "会议已预订成功"} }这个设计看似多此一举,但它解决了三个真实痛点:
- 调试友好:日志里每行都是合法JSON,grep就能定位失败环节;
- 协议清晰:前端调用方无需解析自然语言,直接取
action_input.result; - 扩展安全:新增
WAIT_FOR_CONFIRMATION动作时,无需修改解析逻辑。
再比如工具调用(Tool Calling)。LangChain用BaseTool抽象,Spring AI用AiRequest封装,但它们都回避了一个根本问题:工具返回的原始数据,如何与LLM的语义空间对齐?我们手写时会专门设计ToolAdapter层:
- 当天气API返回
{"temp": 25.3, "unit": "celsius"},Adapter必须转成"当前温度25摄氏度"; - 当数据库查询返回
[{"id": 1, "name": "张三"}],Adapter必须转成"找到1位员工:张三"; - 当Shell命令返回
b'ls: cannot access /tmp: Permission denied\n',Adapter必须识别这是错误并转成"权限不足,无法访问/tmp目录"。
这个转换层,LangChain叫format_to_openai_tool,Spring AI叫ToolResponseMapper,但它们都藏在框架深处。而我们手写时,会把它暴露为独立模块,让学员亲手写10个Adapter——直到他们明白:Agent的智能,70%不在LLM里,而在这些把混乱现实映射成干净语义的胶水代码中。
提示:手写过程中最大的陷阱,是过早引入异步。很多学员一上来就想用
asyncio并发调多个工具,结果陷入回调地狱。我们坚持同步实现——因为ReAct本质是单线程状态机,强行异步只会掩盖状态管理漏洞。等你手写完5个完整ReAct循环,再加async,才是真懂。
3. 从零构建ReAct Agent:四步走完核心闭环
现在,让我们真正动手。这不是伪代码演示,而是可直接复制粘贴运行的最小可行实现。整个Agent内核控制在200行以内,但覆盖了ReAct全部关键环节。
3.1 第一步:定义不可变状态机(State)
Agent不是一堆函数,而是一个有记忆的状态机。我们拒绝用dict或dataclass,选择NamedTuple——它天然不可变,且IDE能自动补全字段名:
from typing import NamedTuple, Optional, List, Dict, Any class AgentState(NamedTuple): # LLM输入上下文(必须包含历史) messages: List[Dict[str, str]] # 当前思考链(Thought) thought: str # 上次动作(Action) action: str # 动作参数(Action Input) action_input: Dict[str, Any] # 上次观测结果(Observation) observation: str # 是否终止 is_finished: bool # 最终结果 final_result: Optional[str] # 初始化状态 def create_initial_state(user_query: str) -> AgentState: return AgentState( messages=[{"role": "user", "content": user_query}], thought="", action="", action_input={}, observation="", is_finished=False, final_result=None )为什么不用dataclass(frozen=True)?因为NamedTuple序列化更轻量,且_replace()方法语义更清晰——每次状态更新都必须显式调用它,强迫你思考“哪些字段变了”。而dataclass的copy.copy()容易让人误以为在修改原对象。
3.2 第二步:实现LLM推理层(LLM Call)
我们不依赖任何第三方SDK,用requests直连OpenAI API(或本地Ollama),重点在于结构化输出约束:
import requests import json def call_llm(state: AgentState, tools: List[Dict]) -> Dict[str, Any]: # 构建ReAct提示词(精简版) prompt = f""" 你是一个ReAct Agent。请严格按以下格式输出: Thought: <你的思考过程> Action: <工具名,必须是{[t['name'] for t in tools]}之一> Action Input: <JSON格式参数> Observation: <等待你调用工具后填入> Thought: <继续思考> ... Final Answer: <最终结果> 当前任务:{state.messages[-1]['content']} 历史交互: """ for msg in state.messages[-3:]: # 只保留最近3轮 prompt += f"{msg['role'].upper()}: {msg['content']}\n" # 强制LLM输出JSON Schema response_format = { "type": "json_schema", "json_schema": { "name": "react_output", "schema": { "type": "object", "properties": { "thought": {"type": "string"}, "action": {"type": "string"}, "action_input": {"type": "object"} }, "required": ["thought", "action", "action_input"] } } } resp = requests.post( "https://api.openai.com/v1/chat/completions", headers={"Authorization": "Bearer sk-..."}, json={ "model": "gpt-4-turbo", "messages": [{"role": "user", "content": prompt}], "response_format": response_format } ) return resp.json()["choices"][0]["message"]["content"]注意这里的关键设计:
- 历史截断:只传最近3轮消息,避免token爆炸;
- Schema强制:用OpenAI的
response_format确保输出可解析; - 无状态调用:LLM只负责生成,不保存任何中间状态——状态管理完全由Agent自己掌控。
3.3 第三步:工具调度与执行(Tool Execution)
工具不是函数,是带元数据的对象。我们定义统一接口:
from abc import ABC, abstractmethod class Tool(ABC): name: str description: str @abstractmethod def execute(self, **kwargs) -> str: pass class WeatherTool(Tool): name = "get_weather" description = "获取指定城市的天气" def execute(self, city: str) -> str: # 真实调用天气API return f"北京今日晴,气温25摄氏度,空气质量优" class SearchTool(Tool): name = "web_search" description = "搜索互联网信息" def execute(self, query: str) -> str: return f"搜索结果:Python Agent教程推荐《Handson AI Agents》" # 工具注册中心 TOOLS = [WeatherTool(), SearchTool()] TOOL_MAP = {t.name: t for t in TOOLS}执行逻辑极其简单,但关键在错误处理:
def execute_tool(state: AgentState) -> str: try: tool = TOOL_MAP[state.action] result = tool.execute(**state.action_input) return result except KeyError: return f"错误:未找到工具 '{state.action}'" except Exception as e: return f"工具执行失败:{str(e)}"这里没有重试、没有降级、没有熔断——因为ReAct的哲学是:失败即反馈,反馈即学习。LLM看到工具执行失败,下次就会换动作,而不是让Agent自己瞎猜。
3.4 第四步:状态更新与循环控制(State Transition)
这才是Agent的“大脑”。我们用纯函数式风格,每次返回新状态:
def step(state: AgentState) -> AgentState: # 1. 调用LLM获取下一步指令 llm_output = call_llm(state, [t.__dict__ for t in TOOLS]) # 2. 解析LLM输出(这里简化,实际需健壮JSON解析) try: parsed = json.loads(llm_output) new_thought = parsed["thought"] new_action = parsed["action"] new_action_input = parsed["action_input"] except (json.JSONDecodeError, KeyError): # 解析失败,强制终止 return state._replace( is_finished=True, final_result="LLM输出格式错误,无法继续" ) # 3. 如果是FINISH动作,提取结果 if new_action == "FINISH": return state._replace( thought=new_thought, action=new_action, action_input=new_action_input, is_finished=True, final_result=new_action_input.get("result", "") ) # 4. 否则执行工具,获取观测 observation = execute_tool(state._replace( action=new_action, action_input=new_action_input )) # 5. 更新状态:追加本轮交互到messages new_messages = state.messages + [ {"role": "assistant", "content": f"Thought: {new_thought}\nAction: {new_action}\nAction Input: {json.dumps(new_action_input)}"}, {"role": "user", "content": f"Observation: {observation}"} ] return AgentState( messages=new_messages, thought=new_thought, action=new_action, action_input=new_action_input, observation=observation, is_finished=False, final_result=None ) # 主循环 def run_agent(user_query: str, max_steps: int = 5) -> str: state = create_initial_state(user_query) for i in range(max_steps): state = step(state) if state.is_finished: return state.final_result return "超过最大步骤数,未完成任务"这个step()函数,就是ReAct的全部灵魂。它不关心LLM用什么模型,不关心工具在哪部署,只专注一件事:给定当前状态,输出下一个状态。这种纯粹性,正是框架难以提供的。
注意:实际项目中,
max_steps必须设为奇数(如5、7),因为ReAct循环是“LLM输出→工具执行→LLM再输出”三步一组。偶数步会导致Observation没被LLM看到就终止。
4. LangChain vs 手写:一场关于抽象边界的实战对比
现在,我们用同一个需求——“查北京天气并推荐合适穿搭”——分别用LangChain和手写版本实现,看抽象带来的得与失。
4.1 LangChain方案(标准写法)
from langchain.agents import create_react_agent, AgentExecutor from langchain import hub from langchain_core.tools import tool @tool def get_weather(city: str) -> str: return f"{city}今日晴,25℃" @tool def recommend_outfit(temp: float) -> str: if temp > 22: return "建议穿短袖+薄外套" else: return "建议穿长袖+毛衣" tools = [get_weather, recommend_outfit] prompt = hub.pull("hwchase17/react") llm = ChatOpenAI(model="gpt-4-turbo") agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) result = agent_executor.invoke({"input": "查北京天气并推荐穿搭"})表面看,代码少、上手快。但当你遇到问题时,调试路径是这样的:
agent_executor.invoke()→ 进入AgentExecutor._call()- →
agent.invoke()→ 进入RunnableSequence.invoke() - →
prompt | llm | output_parser→ 卡在output_parser的parse()方法 - → 发现LLM返回了
Thought: ... Action: get_weather ... Observation: ... Thought: ...,但output_parser只认Final Answer:前缀
你不得不去翻LangChain源码,找到ReActOutputParser,然后重写它——而这时,你已经写了50行代码来修复一个本不该存在的耦合。
4.2 手写方案(增强版)
我们在基础版上增加两个关键能力:
- 工具链式调用:允许一个工具输出作为另一个工具输入;
- 动态工具注册:运行时添加新工具,无需重启Agent。
# 增强型Tool基类,支持输入依赖 class ChainableTool(Tool): input_dependencies: List[str] = [] # 依赖的LLM输出字段名 def can_execute(self, state: AgentState) -> bool: # 检查state中是否有足够字段满足依赖 for dep in self.input_dependencies: if not hasattr(state, dep) or getattr(state, dep) == "": return False return True class OutfitRecommendTool(ChainableTool): name = "recommend_outfit" description = "根据温度推荐穿搭" input_dependencies = ["temperature"] # 依赖temperature字段 def execute(self, temperature: float) -> str: return "建议穿短袖+薄外套" if temperature > 22 else "建议穿长袖+毛衣" # 修改step函数,支持链式调用 def enhanced_step(state: AgentState) -> AgentState: # ... LLM调用与解析逻辑不变 ... # 新增:检查工具是否可执行 tool = TOOL_MAP.get(new_action) if isinstance(tool, ChainableTool) and not tool.can_execute(state): # 依赖不满足,要求LLM重新思考 return state._replace( observation=f"工具'{new_action}'需要字段{tool.input_dependencies},但当前状态缺失" ) # ... 工具执行与状态更新逻辑 ...这个增强版只有30行新增代码,但它解决了LangChain的硬伤:工具间的数据流必须通过LLM中转,无法直接传递。在手写版本里,get_weather工具可以直接把temperature: 25.0注入state,recommend_outfit就能直接读取——省去LLM二次解析,降低幻觉风险。
更重要的是,抽象成本可视化。LangChain的create_react_agent隐藏了所有状态管理细节,而手写版本让你清楚看到:
- 每次
state._replace()消耗多少内存; messages列表增长如何影响token计费;tool.execute()超时如何阻塞整个循环。
这种透明性,是选型决策的基石。当业务要求“单次响应<800ms”,你会立刻意识到:LangChain的AgentExecutor自带重试、日志、异常包装三层开销,而手写版本可以砍掉所有非必要环节。
5. 避坑指南:手写Agent必踩的7个深坑与破局点
手写不是为了炫技,是为了建立对Agent底层逻辑的肌肉记忆。但这条路布满陷阱,我整理了学员最常栽跟头的7个点,每个都附真实案例和解决方案。
5.1 坑1:LLM输出格式漂移——“Thought”突然变成“Think”
现象:Agent运行几天后突然失效,日志显示KeyError: 'thought'。
根因:LLM在压力下改变输出格式,把Thought:写成Think:或My thought:。
破局:放弃正则匹配,改用LLM自身做格式校验。在prompt里加一句:
“请用以下JSON格式输出,不要添加任何额外文本:{...}”
同时,在解析层加fallback:
def robust_parse(llm_output: str) -> Dict: try: return json.loads(llm_output) except: # 用LLM自己修正格式 fix_prompt = f"将以下文本转为JSON:{llm_output}" fixed = call_llm_simple(fix_prompt) # 简化版LLM调用 return json.loads(fixed)5.2 坑2:工具返回乱码——中文变\u4f60\u597d
现象:天气API返回{"city": "\u5317\u4eac"},Agent显示“城市:\u4f60\u597d”。
根因:HTTP响应未声明charset=utf-8,Python默认用latin-1解码。
破局:所有工具执行层强制指定编码:
resp = requests.get(url) resp.encoding = "utf-8" # 关键! return resp.text5.3 坑3:状态爆炸——messages列表无限增长
现象:运行10轮后,state.messages长达200条,token超限报错。
根因:每次state._replace()都追加新消息,未做摘要压缩。
破局:实现滚动摘要:
def compress_messages(messages: List[Dict]) -> List[Dict]: if len(messages) <= 6: return messages # 保留首尾各2条,中间用LLM摘要 summary_prompt = "用100字总结以下对话:" + "\n".join( [f"{m['role']}: {m['content'][:50]}" for m in messages[2:-2]] ) summary = call_llm_simple(summary_prompt) return messages[:2] + [{"role": "system", "content": f"对话摘要:{summary}"}] + messages[-2:]5.4 坑4:工具死锁——Action Input为空字典
现象:LLM输出Action Input: {},工具执行时报TypeError: execute() missing 1 required argument。
根因:LLM未按schema生成参数,而工具函数签名强制非空。
破局:工具层加参数校验:
def execute(self, **kwargs) -> str: # 检查必需参数 required = ["city"] if self.name == "get_weather" else [] for r in required: if r not in kwargs or not kwargs[r]: raise ValueError(f"缺少必需参数:{r}") return self._real_execute(**kwargs)5.5 坑5:Observation污染——混入调试日志
现象:Agent把DEBUG: calling weather API...当成观测结果,LLM据此胡说八道。
根因:工具执行时print语句未关闭。
破局:工具基类统一捕获stdout:
import io from contextlib import redirect_stdout def execute(self, **kwargs) -> str: f = io.StringIO() with redirect_stdout(f): result = self._real_execute(**kwargs) return f.getvalue().strip() or result5.6 坑6:FINISH信号失效——LLM永远不喊停
现象:Agent循环5次后强制终止,但LLM其实已得出结论。
根因:LLM认为“任务完成”不等于“该输出FINISH”,尤其在多步骤任务中。
破局:在prompt里明确定义终止条件:
“当且仅当以下任一条件满足时,输出FINISH:
- 已获取所有用户要求的信息
- 用户问题已被明确回答
- 工具调用返回错误且无法重试”
5.7 坑7:Token黑洞——Observation过长挤占思考空间
现象:天气API返回2000字详细报告,LLM没空间写Thought就截断。
根因:Observation未做摘要,直接塞进上下文。
破局:Observation预处理:
def truncate_observation(obs: str, max_len: int = 500) -> str: if len(obs) <= max_len: return obs # 用LLM摘要(成本高但必要) summary = call_llm_simple(f"用50字摘要以下内容:{obs[:2000]}") return f"摘要:{summary}(原文过长已截断)"这些坑,LangChain文档不会写,Stack Overflow搜不到,只有亲手写过5遍ReAct循环的人,才会在凌晨三点盯着日志时突然顿悟。而这,正是这门课最珍贵的部分——它不教你怎么“正确”,而教你怎么在“错误”中活下来。
6. 从Python到Rust:为什么手写经验能无缝迁移到任何语言
最近“基于Rust语言AI Agent”成为热搜词,很多学员问我:“学完Python手写,转Rust要重学吗?”我的答案是:你学的从来不是Python,而是Agent的本质契约。
Rust版ReAct Agent的核心结构,和我们手写的Python版几乎一致:
// Rust版State定义(对比Python NamedTuple) #[derive(Clone, Debug)] pub struct AgentState { pub messages: Vec<Message>, pub thought: String, pub action: String, pub action_input: serde_json::Value, pub observation: String, pub is_finished: bool, pub final_result: Option<String>, } // Rust版step函数(对比Python step) pub fn step(state: &AgentState, tools: &Vec<Box<dyn Tool>>) -> Result<AgentState, Box<dyn std::error::Error>> { let llm_output = call_llm(state, tools)?; let parsed = parse_llm_output(&llm_output)?; // JSON解析 if parsed.action == "FINISH" { return Ok(AgentState { is_finished: true, final_result: Some(parsed.action_input["result"].to_string()), ..state.clone() }); } let observation = execute_tool(&parsed.action, &parsed.action_input)?; Ok(extend_messages(state, &parsed, &observation)) }差异只在语法层面:
- Python用
NamedTuple._replace(),Rust用struct字段更新; - Python用
requests,Rust用reqwest; - Python用
json.loads(),Rust用serde_json::from_str()。
但状态机逻辑、ReAct循环、工具契约、错误处理范式,完全一致。我带过一个用Rust写Agent的团队,他们最大的收获不是性能提升,而是:
- 终于理解为什么LangChain的
AgentExecutor要用Arc<Mutex<>>——因为Python的GIL让他们误以为线程安全是免费的; - 明白
tokio::spawn不是万能解药,当Observation需要LLM摘要时,异步反而增加复杂度; - 看懂Spring AI的
AiResponse为何要分content和metadata——Rust的Option<T>让这种分离变得自然。
所以,这门课的终极价值,不是让你成为Python专家,而是让你获得一种跨语言的Agent直觉:
- 看到任何Agent框架,第一反应不是“怎么用”,而是“它的State怎么定义?Transition怎么触发?Error怎么传播?”;
- 遇到线上故障,不再盲目查日志,而是按“State→LLM→Tool→Observation→State”链条逐段验证;
- 评估新技术时,本能追问:“它解决了哪个手写时的痛点?又引入了什么新抽象成本?”
这种直觉,无法通过调包获得,只能用手写千行代码浇灌出来。当你亲手实现过ReAct的每一次失败回滚,你就不会再被“AI Agent主流架构”这类虚词迷惑——你知道,所有架构,不过是不同语言对同一组约束的不同表达。
我在实际开发中发现,手写Agent的开发者,接手任何框架的速度都快3倍。因为他们不是在学API,而是在找“这个框架的step函数藏在哪”。这种能力,才是AI时代真正的护城河。