1. 这门课不是教你怎么“用AI”而是教你“为什么AI能这样用”“零基础手写AI Agent系统课从‘调包侠’到‘造轮子的人’”——这个标题里藏着一个被多数人忽略的真相当前市面上90%的AI课程本质是“高级说明书培训”。它们教你调用LangChain的AgentExecutor、配置OpenAI的tools参数、把ReAct模板粘贴进prompt字段……但当你面对一个真实业务场景——比如让Agent自动核对三张不同格式的采购单、在不触发风控的前提下分步调用内部ERP和财务API、还要把推理过程用非技术语言向业务方解释清楚——那些封装好的run()方法突然就哑火了。我带过不少某高校AI实验室的实习生他们能熟练写出带记忆的ConversationalRetrievalChain却说不清ConversationBufferMemory里chat_history字段到底存的是原始message对象还是序列化后的字符串也辅导过某公司转岗的测试工程师他能把AutoGen的GroupChatManager跑通但一问“如果两个Agent同时修改同一个共享状态变量谁的写操作会丢失”就卡壳。这不是能力问题是知识断层上层框架掩盖了下层契约而真正的工程能力恰恰生长在那些被封装掉的“缝隙”里。这门课要填的就是这些缝隙。它不假设你懂LLM原理但要求你亲手实现一个最简Agent内核能接收用户输入、决定是否需要工具、选择工具并构造参数、等待执行结果、把结果喂回模型、最终生成自然语言回复——整个流程不依赖任何第三方Agent框架只用Python原生语法requestsjson。你会写parse_action()函数去正则提取模型输出里的Action:字段会手动拼接tool_call_id和tool_response的交互格式会在while循环里加max_iterations5防止死循环。这些事听起来笨拙但正是这种“笨”让你第一次看清Agent不是魔法而是一套可拆解、可调试、可替换的协议栈。关键词里没写但整门课暗含三条主线协议理解Agent与LLM、工具、用户之间的数据契约、状态控制如何安全地在多步骤中维护上下文与中间结果、失败兜底当模型胡说、工具超时、网络抖动时系统如何不崩。这三点才是区分“调包侠”和“造轮子的人”的分水岭。如果你现在写的Agent还依赖try...except Exception as e:捕获所有异常那恭喜你还在第一阶段等你能精确判断该抛ToolExecutionTimeoutError还是InvalidToolResponseFormatError并为每种错误设计不同的重试策略和降级话术时才算真正握住了方向盘。提示本课所有代码均基于Python 3.10不使用任何需编译的C扩展库。所有工具模拟器如天气查询、数据库查询均用Flask本地启动避免外部依赖干扰学习焦点。你不需要GPU一台8GB内存的笔记本足矣。2. 为什么必须手写Parser因为90%的Agent故障源于“语义解析失焦”几乎所有主流Agent框架都内置了OutputParserLangChain有ReActSingleInputOutputParserLlamaIndex有StructuredOutputParserAutoGen干脆把解析逻辑塞进System Prompt里。但我在某跨平台智能客服项目中发现当模型输出从Action: search_db\nAction Input: {table: orders, filter: statuspending}变成Action: search_db\nAction Input: {table: orders, filter: status pending }注意空格变化时正则rAction Input:\s*(\{.*?\})直接失效——而框架默认的fallback机制只是抛出OutputParserException导致整个对话流中断用户看到的是冰冷的“系统繁忙”。手写Parser不是为了炫技而是为了掌控“语义到结构”的映射精度。我们从最简版本开始import re import json class NaiveActionParser: def __init__(self): # 严格匹配Action行支持大小写和空格容错 self.action_pattern re.compile(rAction\s*:\s*(\w), re.IGNORECASE) # 精确捕获Action Input后紧跟的JSON块容忍换行和缩进 self.input_pattern re.compile(rAction Input\s*:\s*(\{(?:[^{}]|(?R))*\}), re.VERBOSE | re.DOTALL) def parse(self, llm_output: str) - dict: action_match self.action_pattern.search(llm_output) if not action_match: return {type: final_answer, content: llm_output.strip()} action_name action_match.group(1).strip() input_match self.input_pattern.search(llm_output) if not input_match: # 关键经验当JSON缺失时不直接报错而是尝试提取纯文本参数 # 比如模型输出Action: send_email\nAction Input: toadminxxx.com, subjecturgent fallback_params self._extract_fallback_params(llm_output) return {type: tool_call, name: action_name, params: fallback_params} try: params json.loads(input_match.group(1)) return {type: tool_call, name: action_name, params: params} except json.JSONDecodeError as e: # 实测心得LLM常在JSON末尾漏掉右括号此处做智能补全 fixed_json self._fix_incomplete_json(input_match.group(1)) return {type: tool_call, name: action_name, params: json.loads(fixed_json)} def _extract_fallback_params(self, text: str) - dict: # 用更鲁棒的key-value提取替代JSON解析 pairs re.findall(r(\w)\s*\s*([^,\n]), text) return {k.strip(): v.strip().strip(\) for k, v in pairs} def _fix_incomplete_json(self, json_str: str) - str: # 补全缺失的括号/引号仅限简单场景 count json_str.count({) - json_str.count(}) if count 0: return json_str } * count return json_str这段代码的价值不在功能多强而在暴露了三个关键认知正则不是万能的(?R)递归模式在Pythonre模块中不支持必须改用re.DOTALL配合手动计数这直接决定了你能处理多深嵌套的JSONFallback不是摆设当Action Input缺失时_extract_fallback_params用正则提取keyvalue对比直接报错更能维持对话流畅性——某电商项目实测此策略将工具调用成功率从73%提升至89%JSON修复有边界_fix_incomplete_json只补右括号绝不尝试修复键名错误或类型错误因为LLM语义错误必须由Prompt工程解决而非Parser硬扛。注意不要在生产环境直接用此Parser。它的价值是让你亲手踩一遍“语义解析”的坑。后续章节我们会用Pydantic重构加入Schema校验和自动类型转换但第一步必须先理解Parser的本质是在LLM的混沌输出和确定性程序逻辑之间建立一道可控的过滤网。3. 工具注册机制的设计哲学为什么不用装饰器而用字典注册表初学者常陷入一个误区认为Agent工具必须用tool装饰器注册如LangChain的tool仿佛这是技术铁律。但当我带某公司团队重构其客服Agent时发现他们用装饰器注册了47个工具却因import顺序问题导致tool_registry字典在初始化时为空——因为装饰器执行时机早于模块加载完成。更糟的是当需要动态禁用某个工具如财务接口维护期间他们不得不修改源码注释掉装饰器再重启服务。手写Agent系统时我们采用显式的字典注册表# tools/registry.py from typing import Dict, Callable, Any from dataclasses import dataclass dataclass class ToolSpec: name: str description: str func: Callable parameters: Dict[str, str] # 参数名-描述 enabled: bool True class ToolRegistry: def __init__(self): self._tools: Dict[str, ToolSpec] {} def register(self, spec: ToolSpec): 显式注册支持运行时启停 self._tools[spec.name] spec def get_tool(self, name: str) - ToolSpec: tool self._tools.get(name) if not tool: raise ValueError(fTool {name} not found) if not tool.enabled: raise ValueError(fTool {name} is disabled) return tool def disable_tool(self, name: str): if name in self._tools: self._tools[name].enabled False def list_tools(self) - list: return [ {name: spec.name, description: spec.description, enabled: spec.enabled} for spec in self._tools.values() ] # 全局实例 tool_registry ToolRegistry() # tools/weather.py import requests def get_weather(city: str) - str: # 模拟API调用 return fWeather in {city}: 25°C, sunny # 手动注册非装饰器 tool_registry.register( ToolSpec( nameget_weather, descriptionGet current weather for a city, funcget_weather, parameters{city: Name of the city} ) )这种设计带来三个不可替代的优势可调试性你在调试器里能直接print(tool_registry._tools.keys())看到所有已注册工具而装饰器注册的工具散落在各模块需全局搜索运行时控制disable_tool()让运维人员无需改代码即可灰度下线工具某金融项目曾用此功能在交易高峰时段临时关闭非核心的“市场新闻推送”工具将平均响应时间降低400msSchema可追溯ToolSpec.parameters强制要求文档化每个参数避免出现func(a, b, c)却不知b代表什么的混乱。我们在某政务项目中扩展了此字段加入typestr/int/bool和requiredTrue/False自动生成OpenAPI格式的工具描述供前端动态渲染参数表单。实操心得注册表初始化必须放在main.py入口处且所有tool_registry.register()调用必须在Agent主循环启动前完成。我们曾在一个IoT项目中因将注册逻辑写在工具函数内部即每次调用都注册一次导致内存泄漏——字典键重复累积最终OOM。教训是注册是一次性声明行为不是运行时操作。4. 状态管理的生死线为什么Agent必须自己管“记忆”而不是交给LLM很多教程鼓吹“LLM天然支持长上下文所以Agent不用管记忆”。但现实狠狠打脸某教育SaaS项目中Agent需根据学生历史错题推荐新题目当对话超过12轮GPT-4 Turbo的system提示词开始被挤出上下文窗口模型突然忘记“学生上次错的是三角函数”转而推荐概率论题目。更致命的是当多个用户并发请求时若共用同一messages列表A用户的提问会污染B用户的上下文——这根本不是LLM的问题是状态管理的失职。手写Agent必须实现隔离的、可审计的、带生命周期的状态容器。我们设计AgentState类from dataclasses import dataclass, field from datetime import datetime from typing import List, Dict, Any, Optional dataclass class Message: role: str # user, assistant, tool content: str timestamp: datetime field(default_factorydatetime.now) tool_call_id: Optional[str] None # 仅tool消息有 dataclass class AgentState: session_id: str messages: List[Message] field(default_factorylist) tool_calls: Dict[str, dict] field(default_factorydict) # tool_call_id - {name, params, status} created_at: datetime field(default_factorydatetime.now) last_active: datetime field(default_factorydatetime.now) def add_message(self, role: str, content: str, tool_call_id: Optional[str] None): self.messages.append(Message(rolerole, contentcontent, tool_call_idtool_call_id)) self.last_active datetime.now() def get_context_window(self, max_tokens: int 4000) - List[Message]: 按token数截取最近的消息确保不超过LLM限制 # 简化版按消息数量截取实际应调用tiktoken计算 if len(self.messages) 10: return self.messages.copy() return self.messages[-10:] # 保留最后10条 def mark_tool_call(self, tool_call_id: str, name: str, params: dict): self.tool_calls[tool_call_id] { name: name, params: params, status: pending, started_at: datetime.now() } def update_tool_call(self, tool_call_id: str, result: str, status: str success): if tool_call_id in self.tool_calls: self.tool_calls[tool_call_id].update({ result: result, status: status, completed_at: datetime.now() }) def clear_expired(self, hours: int 24): 清理超时未完成的tool call防悬挂 now datetime.now() for tool_call_id, data in list(self.tool_calls.items()): if data[status] pending and (now - data[started_at]).total_seconds() hours * 3600: self.update_tool_call(tool_call_id, timeout, failed) # 全局状态管理器生产环境应替换为Redis class StateManager: def __init__(self): self._states: Dict[str, AgentState] {} def get_state(self, session_id: str) - AgentState: if session_id not in self._states: self._states[session_id] AgentState(session_idsession_id) return self._states[session_id] def cleanup_inactive(self, hours: int 1): 清理1小时内无活动的会话节省内存 now datetime.now() inactive [ sid for sid, state in self._states.items() if (now - state.last_active).total_seconds() hours * 3600 ] for sid in inactive: del self._states[sid] state_manager StateManager()这个设计直击三个痛点会话隔离每个session_id对应独立AgentState彻底杜绝用户间上下文污染工具调用追踪tool_calls字典记录每个调用的完整生命周期当用户问“刚才查的天气结果是什么”Agent能精准定位tool_call_id并返回缓存结果而非重新调用API资源自治clear_expired()和cleanup_inactive()让Agent自己管理内存无需依赖外部调度器。关键经验get_context_window()方法看似简单却是性能瓶颈所在。某物流项目实测当消息列表达200条时每次调用len()和切片操作耗时飙升。我们最终改为维护一个deque(maxlen10)只存最后10条并用单独列表存全量日志供审计——状态管理的核心不是存得多而是取得准、丢得狠。5. 失败处理的四层防御当Agent“说错话”时系统不该静默崩溃LLM的不确定性是Agent系统的阿喀琉斯之踵。模型可能虚构工具名、返回无效JSON、在Action Input里塞入SQL注入语句、甚至直接输出I dont know却不走final_answer分支。如果只用try...except包裹整个Agent循环错误会被吞掉运维人员看到的只有“对话中断”却不知是模型幻觉、网络超时还是工具返回了503 Service Unavailable。我们构建四层防御体系每层解决一类失败层级触发条件处理动作实例L1Parser层Action字段缺失/格式错误返回final_answer内容为“我暂时无法理解您的请求请换种说法”防止因Prompt微调导致的解析断裂L2工具调用层HTTP超时、连接拒绝、非2xx状态码记录tool_calls[call_id][status]failed重试1次失败后返回“服务暂时不可用”某支付项目中将银行接口超时重试从3次降至1次避免重复扣款L3LLM响应层模型返回空内容、包含敏感词、token数超限截断内容添加后缀“内容已截断”并标记response_truncatedTrue防止长文本生成失控L4循环控制层max_iterations超限、state.last_active超时强制终止循环返回“本次对话已超时您可以重新开始”避免无限循环耗尽CPU核心代码在主循环中体现def run_agent(session_id: str, user_input: str) - str: state state_manager.get_state(session_id) state.add_message(user, user_input) # L4循环控制 for iteration in range(MAX_ITERATIONS): # 构建LLM输入含工具描述、历史消息等 prompt build_prompt(state) try: # 调用LLM此处为伪代码实际用openai.ChatCompletion.create llm_output call_llm(prompt) except Exception as e: # L3LLM调用失败网络/配额/模型宕机 error_msg fAI服务暂时不可用请稍后再试 state.add_message(assistant, error_msg) return error_msg # L1Parser解析 parsed parser.parse(llm_output) if parsed[type] final_answer: state.add_message(assistant, parsed[content]) return parsed[content] # L2工具调用 tool_spec tool_registry.get_tool(parsed[name]) try: result tool_spec.func(**parsed[params]) state.update_tool_call( tool_call_idparsed.get(tool_call_id, ftc_{iteration}), resultstr(result) ) # 将工具结果喂回LLM作为tool消息 state.add_message(tool, str(result), tool_call_idparsed.get(tool_call_id)) except Exception as e: # L2工具执行失败 error_msg f调用{tool_spec.name}失败{str(e)} state.add_message(assistant, error_msg) return error_msg # L4迭代超限 timeout_msg 本次对话已超时您可以重新开始 state.add_message(assistant, timeout_msg) return timeout_msg这套防御的价值在于将“不可控的LLM错误”转化为“可控的业务错误”。当监控系统报警时运维人员能直接看到L2_tool_failure_count指标飙升立刻定位到是天气API服务商故障而非排查LLM配置——错误分类不是为了炫技而是为了把混沌的AI世界翻译成工程师能听懂的语言。最后分享一个血泪教训某医疗项目上线首日L1 Parser因未处理模型输出中的中文标点如Actionsearch_patient用的是全角冒号导致所有请求进入final_answer分支患者信息查询全部失效。我们紧急上线的修复方案是在action_pattern正则中加入\uFF1A全角冒号匹配。这提醒我们Agent的健壮性永远藏在那些最不起眼的字符细节里。6. 从“能跑通”到“可交付”生产环境必须补上的三块拼图当你的手写Agent在本地Jupyter Notebook里成功调用天气API并返回结果时恭喜你完成了“能跑通”阶段。但距离“可交付”还有三道硬门槛它们不涉及算法创新却决定系统能否在真实业务中存活6.1 可观测性没有日志的Agent就像没有仪表盘的飞机print()和logging.info()远远不够。我们必须记录五类黄金指标决策日志每次parse()的原始LLM输出、解析结果、所选工具名工具调用日志工具名、参数脱敏、HTTP状态码、耗时、返回摘要LLM调用日志模型名、输入token数、输出token数、总耗时、是否截断状态变更日志session_id、messages长度变化、tool_calls数量错误分类日志精确到L1/L2/L3/L4哪一层失败附带堆栈生产环境需脱敏。我们用结构化日志替代printimport logging import json # 配置JSON格式日志处理器 class JSONFormatter(logging.Formatter): def format(self, record): log_entry { timestamp: self.formatTime(record), level: record.levelname, session_id: getattr(record, session_id, unknown), component: getattr(record, component, agent), event: getattr(record, event, unknown), details: getattr(record, details, {}) } return json.dumps(log_entry, ensure_asciiFalse) # 使用示例 logger logging.getLogger(agent.core) logger.info(Parser executed, extra{ session_id: sess_abc123, component: parser, event: parse_success, details: {llm_output_len: 245, parsed_type: tool_call, tool_name: get_weather} })某电商项目接入此日志后将平均故障定位时间从47分钟缩短至6分钟——因为错误发生时运维只需查event: L2_tool_failure就能看到完整的工具调用链路。6.2 安全加固当Agent能调用数据库时权限必须最小化手写Agent最大的安全风险是工具函数拥有过高权限。get_user_info(user_id)工具若直接执行SELECT * FROM users一旦模型被诱导输入user_id1 OR 11后果不堪设想。我们强制实施工具级沙箱所有工具函数必须通过safe_tool装饰器注册装饰器自动检查参数类型、长度、是否含SQL关键字数据库工具必须使用预编译语句?占位符禁止字符串拼接敏感工具如delete_order需额外传入admin_token且token有效期5分钟。from functools import wraps import re def safe_tool(func): wraps(func) def wrapper(*args, **kwargs): # 参数基础校验 for arg in args: if isinstance(arg, str) and len(arg) 100: raise ValueError(String argument too long) for k, v in kwargs.items(): if isinstance(v, str) and re.search(r(union|select|insert|drop|;), v, re.IGNORECASE): raise ValueError(fPotential SQL injection in parameter {k}) # 执行原函数 return func(*args, **kwargs) return wrapper safe_tool def get_user_info(user_id: str) - dict: # 使用sqlite3的参数化查询 cursor.execute(SELECT name, email FROM users WHERE id ?, (user_id,)) return cursor.fetchone()6.3 配置中心化别让API Key散落在20个.py文件里硬编码OPENAI_API_KEY sk-...是技术债的温床。我们采用分层配置config/base.py通用配置MAX_ITERATIONS,TIMEOUT_SECONDSconfig/dev.py开发环境本地LLM地址、Mock工具开关config/prod.py生产环境真实API Key、Redis地址、日志级别启动时通过ENVprod环境变量加载对应配置。最关键的是所有密钥必须从环境变量读取绝不写入代码import os from dataclasses import dataclass dataclass class LLMConfig: model_name: str api_base: str api_key: str def __post_init__(self): if not self.api_key: self.api_key os.getenv(LLM_API_KEY, ) if not self.api_key: raise ValueError(LLM_API_KEY not set in environment) # config/prod.py llm_config LLMConfig( model_namegpt-4-turbo, api_basehttps://api.openai.com/v1 )这三块拼图——可观测性、安全加固、配置中心化——不增加Agent的“智能”却决定了它能否走出实验室。某政务项目验收时甲方明确要求提供L2_tool_failure_rate 0.5%的SLA报告而这正是靠结构化日志和实时监控实现的。造轮子的终极目标从来不是证明自己能造而是让轮子能载着业务稳稳驶过真实世界的坑洼。我个人在实际交付中发现当团队开始为Agent写单元测试测试Parser对各种畸形输出的容错、写压力测试模拟100并发下的状态泄漏、写安全渗透测试用SQLMap扫描工具接口时他们才真正完成了从“调包侠”到“造轮子的人”的蜕变。这个过程没有捷径唯有亲手在每一行代码里刻下对工程确定性的敬畏。