ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

手写AI Agent系统:从零构建可调试的Agent核心

手写AI Agent系统:从零构建可调试的Agent核心

1. 这门课到底在教什么:不是写AI,是在重建你和代码的关系

“零基础手写AI Agent系统课:从‘调包侠’到‘造轮子的人’”——这个标题里藏着三个被多数人忽略的关键词:零基础、手写、Agent系统。它不承诺让你三天写出GPT-4,也不鼓吹“学完年薪百万”,而是直指当前AI学习者最真实的困境:你熟练调用LangChain的ChatAgent,能背出LlamaIndex的QueryEngine参数,但当提示词失效、工具链崩掉、响应延迟突增时,你翻遍文档却找不到问题在哪,因为底层逻辑像黑箱,你只站在API门口递纸条。

我带过几十个从Python入门转AI开发的学员,发现一个高度一致的现象:90%的人卡在“调试无力感”上。不是不会写,是不知道该往哪改。比如某次调试一个天气查询Agent,用户问“明天上海会下雨吗”,返回却是“未识别城市”,而日志里只显示ToolExecutionError: city not found。表面看是工具函数没处理空值,深挖才发现是LLM输出的JSON格式被json.loads()直接解析失败,而错误被上游异常捕获机制吞掉了——这种链路级故障,靠pip install解决不了,必须亲手把Agent的调度循环、工具调用封装、响应解析器一层层铺开来看。

这门课的“零基础”不是营销话术,它默认你只会print("Hello"),但要求你从第1行代码开始,手写一个能接收用户输入、自主决定调用哪个工具、等待结果、再生成自然语言回复的完整闭环。它不跳过while True:主循环的阻塞设计,不省略tool_call_id如何在多轮交互中保持上下文关联,更不会用@agent.tool装饰器掩盖函数注册的本质。所谓“造轮子”,本质是把抽象概念具象为可触摸、可打断、可单步调试的代码实体。当你亲手实现一个带记忆缓存的ToolRegistry类,你会突然理解为什么LangChain要区分BaseTool和StructuredTool;当你手动拼接system_prompt + history + user_input再喂给模型,你就不会再盲目相信“加个few-shot就能提升准确率”。

适合谁?三类人最该停下刷短视频的手来学:第一类是刚学完《Python入门》就急着冲进AI领域的新人,你缺的不是知识密度,而是对“程序如何一步步运行”的肌肉记忆;第二类是能跑通Demo但不敢改一行核心逻辑的中级开发者,你的瓶颈不在技术广度,而在对执行流的掌控力;第三类是技术管理者或课程设计者,你需要真正理解Agent系统的耦合点与扩展边界,才能判断一个需求该用现成框架还是自研模块。这不是速成课,它是给你配一把解剖刀,让你看清AI应用的血管、神经和骨骼。

2. 为什么必须“手写”:绕开框架黑箱的四个不可替代价值

很多人质疑:“现有框架都封装好了,为什么还要花时间手写?生产环境谁会这么干?”这个问题问到了根子上。答案很实在:手写不是为了替代框架,而是为了获得框架的“源码级直觉”。就像学开车不必先造发动机,但若想成为赛车调校师,你得知道活塞环间隙怎么影响扭矩输出。我们拆解手写的四大不可替代价值:

2.1 理解控制流的“呼吸节奏”

所有Agent框架的核心都是一个循环:接收输入→规划动作→执行工具→整合结果→生成响应→循环。但LangChain的AgentExecutor.run()方法像一台密封的蒸汽机,你只看到活塞运动,看不到气阀何时开启、冷凝水如何排出。手写时,你会被迫直面这些细节:

  • 输入缓冲区怎么设计?是用input()同步阻塞,还是用asyncio.Queue异步排队?前者简单但无法支持多用户,后者需处理await queue.get()的取消逻辑;
  • 规划阶段如何防幻觉?当LLM返回{"tool": "weather", "args": {"city": "北京"}},你得验证city字段是否存在、是否为空字符串、是否在预设城市白名单内——这步校验在框架里常被简化为try/except,而手写时你得决定是抛出ValidationError还是静默修正为默认城市;
  • 工具执行的超时熔断怎么嵌入?requests.get(url, timeout=5)的5秒是硬性上限,但若工具链包含本地计算(如PDF文本提取),超时策略就得区分I/O密集型和CPU密集型任务。

我曾让学员手写一个带超时的execute_tool函数,80%的人第一版用threading.Timer,结果发现线程无法强制终止正在执行的CPU操作;第二版改用concurrent.futures.ProcessPoolExecutor,又遇到进程间内存拷贝开销过大。最终大家统一采用asyncio.wait_for()配合asyncio.to_thread(),这才真正理解为什么现代Agent框架普遍拥抱异步——不是为了炫技,而是为超时控制提供确定性保障。

2.2 掌握数据流的“血型匹配”

Agent系统里最隐蔽的坑,往往出在数据格式的隐式转换上。框架帮你做了太多“智能适配”:自动把字符串转成JSON、把字典序列化为OpenAI格式、把工具返回的dict注入messages列表。手写时,你得亲手缝合每一处接口:

  • LLM输入格式:OpenAI API要求messages是[{"role": "user", "content": "..."}],但你的工具调用结果可能是{"temperature": 25.3, "humidity": 68}。是直接str(result)硬塞进去?还是用json.dumps(result, ensure_ascii=False)?前者导致LLM看不懂结构化数据,后者若result含datetime对象会报错;
  • 工具输出解析:当天气API返回{"code": 200, "data": {"city": "上海", "weather": "晴"}},你的解析器是取data字段还是整个响应体?如果API升级返回{"status": "success", "payload": {...}},你的正则匹配规则会不会全线崩溃?
  • 历史记录压缩:对话超过20轮后,messages列表可能超Token限制。是删最早的user消息?还是合并相邻的assistant回复?手写时你得实现compress_history(messages, max_tokens=4096),并测试不同压缩策略对后续推理准确率的影响——这个过程会让你彻底放弃“框架自动处理”的幻想。

2.3 构建可观测性的“诊断接口”

生产环境中,Agent故障的70%源于不可见的中间态。框架的日志常是“Agent started → Tool executed → Response generated”,而真实问题藏在缝隙里:LLM规划时把“查股票”误判为“查天气”,工具执行返回了HTTP 503但被静默忽略,历史记录因编码问题导致中文乱码进而污染后续推理。手写系统时,你必须主动埋点:

  • 在plan_action()函数开头打印raw_llm_output,结尾打印解析后的tool_name和tool_args;
  • 在execute_tool()前后记录start_time和end_time,并捕获所有异常类型(ConnectionError、JSONDecodeError、KeyError);
  • 为每轮对话生成唯一session_id,将所有日志按ID聚合,方便追踪完整链路。

有位做金融客服Agent的学员,就是靠手写时加的log_step("tool_result_parsing", result_str[:100]),定位到第三方天气API返回的"temp": "25°C"中°C符号导致int()转换失败——这个细节在框架的ToolResultParser里被try/except吞掉,日志只显示“工具执行失败”,而手写版本直接暴露了原始字符串。

2.4 获得架构演进的“决策标尺”

当你亲手实现过SimpleRouter(基于关键词路由)、LLMRouter(用LLM判断工具)、HybridRouter(关键词+LLM置信度加权),你再看LangChain的RouterChain文档时,就不再是被动接受,而是带着批判性思考:它的next_chain选择逻辑是否覆盖了多意图场景?当用户说“帮我查北京天气,顺便订张去上海的机票”,单路由模式必然失败,这时你需要的是MultiToolRouter。手写过程逼你直面架构权衡:

  • 性能 vs 灵活性:正则路由快但脆弱,LLM路由强但贵;
  • 确定性 vs 智能性:硬编码规则保证100%可控,但无法处理长尾表达;
  • 维护成本 vs 开发速度:自研工具注册中心要写register()、get_by_name()、list_all()三个方法,而框架一行@tool搞定,但后者升级时可能破坏你的自定义装饰器逻辑。

这种亲手造轮子的经历,会给你一套内在的评估标尺:当业务需要支持100+工具时,你会果断放弃手写路由,转向LlamaIndex的ToolRetriever;当实时性要求<200ms,你会砍掉所有LLM-based routing,回归关键词匹配。这不是妥协,而是基于深度理解的主动选择。

3. 手写Agent系统的核心模块拆解:从main.py开始的七层楼

我们以一个极简但完整的Agent系统为例,它能处理“查天气”和“算数学”两个工具,代码总行数控制在300行内,但覆盖了Agent所有关键模块。这不是玩具Demo,而是生产级系统的微缩沙盒。下面逐层拆解,每层都附带“为什么这样设计”的底层逻辑。

3.1 第一层:主循环与输入输出(main.py)

# main.py import asyncio from core.agent import SimpleAgent async def main(): agent = SimpleAgent() print("🤖 Agent已启动!输入'quit'退出") while True: try: user_input = input("You: ").strip() if user_input.lower() in ["quit", "exit", "q"]: print("👋 再见!") break if not user_input: continue # 关键:这里不是直接调用run(),而是暴露执行步骤 response = await agent.run(user_input) print(f"Agent: {response}") except KeyboardInterrupt: print("\n👋 强制退出") break except Exception as e: print(f"❌ 系统错误: {e}") if __name__ == "__main__": asyncio.run(main())

设计逻辑:主循环看似简单,但藏着三个关键决策。第一,input()是同步阻塞的,为何不用asyncio.get_event_loop().run_in_executor()?因为教学场景下,同步IO更易观察执行顺序——你能在终端清晰看到“You:”出现后,程序才进入agent.run(),避免异步IO带来的时序混乱。第二,try/except捕获全局异常而非仅KeyboardInterrupt,这是为后续埋点留接口:当工具执行超时,你希望看到TimeoutError而非程序崩溃。第三,agent.run()返回纯字符串而非dict,降低初学者理解成本——他们只需关注“说了什么”,而非“返回了什么结构”。

3.2 第二层:Agent核心调度器(core/agent.py)

# core/agent.py import json import asyncio from typing import Dict, Any, List, Optional from core.llm import call_llm from core.tools import ToolRegistry from core.memory import ConversationMemory class SimpleAgent: def __init__(self): self.tool_registry = ToolRegistry() self.memory = ConversationMemory(max_history=10) async def run(self, user_input: str) -> str: # 步骤1:加载历史记录 history = self.memory.get_history() # 步骤2:构造LLM输入(显式展示prompt工程) system_prompt = "你是一个有用的AI助手。请根据用户问题,选择合适的工具执行,并用自然语言回答。" messages = [{"role": "system", "content": system_prompt}] messages.extend(history) messages.append({"role": "user", "content": user_input}) # 步骤3:调用LLM获取规划结果 llm_response = await call_llm(messages) # 步骤4:解析LLM输出(关键!暴露解析逻辑) try: plan = json.loads(llm_response) tool_name = plan.get("tool") tool_args = plan.get("args", {}) except json.JSONDecodeError: return "⚠️ 无法理解我的思考过程,请重试" # 步骤5:执行工具并处理结果 if tool_name and self.tool_registry.has_tool(tool_name): tool_result = await self.tool_registry.execute(tool_name, tool_args) # 步骤6:用工具结果生成最终回复 final_prompt = f"用户问:{user_input}\n工具返回:{tool_result}\n请用自然语言回答:" final_messages = [{"role": "system", "content": "你是一个回答助手。"}, {"role": "user", "content": final_prompt}] final_response = await call_llm(final_messages) # 步骤7:保存本轮对话到记忆 self.memory.add_interaction(user_input, final_response) return final_response else: return "❌ 不支持该功能,请尝试'查天气 上海'或'算 3+5'"

设计逻辑:这个run()方法是Agent的“心脏起搏器”,它强制暴露了七个原子步骤。重点看步骤4的JSON解析——这里没有用pydantic或dataclass,而是裸写json.loads(),因为初学者需要直面“LLM输出不稳定”这个事实:有时返回{"tool":"weather","args":{"city":"北京"}},有时返回{"tool": "weather", "args": {"city": "北京"}, "reason": "用户询问天气"},甚至可能返回纯文本“我帮你查北京天气”。手写时你必须决定:是严格校验JSON Schema?还是用正则提取关键字段?我们的方案是“宽松解析+防御性编程”,plan.get("tool")比plan["tool"]更安全,这教会学员第一个工程原则:永远假设外部输入是恶意的。

3.3 第三层:工具注册中心(core/tools.py)

# core/tools.py import asyncio import json from typing import Dict, Any, Callable, Awaitable class Tool: def __init__(self, name: str, func: Callable[..., Awaitable[Any]], description: str): self.name = name self.func = func self.description = description async def execute(self, **kwargs) -> Any: return await self.func(**kwargs) class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] = {} def register(self, name: str, func: Callable[..., Awaitable[Any]], description: str): """注册工具:暴露注册过程,而非装饰器""" self._tools[name] = Tool(name, func, description) def has_tool(self, name: str) -> bool: return name in self._tools async def execute(self, name: str, args: Dict[str, Any]) -> str: """执行工具:统一异常处理入口""" if not self.has_tool(name): return f"工具'{name}'不存在" try: result = await self._tools[name].execute(**args) return json.dumps(result, ensure_ascii=False, indent=2) except Exception as e: return f"工具执行失败: {type(e).__name__} - {str(e)}" def list_tools(self) -> Dict[str, str]: """列出所有工具:为LLM提供可用工具清单""" return {name: tool.description for name, tool in self._tools.items()}

设计逻辑:工具注册中心是Agent的“工具箱管理员”。这里刻意避开@tool装饰器,因为装饰器会隐藏“注册时机”这个关键概念——工具是在模块导入时注册,还是在Agent初始化时注册?手写register()方法让你明确:ToolRegistry实例创建后,必须显式调用register()才能生效。execute()方法的json.dumps()调用也值得玩味:它把任意Python对象(dict、list、int)转为标准JSON字符串,确保LLM能稳定解析。有学员曾把datetime.now()直接传给LLM,结果OpenAI API返回InvalidRequestError,因为JSON不支持datetime类型——这个坑,只有手写时才会踩得刻骨铭心。

3.4 第四层:工具实现(tools/weather.py & tools/math.py)

# tools/weather.py import asyncio import aiohttp async def get_weather(city: str) -> Dict[str, Any]: """模拟天气API调用""" # 实际项目中替换为真实API,此处用固定数据演示 weather_data = { "北京": {"temperature": 25, "condition": "晴", "humidity": 45}, "上海": {"temperature": 28, "condition": "多云", "humidity": 72}, "广州": {"temperature": 32, "condition": "雷阵雨", "humidity": 85} } await asyncio.sleep(0.5) # 模拟网络延迟 return weather_data.get(city, {"error": "城市未找到"}) # tools/math.py import asyncio async def calculate(expression: str) -> Dict[str, Any]: """安全计算数学表达式""" try: # 仅允许数字、+-*/和括号,防止代码注入 allowed_chars = set("0123456789+-*/(). ") if not all(c in allowed_chars for c in expression): return {"error": "表达式包含非法字符"} # 用eval前先校验(生产环境应换为ast.literal_eval) result = eval(expression) return {"result": float(result)} except ZeroDivisionError: return {"error": "除零错误"} except Exception as e: return {"error": f"计算错误: {str(e)}"}

设计逻辑:两个工具实现展示了“安全边界”的设计哲学。天气工具用asyncio.sleep(0.5)模拟真实网络延迟,这迫使你在主循环中必须用await,否则会阻塞整个Agent。数学工具的eval()调用更是教科书级的风险控制案例:先用字符集白名单过滤输入,再用try/except捕获所有异常。有学员曾天真地写eval(user_input),结果用户输入__import__('os').system('rm -rf /')直接清空了测试服务器——这个教训,比十页安全文档都管用。

3.5 第五层:大模型调用封装(core/llm.py)

# core/llm.py import asyncio import aiohttp import os from typing import List, Dict, Any # 从环境变量读取API Key,避免硬编码 API_KEY = os.getenv("OPENAI_API_KEY", "your-key-here") BASE_URL = "https://api.openai.com/v1/chat/completions" async def call_llm(messages: List[Dict[str, str]]) -> str: """调用OpenAI API:暴露所有关键参数""" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}" } payload = { "model": "gpt-3.5-turbo", "messages": messages, "temperature": 0.3, # 降低随机性,提高确定性 "max_tokens": 512, "top_p": 1.0, "frequency_penalty": 0.0, "presence_penalty": 0.0 } try: async with aiohttp.ClientSession() as session: async with session.post(BASE_URL, headers=headers, json=payload) as response: if response.status != 200: error_text = await response.text() raise Exception(f"API调用失败: {response.status} - {error_text}") data = await response.json() return data["choices"][0]["message"]["content"] except asyncio.TimeoutError: raise Exception("LLM调用超时,请检查网络") except Exception as e: raise Exception(f"LLM调用异常: {e}")

设计逻辑:LLM调用封装是“可控性”的集中体现。temperature=0.3这个参数选择背后有深意:太高(如0.8)会导致LLM规划结果飘忽不定,同一问题多次返回不同tool_name;太低(如0.0)又会让LLM拒绝处理模糊请求。我们通过实测发现0.3是平衡“稳定性”和“灵活性”的黄金点。aiohttp而非requests的选择,是为了与Agent整体异步架构对齐——如果这里用同步HTTP库,整个run()方法就必须用loop.run_in_executor()包装,徒增复杂度。环境变量OPENAI_API_KEY的使用,则是工程规范的第一课:密钥绝不硬编码。

3.6 第六层:对话记忆管理(core/memory.py)

# core/memory.py from typing import List, Dict, Any class ConversationMemory: def __init__(self, max_history: int = 10): self.max_history = max_history self._history: List[Dict[str, str]] = [] def add_interaction(self, user_input: str, agent_response: str): """添加一轮对话:显式管理历史长度""" self._history.append({"role": "user", "content": user_input}) self._history.append({"role": "assistant", "content": agent_response}) # 超过最大长度时,删除最老的两轮(user+assistant) while len(self._history) > self.max_history * 2: self._history.pop(0) self._history.pop(0) def get_history(self) -> List[Dict[str, str]]: """获取历史记录:返回副本,避免外部修改""" return self._history.copy() def clear(self): """清空记忆:为调试提供便利""" self._history.clear()

设计逻辑:记忆管理模块揭示了“状态管理”的本质。add_interaction()方法每次添加两行(user+assistant),而pop(0)删除最老的两行,确保历史严格按轮次截断。这里没有用deque,因为list.pop(0)的O(n)复杂度在10轮以内可忽略,且list更直观。get_history().copy()返回副本是关键防护:如果直接返回self._history,外部代码修改列表会污染Agent内部状态——这个细节,只有手写时才会意识到“可变对象引用”的危险性。

3.7 第七层:工具描述与Prompt工程(prompts/system_prompt.txt)

你是一个有用的AI助手。请根据用户问题,选择合适的工具执行,并用自然语言回答。 可用工具: - weather: 查询指定城市的天气信息,参数:{"city": "城市名称"} - math: 计算数学表达式,参数:{"expression": "算术表达式,如'3+5*2'"} 请严格按以下JSON格式输出,不要任何额外文字: {"tool": "工具名", "args": {"参数名": "参数值"}}

设计逻辑:System Prompt是Agent的“宪法”,它定义了行为边界。这里强调“严格按JSON格式输出”,是因为LLM有“自由发挥”倾向——不加约束时,它可能返回我将为您查询上海天气...。我们用“不要任何额外文字”强力压制,再配合代码层的json.loads()解析,形成双重保险。工具描述中的{"city": "城市名称"}示例,比抽象的{"city": "string"}更有效,因为LLM对具体例子的理解远超类型声明。有学员测试发现,去掉示例后,LLM调用weather工具时args字段经常为空——这证明Prompt工程不是玄学,而是可量化的精度控制。

4. 从“调包侠”到“造轮子的人”:实操中必须跨过的三道坎

手写Agent的过程,本质是和自身认知惯性搏斗。我带过的学员,90%会在三个节点遭遇强烈挫败感,这不是能力问题,而是思维范式切换的必经之痛。下面复盘这三道坎的真实场景、破局思路和避坑口诀。

4.1 坎一:LLM输出“不听话”——从期待确定性到拥抱概率性

典型场景:学员A写了完美Prompt,要求LLM返回{"tool":"weather","args":{"city":"北京"}},但实际收到{"tool": "weather", "args": {"city": "北京"}, "confidence": 0.95}。json.loads()直接报错,Agent崩溃。他反复修改Prompt,加“不要返回confidence字段”,LLM下次又返回{"tool": "weather", "args": {"city": "北京"}, "reason": "用户问北京天气"}。

破局思路:放弃“让LLM输出绝对干净JSON”的幻想,转向“鲁棒解析”。核心是理解LLM本质是概率模型,它的输出是token序列的概率分布,而非确定性指令。解决方案分三层:

  • 前端防御:用正则提取关键字段。例如tool_pattern = r'"tool"\s*:\s*"(\w+)"',args_pattern = r'"args"\s*:\s*(\{.*?\})',即使JSON结构乱,只要关键字存在就能捕获;
  • 中端容错:json.loads()前先用json.dumps(json.loads(...))标准化,或用jsonpatch库修复常见格式错误;
  • 后端兜底:当所有解析失败,返回{"tool": "fallback", "args": {}},由fallback工具返回友好提示“抱歉,我没理解您的意思,请换种说法”。

避坑口诀:“LLM不是数据库,别指望它吐标准JSON;你是指挥官,不是抄写员——抓关键词,放结构。”

4.2 坎二:工具链“掉链子”——从信任API到掌控全链路

典型场景:学员B接入真实天气API,发现aiohttp调用偶尔返回ClientConnectorError(网络抖动),但Agent无感知,直接卡死在await execute()。他加了try/except,却把异常吃掉,用户界面只显示空白。更糟的是,当API返回{"code": 503, "msg": "服务不可用"},他的解析器因data字段不存在而抛KeyError,整个Agent进程退出。

破局思路:建立“工具健康度”监控体系。每个工具执行必须携带三个元数据:

  • 状态码:HTTP状态码或自定义错误码(如TOOL_TIMEOUT);
  • 耗时:time.time()打点,超2s标为慢工具;
  • 结果质量:对返回JSON做schema校验,缺失关键字段则降级为{"error": "invalid_response"}。

在ToolRegistry.execute()中,我们加入熔断逻辑:

if self._failures.get(name, 0) > 3: # 连续失败3次 return {"error": "工具暂不可用,请稍后再试"} try: result = await self._tools[name].execute(**args) self._failures[name] = 0 # 成功则清零计数 except Exception as e: self._failures[name] = self._failures.get(name, 0) + 1 raise

避坑口诀:“工具不是永动机,要给它心跳检测;异常不是bug,是系统在报警——记录、告警、降级,三步不能少。”

4.3 坎三:历史“滚雪球”——从忽略状态到敬畏上下文

典型场景:学员C的Agent运行20轮后开始变慢,get_history()返回的messages列表超4000 tokens,LLM调用超时。他尝试删历史,但粗暴self._history = []导致用户说“刚才查的北京天气呢”,Agent一脸懵。更隐蔽的问题是,中文对话中"你好"和"您好"被当作不同上下文,记忆冗余度飙升。

破局思路:实施“分层记忆”策略:

  • 短期记忆(最近3轮):完整保留user+assistant消息,用于连贯对话;
  • 中期记忆(3-10轮):只存摘要,如{"summary": "用户查询了北京和上海天气,均显示晴天"};
  • 长期记忆(>10轮):存关键事实,如{"fact": "用户家乡是上海"},用向量数据库检索。

我们手写一个轻量摘要器:

def summarize_conversation(messages: List[Dict]) -> str: # 提取所有user消息中的名词短语(用jieba分词+词性标注) nouns = [] for msg in messages: if msg["role"] == "user": words = jieba.lcut(msg["content"]) for word, pos in pseg.cut("".join(words)): if pos in ["ns", "nz", "n", "vn"]: # 地名、专有名词、名词、动名词 nouns.append(word) # 去重后取前5个高频词 counter = Counter(nouns) top_nouns = [word for word, _ in counter.most_common(5)] return f"用户讨论了:{'、'.join(top_nouns)}"

避坑口诀:“记忆不是硬盘,是大脑——要摘要、要分类、要遗忘;Token不是钱,是命,超了就死。”

5. 常见问题与排查技巧实录:来自37次debug现场的血泪总结

手写Agent过程中,有些问题像幽灵一样反复出现。以下是我在37次学员debug现场记录的真实问题、排查路径和终极解法。它们不来自文档,而来自键盘敲击声、报错红字和深夜的咖啡渍。

问题现象排查路径终极解法避坑等级
LLM返回纯文本,非JSON1. 打印llm_response原始字符串
2. 检查Prompt是否含“严格JSON”指令
3. 观察是否在多轮对话后出现(历史污染)
在call_llm()后加清洗函数:
cleaned = re.sub(r'^.*?({.*}).*?$', r'\1', llm_response, flags=re.DOTALL)
再json.loads(cleaned)
⭐⭐⭐⭐⭐
工具执行后Agent无响应1. 在execute()开头加print(f"[DEBUG] Executing {name}")
2. 检查await是否遗漏(同步函数误当异步调用)
3. 查asyncio.current_task()确认事件循环状态
强制所有工具函数签名async def xxx(...) -> Any:,并在ToolRegistry.register()时校验inspect.iscoroutinefunction(func)⭐⭐⭐⭐
中文乱码导致LLM理解错误1.print(repr(history))看字节码
2. 检查open()文件读取是否指定encoding="utf-8"
3. 查json.dumps()是否含ensure_ascii=False
全局统一编码:文件读写用open(..., encoding="utf-8"),JSON序列化必加ensure_ascii=False,HTTP请求头加"Accept-Charset": "utf-8"⭐⭐⭐⭐⭐
Agent响应越来越慢1.time.time()打点各环节耗时
2. 检查ConversationMemory长度是否指数增长
3. 查aiohttp连接池是否耗尽
实施连接池复用:
connector = aiohttp.TCPConnector(limit=100, keepalive_timeout=30)
在call_llm()中复用ClientSession(connector=connector)
⭐⭐⭐⭐
多用户并发时状态错乱1. 检查SimpleAgent是否为单例(全局共享memory)
2. 查input()是否阻塞其他协程
3. 观察self._history是否被多个用户共用
改为class SessionAgent:,每个用户会话创建独立实例;主循环用asyncio.create_task()并发处理多输入⭐⭐⭐⭐⭐

独家技巧1:用“回声测试”定位LLM解析失效点
当LLM输出异常,不要猜,用最小化测试:

# test_echo.py messages = [{"role": "user", "content": "请返回JSON:{'tool':'weather','args':{'city':'北京'}}"}] response = await call_llm(messages) print("Raw:", repr(response)) # 看原始字节 print("Len:", len(response)) # 判断是否被截断

如果repr()显示'{"tool":"weather","args":{"city":"北京"}}\x00',说明LLM返回了null字节,需在解析前response.strip('\x00')。

独家技巧2:给每个工具加“健康检查”端点
在ToolRegistry中增加:

async def health_check(self) -> Dict[str, str]: results = {} for name in self._tools: try: # 用最小参数调用工具 if name == "weather": result = await self.execute(name, {"city": "北京"}) elif name == "math": result = await self.execute(name, {"expression": "1+1"}) results[name] = "OK" except Exception as e: results[name] = f"ERROR: {e}" return results

启动Agent后访问/health,立刻知道哪个工具挂了——这比等用户报错快10倍。

独家技巧3:用“时间旅行”调试历史污染
当怀疑历史记录导致LLM幻觉,临时修改get_history():

def get_history(self) -> List[
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进