1. 项目概述为什么我们要亲手“捏”一个AI Agent最近几个月AI Agent这个概念火得不行几乎成了技术圈和产品圈的“显学”。你可能在各种地方都看到过这个词但说实话很多讨论都停留在概念层面或者直接甩给你一个庞大的开源框架让人望而生畏。作为一个在AI应用层折腾了挺久的人我总觉得不亲手从零开始“捏”一个能跑起来的Agent就很难真正理解它的筋骨和脉络。这就好比学开车光看说明书不行你得真的坐进驾驶室点火、挂挡、踩油门感受那个反馈过程。所以今天我们不谈那些宏大的架构和晦涩的理论就聚焦一个非常具体的目标手写一个能理解你的指令、调用工具Function Calling、并自动串联多个步骤任务链来完成复杂任务的AI Agent。这个Agent的核心逻辑我们将用最直观的Python代码来实现避开那些一开始就让人头晕的抽象层。你会发现从Function Calling到任务链其核心思想远比想象中要朴素和清晰。这不仅是学习更是一次“祛魅”的过程——当你亲手实现后就会明白那些听起来高大上的框架底层到底在解决什么问题。这个项目适合谁呢首先当然是希望对AI Agent有实操理解的开发者无论你是前端、后端还是算法岗。其次是那些有产品想法想快速验证Agent能力能否解决某个场景需求的产品经理或创业者。最后哪怕你只是个技术爱好者这个过程也能帮你建立起对当前AI应用开发最核心范式指令理解、工具调用、规划执行的直观认知。我们不会使用特别复杂的库核心将围绕与大语言模型LLM的API交互展开所以只要你有基本的Python编程能力和一个能调用的大模型API比如OpenAI、DeepSeek、智谱等就可以跟着一起动手。2. 核心设计思路化繁为简的Agent构建哲学在开始写代码之前我们必须先统一思想我们要构建的Agent其核心工作流是什么市面上很多框架会引入“记忆”、“反思”、“技能库”等复杂概念但在最简版本里我们可以将其抽象为一个循环“解析用户意图 - 规划/选择工具 - 执行工具 - 整合结果 - 返回给用户或进入下一轮”这个循环的驱动力就是大语言模型LLM。LLM在这里扮演着“大脑”或“决策中心”的角色但它不能直接操作外部世界比如查询数据库、发送邮件、计算数学题。它需要通过“工具”Tools来延伸自己的能力。而“Function Calling”就是LLM与工具之间约定好的、结构化的通信协议。2.1 为什么是Function CallingFunction Calling并不是一个魔法黑盒。简单来说它允许我们以JSON Schema的形式向LLM描述一个工具这个工具叫什么名字是干什么用的需要哪些参数每个参数的名字、类型、描述然后当我们把用户的问题和这些工具的描述一起交给LLM时LLM可能会回复说“哦要解决这个问题我需要调用那个叫‘get_weather’的工具并且参数应该设为{“city”: “北京”}。”这个过程的精妙之处在于标准化和结构化。LLM的输出不再是自由发挥的文本而是一个结构化的JSON对象我们的程序可以轻松地解析这个JSON找到需要调用的函数名和参数然后用Python真的去调用对应的函数拿到结果。这解决了早期AI应用开发中一个巨大的痛点如何从LLM自由文本的输出中稳定、可靠地提取出可程序化执行的指令。2.2 任务链Task Chain是如何形成的单个工具调用只能完成简单任务。比如“北京天气怎么样”对应一次get_weather调用。但用户的需求往往是复杂的、多步骤的。例如“帮我查一下北京明天的天气如果下雨就搜索一下附近的室内体育馆并把第一个的结果摘要发给我。”这个任务无法通过一次工具调用完成。它需要调用天气工具获取北京明天的天气。根据结果是否下雨决定下一步行动。如果下雨调用搜索工具查询“北京 室内体育馆”。对搜索结果进行处理提取第一个的摘要。将摘要返回给用户。这里的关键在于第2步决策。这个决策同样由LLM来做。我们把第一个工具执行的结果连同原始的用户问题再次提交给LLM并问它“基于当前的结果和最初的目标下一步应该做什么是继续调用工具还是可以结束任务并给出最终答案” LLM根据所有上下文决定调用下一个工具或者生成最终回复。这个“执行-观察-再规划”的循环就构成了一个自动化的任务链。任务链的复杂性取决于LLM规划能力的高低和我们提供给它的工具集的丰富程度。我们的设计将紧紧围绕这个“循环”展开。我们会先实现最基础的Function Calling然后让这个循环能够迭代起来形成任务链。最后我们会讨论如何让这个系统更健壮、更实用。3. 基础搭建实现核心的Function Calling机制让我们从最核心的部分开始如何让LLM学会“打电话”Call Function。这里我们以OpenAI的Chat Completion API为例其他厂商的API在思路上大同小异。3.1 定义工具用JSON Schema描述你的能力首先我们需要用LLM能理解的方式告诉它它手头有哪些“工具”可用。这通过一个“工具列表”tools参数来实现列表中的每个工具都是一个字典包含工具的名称、描述和参数模式。假设我们有两个简单的工具get_current_time: 获取当前时间。calculator: 执行一个数学计算。# 工具定义 tools [ { type: function, function: { name: get_current_time, description: 获取当前的日期和时间, parameters: { type: object, properties: {}, # 这个工具不需要参数 required: [] } } }, { type: function, function: { name: calculator, description: 执行一个数学计算支持加()、减(-)、乘(*)、除(/), parameters: { type: object, properties: { expression: { type: string, description: 数学表达式例如3 5 * 2 } }, required: [expression] } } } ]注意description字段至关重要LLM完全依靠这个描述来判断在什么情况下该调用这个工具。描述要清晰、准确说明工具的用途和参数的意义。例如calculator的描述里明确了支持的运算符这能极大提高LLM调用的准确性。3.2 实现工具函数大脑的“手和脚”定义了工具说明书接下来就要实现工具本身。这些就是普通的Python函数。import datetime import math def get_current_time(): 返回当前时间字符串 now datetime.datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S) def calculator(expression: str) - str: 计算数学表达式。注意使用eval有安全风险此处仅用于演示。 try: # 警告在生产环境中直接使用eval是危险的可能执行恶意代码。 # 这里仅为演示真实场景应使用安全的表达式解析库如ast.literal_eval配合自定义运算符处理。 result eval(expression, {__builtins__: None}, {math: math}) return str(result) except Exception as e: return f计算错误{e}实操心得calculator函数里使用了eval这在实际项目中是绝对要避免的安全隐患。一个用户如果输入__import__(os).system(rm -rf /)后果不堪设想。演示代码为了简洁用了它但你一定要记住真实项目里必须使用更安全的方式比如ast.literal_eval解析数字和运算符或者使用专门的数学表达式库。3.3 与LLM对话并解析工具调用请求现在我们把用户问题、工具描述一起发给LLM并告诉它“你可以调用这些工具哦。”通过设置tool_choiceauto或tool_choicerequired。import openai import json # 假设你已经设置了OPENAI_API_KEY client openai.OpenAI() def chat_with_llm(messages, tools): 发送消息给LLM并允许其调用工具 response client.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4 messagesmessages, toolstools, tool_choiceauto, # 让模型自行决定是否调用工具 ) return response.choices[0].message # 初始化对话 messages [{role: user, content: 现在几点了}] llm_response_message chat_with_llm(messages, tools) print(fLLM的回复消息对象: {llm_response_message})运行后llm_response_message可能是一个普通的content回复比如“我是一个AI无法获取实时信息”但更可能是一个包含tool_calls属性的对象。tool_calls是一个列表里面包含了LLM想要调用的工具信息。# 检查是否有工具调用 if llm_response_message.tool_calls: print(LLM想要调用工具) for tool_call in llm_response_message.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) # 参数是JSON字符串 print(f工具名{func_name}, 参数{func_args}) # 根据工具名找到对应的本地函数并执行 if func_name get_current_time: result get_current_time() elif func_name calculator: result calculator(**func_args) # 解包参数字典 else: result f未知工具{func_name} print(f工具执行结果{result}) # 关键步骤将工具执行结果作为新的消息追加到对话历史中 messages.append(llm_response_message) # 追加LLM的请求消息 messages.append({ role: tool, tool_call_id: tool_call.id, # 必须匹配对应的tool_call id content: result, }) # 有了工具执行结果后再次调用LLM让它基于结果生成面向用户的回复 final_response_message chat_with_llm(messages, tools) print(f最终回复{final_response_message.content}) else: # 没有工具调用直接输出内容 print(fLLM直接回复{llm_response_message.content})这段代码完成了第一次“循环”用户提问 - LLM决定调用get_current_time工具 - 我们执行工具得到“2023-10-27 14:30:00” - 我们将结果以role: tool的身份反馈给LLM - LLM消化这个结果生成最终的用户回复“当前时间是2023年10月27日 14:30:00”。这就是Function Calling最核心的闭环。tool_call_id的对应关系确保了在多工具调用场景下LLM能分清哪个结果是哪个工具返回的。4. 从单次调用到任务链构建自动化循环实现了单次工具调用我们已经让AI有了“手”。接下来我们要赋予它“多步思考”和“持续执行”的能力也就是任务链。4.1 设计一个可持续运行的Agent循环任务链的本质是让上面那个“调用-执行-反馈”的循环持续进行下去直到LLM认为任务完成主动输出最终答案为止。我们需要一个while循环来驱动这个过程。def run_agent(user_query, tools, max_steps10): 运行一个简单的Agent。 :param user_query: 用户初始问题 :param tools: 可用工具列表 :param max_steps: 最大循环步数防止无限循环 :return: Agent的最终输出 messages [{role: user, content: user_query}] step 0 while step max_steps: step 1 print(f\n--- 第 {step} 步 ---) # 1. 调用LLM获取决策可能是回复也可能是工具调用 llm_message chat_with_llm(messages, tools) messages.append(llm_message) # 将LLM的回应加入历史 # 2. 检查是否为最终回复 if not llm_message.tool_calls: print(Agent决定结束任务。) return llm_message.content # 3. 执行所有被请求的工具调用 for tool_call in llm_message.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) print(f执行工具: {func_name}, 参数: {func_args}) # 这里应该有一个从工具名到实际函数的映射为了清晰我们用if-else if func_name get_current_time: tool_result get_current_time() elif func_name calculator: tool_result calculator(**func_args) else: tool_result fError: 未知工具 {func_name} # 4. 将工具执行结果反馈给LLM messages.append({ role: tool, tool_call_id: tool_call.id, content: str(tool_result), # 确保内容是字符串 }) print(f工具结果: {tool_result}) # 如果达到最大步数仍未结束 return f任务未在{max_steps}步内完成可能陷入循环。最新消息{messages[-1].get(content, N/A)}现在让我们用一个需要多步的任务来测试它。我们需要增加一个工具比如一个简单的网络搜索模拟工具。# 新增一个模拟搜索工具 tools.append({ type: function, function: { name: search_web, description: 根据关键词模拟网络搜索返回模拟结果。, parameters: { type: object, properties: { query: { type: string, description: 搜索关键词 } }, required: [query] } } }) def search_web(query: str) - str: 模拟搜索返回固定格式的假数据 # 这里只是一个模拟。真实情况你会调用SerpAPI、Google Search API等。 mock_results [ f关于{query}的百科介绍模拟结果1, f{query}的最新新闻报道模拟结果2, f讨论{query}的技术论坛帖子模拟结果3, ] return \n---\n.join(mock_results) # 测试一个复杂任务 final_answer run_agent( user_query请先计算一下(15 7) * 2等于多少然后基于这个数字搜索一下相关的结果。, toolstools, max_steps5 ) print(f\n 最终答案 \n{final_answer})这个Agent会先调用calculator计算(157)*2得到44然后将这个结果或结合原问题作为上下文决定调用search_web搜索关键词可能是“44”或者“44 结果”。LLM会根据我们的工具描述和对话历史自主做出这个决策。这就形成了一个简单的两步骤任务链。4.2 处理复杂依赖与条件分支上面的例子是线性的。但真实场景中任务链可能有分支。比如我们开头的例子“查天气如果下雨则搜索室内体育馆”。这需要LLM根据中间结果做出判断。我们的run_agent循环已经天然支持了这种模式关键在于我们如何设计工具和描述。我们可以设计一个get_weather工具它返回的结构化信息中需要包含“天气状况”如“晴”、“雨”。LLM在收到这个结果后会结合最初的指令“如果下雨就搜索...”在下一步决策时它就会判断如果天气状况包含“雨”则调用search_web工具否则直接生成最终回复“天气晴好无需搜索室内场馆”。# 模拟天气工具 def get_weather(city: str) - str: # 模拟返回结构化数据用JSON字符串便于LLM解析 import random conditions [晴, 多云, 小雨, 大雨, 阴] condition random.choice(conditions) temp random.randint(15, 30) return json.dumps({city: city, condition: condition, temperature: temp, unit: 摄氏度}, ensure_asciiFalse) # 更新工具列表添加get_weather # ... (将get_weather的描述加入tools列表) # 测试条件任务 final_answer run_agent( user_query查询一下北京的天气如果下雨就搜索‘北京 室内游泳馆’。, toolstools, # 假设tools已包含get_weather和search_web max_steps5 )在这个循环中LLM在第一步调用get_weather(北京)拿到结果{condition: 小雨, ...}。在第二步LLM看到这个结果并回忆起用户指令中的条件“如果下雨”于是它判断需要调用search_web(北京 室内游泳馆)。整个决策逻辑完全由LLM根据上下文推导我们的代码只需要提供一个稳定的“执行-反馈”循环即可。注意事项这里有一个关键点为了让LLM更好地理解工具返回的结果以做出决策工具返回的内容最好也是结构化的比如JSON。虽然LLM能理解自然语言但结构化数据减少了歧义提高了后续步骤的可靠性。你可以让工具返回JSON字符串并在工具描述中说明返回值的结构。5. 工程化与优化让你的Agent更健壮一个能跑起来的原型和一个健壮的可用系统之间还有不少距离。下面我们来探讨几个关键的工程化优化点。5.1 工具管理的标准化用一堆if-elif来匹配工具名和函数显然不是长久之计。我们需要一个中心化的工具注册和管理机制。class ToolRegistry: def __init__(self): self._tools [] # 存放OpenAI格式的工具描述 self._functions {} # 存放工具名到实际函数的映射 def register(self, func, description: str, parameters: dict): 注册一个工具 tool_schema { type: function, function: { name: func.__name__, description: description, parameters: { type: object, properties: parameters.get(properties, {}), required: parameters.get(required, []) } } } self._tools.append(tool_schema) self._functions[func.__name__] func return func # 方便用作装饰器 def get_tools_schema(self): return self._tools def execute(self, tool_name: str, **kwargs): 执行一个已注册的工具 if tool_name not in self._functions: raise ValueError(f工具未注册: {tool_name}) return self._functions[tool_name](**kwargs) # 使用装饰器注册工具更加优雅 registry ToolRegistry() registry.register( description获取当前的日期和时间, parameters{type: object, properties: {}, required: []} ) def get_current_time(): now datetime.datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S) registry.register( description执行数学计算支持加()、减(-)、乘(*)、除(/), parameters{ type: object, properties: { expression: {type: string, description: 数学表达式如 3 5 * 2} }, required: [expression] } ) def calculator(expression: str): # ... (安全实现) pass # 在run_agent中工具执行部分变为 # tool_result registry.execute(func_name, **func_args)这样工具的管理变得清晰、可扩展。新增工具只需用registry.register装饰一下即可。5.2 对话历史管理与上下文长度LLM有上下文窗口限制如4K、8K、128K tokens。我们的messages列表会随着对话轮次和工具调用结果不断增长最终可能超出限制。我们需要一个策略来管理历史。简单截断只保留最近N条消息或确保总token数不超过阈值。缺点是可能丢失关键的前期指令。摘要压缩定期用LLM对之前的对话历史进行总结将冗长的历史压缩成一段简短的摘要然后用“摘要近期消息”作为新的上下文。这是更高级和实用的策略但实现稍复杂。关键信息提取对于工具返回的冗长结果如搜索返回的10个网页摘要可以先用LLM提取与当前任务最相关的几句只把精华放入上下文。在原型阶段我们可以先采用简单截断并密切关注token消耗。OpenAI的API返回中通常包含usage字段可以据此进行监控。5.3 错误处理与稳定性一个健壮的Agent必须能处理各种异常。工具执行错误网络超时、API调用失败、参数错误等。我们的工具函数应该用try...except捕获异常并返回清晰的错误信息如“Error: 计算服务暂时不可用”。这个错误信息会被反馈给LLMLLM有可能尝试其他方法或向用户报告问题。LLM输出格式错误LLM可能返回不符合tool_calls格式的内容或者参数不符合JSON Schema。我们需要在解析tool_call.function.arguments时做好json.loads的异常捕获并给予降级处理例如提示LLM重新生成。循环检测Agent可能陷入死循环反复调用相同的工具。除了设置max_steps硬性限制还可以检测最近几步的行动是否重复如果重复则中断并提示。结果验证对于关键操作如发送邮件、修改数据库在执行前可以增加一个确认环节例如让LLM生成一个操作摘要经用户确认或另一套安全规则确认后再执行。# 在工具执行环节增加错误处理 try: tool_result registry.execute(func_name, **func_args) except Exception as e: tool_result f工具执行失败: {str(e)} # 可以选择将严重错误记录日志甚至终止Agent运行5.4 思维链Chain-of-Thought与规划能力提升要让Agent处理更复杂的任务有时需要提升其规划能力。除了依赖LLM自身的推理我们还可以在提示词Prompt上做文章。在系统提示System Prompt中明确角色和流程例如“你是一个任务规划助手。请逐步思考每次只调用一个最必要的工具。根据工具返回的结果决定下一步是继续调用工具还是给出最终答案。”要求LLM输出思考过程在user或assistant的消息中要求LLM先以“思考...”的形式输出推理再决定行动。虽然OpenAI的Function Calling API不直接支持在tool_calls同时输出content但你可以通过设计对话流程来模拟比如先让LLM以普通文本输出计划然后再在下一轮中调用工具。分解复杂指令对于非常复杂的用户请求可以设计一个“任务分解”工具让LLM先将大任务拆解成清晰的子任务列表然后再逐个执行。这相当于实现了一个简单的规划模块。6. 实战构建一个多技能的个人助理Agent现在让我们把上面的所有部分组合起来构建一个稍微丰富一点的个人助理Agent它具备查询时间、计算、模拟搜索、查询天气、甚至记录备忘录模拟的能力。import json import datetime import random from typing import Dict, List class PersonalAssistantAgent: def __init__(self, llm_client, system_promptNone): self.client llm_client self.messages [] if system_prompt: self.messages.append({role: system, content: system_prompt}) self.tool_registry ToolRegistry() self._register_builtin_tools() def _register_builtin_tools(self): 注册内置工具 self.tool_registry.register( description获取当前日期和时间, parameters{type: object, properties: {}, required: []} ) def get_current_time(): return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) self.tool_registry.register( description执行数学计算支持加减乘除及math库常用函数, parameters{ type: object, properties: { expression: {type: string, description: 数学表达式如 sqrt(9) 5 * 2} }, required: [expression] } ) def calculator(expression: str): # 使用一个更安全的计算方式示例仅支持有限操作 # 实际应用中应使用更完善的库如 numexpr 或自定义安全解析器 allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} allowed_names.update({abs: abs, round: round}) try: # 警告此方法仍非绝对安全仅用于演示 compiled_expr compile(expression, string, eval) for name in compiled_expr.co_names: if name not in allowed_names: raise ValueError(f禁止使用的名称: {name}) result eval(compiled_expr, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误: {e} self.tool_registry.register( description模拟网络搜索返回几条模拟结果摘要, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词} }, required: [query] } ) def search_web(query: str): mock_data { Python: [Python是一种高级编程语言。, 以简洁易读著称。, 广泛用于Web开发和数据分析。], 天气: [天气是大气状态的短期变化。, 通常用温度、湿度、降水等描述。, 天气预报依赖于气象模型。], AI: [人工智能是模拟人类智能的技术。, 包括机器学习、深度学习等领域。, 正在改变许多行业。] } for key, results in mock_data.items(): if key.lower() in query.lower(): return \n.join([f{i1}. {r} for i, r in enumerate(results[:3])]) return f未找到与{query}高度相关的模拟结果。这里是一些通用信息...模拟 self.tool_registry.register( description查询指定城市的模拟天气信息, parameters{ type: object, properties: { city: {type: string, description: 城市名称如北京、上海} }, required: [city] } ) def get_weather(city: str): conditions [晴, 多云, 阴, 小雨, 中雨, 大雨, 雪] condition random.choice(conditions) temp random.randint(0, 35) return json.dumps({ city: city, condition: condition, temperature: temp, unit: 摄氏度, tip: 记得带伞 if 雨 in condition else 天气不错 }, ensure_asciiFalse) # 一个简单的“备忘录”模拟存储在内存中 self.notes: List[Dict] [] self.tool_registry.register( description添加一条文本备忘录, parameters{ type: object, properties: { note: {type: string, description: 备忘录内容} }, required: [note] } ) def add_note(note: str): note_id len(self.notes) 1 self.notes.append({id: note_id, content: note, time: datetime.datetime.now().isoformat()}) return f已添加备忘录 (ID: {note_id}): {note} self.tool_registry.register( description列出所有备忘录, parameters{type: object, properties: {}, required: []} ) def list_notes(): if not self.notes: return 当前没有备忘录。 return \n.join([f{n[id]}. [{n[time]}] {n[content]} for n in self.notes]) def chat(self, user_input: str, max_turns8): 与Agent进行一轮对话可能触发多步工具调用 self.messages.append({role: user, content: user_input}) for turn in range(max_turns): # 调用LLM response self.client.chat.completions.create( modelgpt-3.5-turbo, messagesself.messages, toolsself.tool_registry.get_tools_schema(), tool_choiceauto, ) msg response.choices[0].message self.messages.append(msg) # 如果没有工具调用返回最终回复 if not msg.tool_calls: return msg.content # 执行工具调用 for tool_call in msg.tool_calls: func_name tool_call.function.name try: func_args json.loads(tool_call.function.arguments) except json.JSONDecodeError: tool_result 错误工具参数格式无效。 else: try: tool_result self.tool_registry.execute(func_name, **func_args) except Exception as e: tool_result f工具{func_name}执行出错: {str(e)} # 将结果反馈给LLM self.messages.append({ role: tool, tool_call_id: tool_call.id, content: str(tool_result), }) return 对话轮次已达上限请简化您的问题。 # 使用示例 if __name__ __main__: client openai.OpenAI(api_keyyour-api-key) # 请替换为你的API Key system_prompt 你是一个乐于助人的个人助理。请逐步思考利用可用工具解决问题。 每次尽量只调用一个最必要的工具。根据工具返回的结果决定下一步是继续调用工具还是给出友好、清晰的最终答案。 agent PersonalAssistantAgent(client, system_prompt) queries [ 现在几点了顺便帮我计算一下(12.5 4.3) * 2 等于多少。, 北京天气怎么样如果下雨就搜索一下‘室内活动推荐’。, 帮我记一下明天下午三点有个团队会议。然后再列出我所有的备忘录。, ] for query in queries: print(f\n用户: {query}) answer agent.chat(query) print(f助理: {answer}) print(- * 40) # 重置对话历史避免上下文过长影响下一个独立问题 agent.messages [agent.messages[0]] if agent.messages else []这个PersonalAssistantAgent类集成了工具注册、对话管理、错误处理等核心功能。你可以通过registry.register装饰器轻松扩展新的工具。它已经能够处理包含多个步骤和条件判断的复杂查询。7. 常见问题与排查技巧实录在实际开发和调试手写Agent的过程中你肯定会遇到各种各样的问题。下面是我踩过的一些坑和总结的排查思路。7.1 LLM不调用工具症状无论你怎么问LLM都只用自然语言回答从不触发tool_calls。排查检查工具描述这是最常见的原因。description是否清晰、准确地描述了工具的功能和适用场景LLM是根据描述来判断是否调用的。试着把描述写得更具体、更具指向性。检查tool_choice参数你设置成“auto”了吗如果设成“none”LLM就不会调用工具。“required”会强制LLM调用工具但可能导致它调用不合适的工具。检查系统提示系统提示system消息是否鼓励或要求LLM使用工具例如可以加上“请优先使用我提供的工具来获取信息或执行操作。”问题是否太简单对于“你好吗”这种问题LLM认为无需工具即可回答。尝试问一个明确需要外部信息的问题如“纽约现在几点”模型能力确保你使用的模型支持Function Calling。GPT-3.5-turbo和GPT-4系列都支持。7.2 工具调用参数错误症状LLM调用了工具但参数不对比如类型错误、缺少必填参数、或者参数值毫无意义。排查细化参数描述在parameters的properties里为每个参数提供清晰的description。例如“city”参数可以描述为“城市中文名如‘北京市’、‘上海市’”。提供示例在description或对话上下文中通过示例展示参数的格式。LLM会学习这些模式。在系统提示中约束可以在系统提示中说明“请确保为工具调用提供准确、完整的参数”。后置校验与重试在代码中校验参数如果发现明显错误如城市名称为空可以将错误信息反馈给LLM并要求它重新生成调用。这相当于一个简单的自我修正循环。7.3 Agent陷入死循环或逻辑混乱症状Agent反复调用同一个工具或者在几个工具间无效切换无法达成目标。排查与解决设置最大步数这是最基本的防护如我们的max_steps或max_turns。检查工具返回结果工具返回的结果是否清晰、结构化一个混乱或冗长的结果可能导致LLM无法理解进而做出错误决策。尽量让工具返回简洁、关键的信息。增强系统提示的规划性在系统提示中明确要求“逐步思考”、“先制定计划”、“每次只解决当前最核心的子问题”。这能一定程度上提升LLM的规划能力。引入“任务状态”跟踪对于复杂任务可以设计一个简单的状态机或者让LLM在每一步后输出当前任务的完成状态摘要帮助它保持方向。7.4 上下文长度爆炸症状对话进行到后面LLM回复变慢、成本增高甚至开始遗忘最初的指令。解决策略定期摘要这是最有效的策略。每进行若干步或当上下文token数接近阈值时用LLM对之前的对话历史特别是早期的用户指令和关键中间结果进行总结生成一段简短的“背景摘要”然后用这个摘要替换掉旧的长篇历史。选择性记忆只将与当前步骤最相关的几条历史消息保留在上下文中。这需要你定义“相关性”实现起来较复杂。使用长上下文模型如果成本允许直接使用支持128K或更长上下文的模型如GPT-4 Turbo。但这只是缓解不是根治。7.5 安全与成本控制成本每次工具调用和LLM回复都消耗token。复杂任务链可能消耗大量token。务必在代码中记录和监控usage并为API设置用量告警或硬性限制。工具安全再次强调永远不要相信来自LLM的未经验证的输入去执行危险操作。像calculator中使用eval是极端危险的示例。对于文件操作、数据库查询、系统命令等必须进行严格的输入验证、权限检查和沙箱隔离。内容安全LLM可能被诱导生成或工具可能返回不受控制的内容。需要对最终输出给用户的内容进行必要的过滤和审查尤其是在面向公众的应用中。手写一个AI Agent的过程是一个不断与LLM“沟通”和“协作”的过程。你通过工具描述和系统提示来塑造它的行为它通过推理和调用工具来解决问题。这个过程充满了挑战但也极具乐趣和启发性。当你看到几行简单的代码串联起一个能自动完成多步任务的智能体时那种成就感是无可替代的。希望这篇长文能为你打下坚实的基础让你有能力去探索更广阔的AI Agent世界。 从零手写AI Agent：基于Function Calling与任务链的智能体构建实践
1. 项目概述为什么我们要亲手“捏”一个AI Agent最近几个月AI Agent这个概念火得不行几乎成了技术圈和产品圈的“显学”。你可能在各种地方都看到过这个词但说实话很多讨论都停留在概念层面或者直接甩给你一个庞大的开源框架让人望而生畏。作为一个在AI应用层折腾了挺久的人我总觉得不亲手从零开始“捏”一个能跑起来的Agent就很难真正理解它的筋骨和脉络。这就好比学开车光看说明书不行你得真的坐进驾驶室点火、挂挡、踩油门感受那个反馈过程。所以今天我们不谈那些宏大的架构和晦涩的理论就聚焦一个非常具体的目标手写一个能理解你的指令、调用工具Function Calling、并自动串联多个步骤任务链来完成复杂任务的AI Agent。这个Agent的核心逻辑我们将用最直观的Python代码来实现避开那些一开始就让人头晕的抽象层。你会发现从Function Calling到任务链其核心思想远比想象中要朴素和清晰。这不仅是学习更是一次“祛魅”的过程——当你亲手实现后就会明白那些听起来高大上的框架底层到底在解决什么问题。这个项目适合谁呢首先当然是希望对AI Agent有实操理解的开发者无论你是前端、后端还是算法岗。其次是那些有产品想法想快速验证Agent能力能否解决某个场景需求的产品经理或创业者。最后哪怕你只是个技术爱好者这个过程也能帮你建立起对当前AI应用开发最核心范式指令理解、工具调用、规划执行的直观认知。我们不会使用特别复杂的库核心将围绕与大语言模型LLM的API交互展开所以只要你有基本的Python编程能力和一个能调用的大模型API比如OpenAI、DeepSeek、智谱等就可以跟着一起动手。2. 核心设计思路化繁为简的Agent构建哲学在开始写代码之前我们必须先统一思想我们要构建的Agent其核心工作流是什么市面上很多框架会引入“记忆”、“反思”、“技能库”等复杂概念但在最简版本里我们可以将其抽象为一个循环“解析用户意图 - 规划/选择工具 - 执行工具 - 整合结果 - 返回给用户或进入下一轮”这个循环的驱动力就是大语言模型LLM。LLM在这里扮演着“大脑”或“决策中心”的角色但它不能直接操作外部世界比如查询数据库、发送邮件、计算数学题。它需要通过“工具”Tools来延伸自己的能力。而“Function Calling”就是LLM与工具之间约定好的、结构化的通信协议。2.1 为什么是Function CallingFunction Calling并不是一个魔法黑盒。简单来说它允许我们以JSON Schema的形式向LLM描述一个工具这个工具叫什么名字是干什么用的需要哪些参数每个参数的名字、类型、描述然后当我们把用户的问题和这些工具的描述一起交给LLM时LLM可能会回复说“哦要解决这个问题我需要调用那个叫‘get_weather’的工具并且参数应该设为{“city”: “北京”}。”这个过程的精妙之处在于标准化和结构化。LLM的输出不再是自由发挥的文本而是一个结构化的JSON对象我们的程序可以轻松地解析这个JSON找到需要调用的函数名和参数然后用Python真的去调用对应的函数拿到结果。这解决了早期AI应用开发中一个巨大的痛点如何从LLM自由文本的输出中稳定、可靠地提取出可程序化执行的指令。2.2 任务链Task Chain是如何形成的单个工具调用只能完成简单任务。比如“北京天气怎么样”对应一次get_weather调用。但用户的需求往往是复杂的、多步骤的。例如“帮我查一下北京明天的天气如果下雨就搜索一下附近的室内体育馆并把第一个的结果摘要发给我。”这个任务无法通过一次工具调用完成。它需要调用天气工具获取北京明天的天气。根据结果是否下雨决定下一步行动。如果下雨调用搜索工具查询“北京 室内体育馆”。对搜索结果进行处理提取第一个的摘要。将摘要返回给用户。这里的关键在于第2步决策。这个决策同样由LLM来做。我们把第一个工具执行的结果连同原始的用户问题再次提交给LLM并问它“基于当前的结果和最初的目标下一步应该做什么是继续调用工具还是可以结束任务并给出最终答案” LLM根据所有上下文决定调用下一个工具或者生成最终回复。这个“执行-观察-再规划”的循环就构成了一个自动化的任务链。任务链的复杂性取决于LLM规划能力的高低和我们提供给它的工具集的丰富程度。我们的设计将紧紧围绕这个“循环”展开。我们会先实现最基础的Function Calling然后让这个循环能够迭代起来形成任务链。最后我们会讨论如何让这个系统更健壮、更实用。3. 基础搭建实现核心的Function Calling机制让我们从最核心的部分开始如何让LLM学会“打电话”Call Function。这里我们以OpenAI的Chat Completion API为例其他厂商的API在思路上大同小异。3.1 定义工具用JSON Schema描述你的能力首先我们需要用LLM能理解的方式告诉它它手头有哪些“工具”可用。这通过一个“工具列表”tools参数来实现列表中的每个工具都是一个字典包含工具的名称、描述和参数模式。假设我们有两个简单的工具get_current_time: 获取当前时间。calculator: 执行一个数学计算。# 工具定义 tools [ { type: function, function: { name: get_current_time, description: 获取当前的日期和时间, parameters: { type: object, properties: {}, # 这个工具不需要参数 required: [] } } }, { type: function, function: { name: calculator, description: 执行一个数学计算支持加()、减(-)、乘(*)、除(/), parameters: { type: object, properties: { expression: { type: string, description: 数学表达式例如3 5 * 2 } }, required: [expression] } } } ]注意description字段至关重要LLM完全依靠这个描述来判断在什么情况下该调用这个工具。描述要清晰、准确说明工具的用途和参数的意义。例如calculator的描述里明确了支持的运算符这能极大提高LLM调用的准确性。3.2 实现工具函数大脑的“手和脚”定义了工具说明书接下来就要实现工具本身。这些就是普通的Python函数。import datetime import math def get_current_time(): 返回当前时间字符串 now datetime.datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S) def calculator(expression: str) - str: 计算数学表达式。注意使用eval有安全风险此处仅用于演示。 try: # 警告在生产环境中直接使用eval是危险的可能执行恶意代码。 # 这里仅为演示真实场景应使用安全的表达式解析库如ast.literal_eval配合自定义运算符处理。 result eval(expression, {__builtins__: None}, {math: math}) return str(result) except Exception as e: return f计算错误{e}实操心得calculator函数里使用了eval这在实际项目中是绝对要避免的安全隐患。一个用户如果输入__import__(os).system(rm -rf /)后果不堪设想。演示代码为了简洁用了它但你一定要记住真实项目里必须使用更安全的方式比如ast.literal_eval解析数字和运算符或者使用专门的数学表达式库。3.3 与LLM对话并解析工具调用请求现在我们把用户问题、工具描述一起发给LLM并告诉它“你可以调用这些工具哦。”通过设置tool_choiceauto或tool_choicerequired。import openai import json # 假设你已经设置了OPENAI_API_KEY client openai.OpenAI() def chat_with_llm(messages, tools): 发送消息给LLM并允许其调用工具 response client.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4 messagesmessages, toolstools, tool_choiceauto, # 让模型自行决定是否调用工具 ) return response.choices[0].message # 初始化对话 messages [{role: user, content: 现在几点了}] llm_response_message chat_with_llm(messages, tools) print(fLLM的回复消息对象: {llm_response_message})运行后llm_response_message可能是一个普通的content回复比如“我是一个AI无法获取实时信息”但更可能是一个包含tool_calls属性的对象。tool_calls是一个列表里面包含了LLM想要调用的工具信息。# 检查是否有工具调用 if llm_response_message.tool_calls: print(LLM想要调用工具) for tool_call in llm_response_message.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) # 参数是JSON字符串 print(f工具名{func_name}, 参数{func_args}) # 根据工具名找到对应的本地函数并执行 if func_name get_current_time: result get_current_time() elif func_name calculator: result calculator(**func_args) # 解包参数字典 else: result f未知工具{func_name} print(f工具执行结果{result}) # 关键步骤将工具执行结果作为新的消息追加到对话历史中 messages.append(llm_response_message) # 追加LLM的请求消息 messages.append({ role: tool, tool_call_id: tool_call.id, # 必须匹配对应的tool_call id content: result, }) # 有了工具执行结果后再次调用LLM让它基于结果生成面向用户的回复 final_response_message chat_with_llm(messages, tools) print(f最终回复{final_response_message.content}) else: # 没有工具调用直接输出内容 print(fLLM直接回复{llm_response_message.content})这段代码完成了第一次“循环”用户提问 - LLM决定调用get_current_time工具 - 我们执行工具得到“2023-10-27 14:30:00” - 我们将结果以role: tool的身份反馈给LLM - LLM消化这个结果生成最终的用户回复“当前时间是2023年10月27日 14:30:00”。这就是Function Calling最核心的闭环。tool_call_id的对应关系确保了在多工具调用场景下LLM能分清哪个结果是哪个工具返回的。4. 从单次调用到任务链构建自动化循环实现了单次工具调用我们已经让AI有了“手”。接下来我们要赋予它“多步思考”和“持续执行”的能力也就是任务链。4.1 设计一个可持续运行的Agent循环任务链的本质是让上面那个“调用-执行-反馈”的循环持续进行下去直到LLM认为任务完成主动输出最终答案为止。我们需要一个while循环来驱动这个过程。def run_agent(user_query, tools, max_steps10): 运行一个简单的Agent。 :param user_query: 用户初始问题 :param tools: 可用工具列表 :param max_steps: 最大循环步数防止无限循环 :return: Agent的最终输出 messages [{role: user, content: user_query}] step 0 while step max_steps: step 1 print(f\n--- 第 {step} 步 ---) # 1. 调用LLM获取决策可能是回复也可能是工具调用 llm_message chat_with_llm(messages, tools) messages.append(llm_message) # 将LLM的回应加入历史 # 2. 检查是否为最终回复 if not llm_message.tool_calls: print(Agent决定结束任务。) return llm_message.content # 3. 执行所有被请求的工具调用 for tool_call in llm_message.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) print(f执行工具: {func_name}, 参数: {func_args}) # 这里应该有一个从工具名到实际函数的映射为了清晰我们用if-else if func_name get_current_time: tool_result get_current_time() elif func_name calculator: tool_result calculator(**func_args) else: tool_result fError: 未知工具 {func_name} # 4. 将工具执行结果反馈给LLM messages.append({ role: tool, tool_call_id: tool_call.id, content: str(tool_result), # 确保内容是字符串 }) print(f工具结果: {tool_result}) # 如果达到最大步数仍未结束 return f任务未在{max_steps}步内完成可能陷入循环。最新消息{messages[-1].get(content, N/A)}现在让我们用一个需要多步的任务来测试它。我们需要增加一个工具比如一个简单的网络搜索模拟工具。# 新增一个模拟搜索工具 tools.append({ type: function, function: { name: search_web, description: 根据关键词模拟网络搜索返回模拟结果。, parameters: { type: object, properties: { query: { type: string, description: 搜索关键词 } }, required: [query] } } }) def search_web(query: str) - str: 模拟搜索返回固定格式的假数据 # 这里只是一个模拟。真实情况你会调用SerpAPI、Google Search API等。 mock_results [ f关于{query}的百科介绍模拟结果1, f{query}的最新新闻报道模拟结果2, f讨论{query}的技术论坛帖子模拟结果3, ] return \n---\n.join(mock_results) # 测试一个复杂任务 final_answer run_agent( user_query请先计算一下(15 7) * 2等于多少然后基于这个数字搜索一下相关的结果。, toolstools, max_steps5 ) print(f\n 最终答案 \n{final_answer})这个Agent会先调用calculator计算(157)*2得到44然后将这个结果或结合原问题作为上下文决定调用search_web搜索关键词可能是“44”或者“44 结果”。LLM会根据我们的工具描述和对话历史自主做出这个决策。这就形成了一个简单的两步骤任务链。4.2 处理复杂依赖与条件分支上面的例子是线性的。但真实场景中任务链可能有分支。比如我们开头的例子“查天气如果下雨则搜索室内体育馆”。这需要LLM根据中间结果做出判断。我们的run_agent循环已经天然支持了这种模式关键在于我们如何设计工具和描述。我们可以设计一个get_weather工具它返回的结构化信息中需要包含“天气状况”如“晴”、“雨”。LLM在收到这个结果后会结合最初的指令“如果下雨就搜索...”在下一步决策时它就会判断如果天气状况包含“雨”则调用search_web工具否则直接生成最终回复“天气晴好无需搜索室内场馆”。# 模拟天气工具 def get_weather(city: str) - str: # 模拟返回结构化数据用JSON字符串便于LLM解析 import random conditions [晴, 多云, 小雨, 大雨, 阴] condition random.choice(conditions) temp random.randint(15, 30) return json.dumps({city: city, condition: condition, temperature: temp, unit: 摄氏度}, ensure_asciiFalse) # 更新工具列表添加get_weather # ... (将get_weather的描述加入tools列表) # 测试条件任务 final_answer run_agent( user_query查询一下北京的天气如果下雨就搜索‘北京 室内游泳馆’。, toolstools, # 假设tools已包含get_weather和search_web max_steps5 )在这个循环中LLM在第一步调用get_weather(北京)拿到结果{condition: 小雨, ...}。在第二步LLM看到这个结果并回忆起用户指令中的条件“如果下雨”于是它判断需要调用search_web(北京 室内游泳馆)。整个决策逻辑完全由LLM根据上下文推导我们的代码只需要提供一个稳定的“执行-反馈”循环即可。注意事项这里有一个关键点为了让LLM更好地理解工具返回的结果以做出决策工具返回的内容最好也是结构化的比如JSON。虽然LLM能理解自然语言但结构化数据减少了歧义提高了后续步骤的可靠性。你可以让工具返回JSON字符串并在工具描述中说明返回值的结构。5. 工程化与优化让你的Agent更健壮一个能跑起来的原型和一个健壮的可用系统之间还有不少距离。下面我们来探讨几个关键的工程化优化点。5.1 工具管理的标准化用一堆if-elif来匹配工具名和函数显然不是长久之计。我们需要一个中心化的工具注册和管理机制。class ToolRegistry: def __init__(self): self._tools [] # 存放OpenAI格式的工具描述 self._functions {} # 存放工具名到实际函数的映射 def register(self, func, description: str, parameters: dict): 注册一个工具 tool_schema { type: function, function: { name: func.__name__, description: description, parameters: { type: object, properties: parameters.get(properties, {}), required: parameters.get(required, []) } } } self._tools.append(tool_schema) self._functions[func.__name__] func return func # 方便用作装饰器 def get_tools_schema(self): return self._tools def execute(self, tool_name: str, **kwargs): 执行一个已注册的工具 if tool_name not in self._functions: raise ValueError(f工具未注册: {tool_name}) return self._functions[tool_name](**kwargs) # 使用装饰器注册工具更加优雅 registry ToolRegistry() registry.register( description获取当前的日期和时间, parameters{type: object, properties: {}, required: []} ) def get_current_time(): now datetime.datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S) registry.register( description执行数学计算支持加()、减(-)、乘(*)、除(/), parameters{ type: object, properties: { expression: {type: string, description: 数学表达式如 3 5 * 2} }, required: [expression] } ) def calculator(expression: str): # ... (安全实现) pass # 在run_agent中工具执行部分变为 # tool_result registry.execute(func_name, **func_args)这样工具的管理变得清晰、可扩展。新增工具只需用registry.register装饰一下即可。5.2 对话历史管理与上下文长度LLM有上下文窗口限制如4K、8K、128K tokens。我们的messages列表会随着对话轮次和工具调用结果不断增长最终可能超出限制。我们需要一个策略来管理历史。简单截断只保留最近N条消息或确保总token数不超过阈值。缺点是可能丢失关键的前期指令。摘要压缩定期用LLM对之前的对话历史进行总结将冗长的历史压缩成一段简短的摘要然后用“摘要近期消息”作为新的上下文。这是更高级和实用的策略但实现稍复杂。关键信息提取对于工具返回的冗长结果如搜索返回的10个网页摘要可以先用LLM提取与当前任务最相关的几句只把精华放入上下文。在原型阶段我们可以先采用简单截断并密切关注token消耗。OpenAI的API返回中通常包含usage字段可以据此进行监控。5.3 错误处理与稳定性一个健壮的Agent必须能处理各种异常。工具执行错误网络超时、API调用失败、参数错误等。我们的工具函数应该用try...except捕获异常并返回清晰的错误信息如“Error: 计算服务暂时不可用”。这个错误信息会被反馈给LLMLLM有可能尝试其他方法或向用户报告问题。LLM输出格式错误LLM可能返回不符合tool_calls格式的内容或者参数不符合JSON Schema。我们需要在解析tool_call.function.arguments时做好json.loads的异常捕获并给予降级处理例如提示LLM重新生成。循环检测Agent可能陷入死循环反复调用相同的工具。除了设置max_steps硬性限制还可以检测最近几步的行动是否重复如果重复则中断并提示。结果验证对于关键操作如发送邮件、修改数据库在执行前可以增加一个确认环节例如让LLM生成一个操作摘要经用户确认或另一套安全规则确认后再执行。# 在工具执行环节增加错误处理 try: tool_result registry.execute(func_name, **func_args) except Exception as e: tool_result f工具执行失败: {str(e)} # 可以选择将严重错误记录日志甚至终止Agent运行5.4 思维链Chain-of-Thought与规划能力提升要让Agent处理更复杂的任务有时需要提升其规划能力。除了依赖LLM自身的推理我们还可以在提示词Prompt上做文章。在系统提示System Prompt中明确角色和流程例如“你是一个任务规划助手。请逐步思考每次只调用一个最必要的工具。根据工具返回的结果决定下一步是继续调用工具还是给出最终答案。”要求LLM输出思考过程在user或assistant的消息中要求LLM先以“思考...”的形式输出推理再决定行动。虽然OpenAI的Function Calling API不直接支持在tool_calls同时输出content但你可以通过设计对话流程来模拟比如先让LLM以普通文本输出计划然后再在下一轮中调用工具。分解复杂指令对于非常复杂的用户请求可以设计一个“任务分解”工具让LLM先将大任务拆解成清晰的子任务列表然后再逐个执行。这相当于实现了一个简单的规划模块。6. 实战构建一个多技能的个人助理Agent现在让我们把上面的所有部分组合起来构建一个稍微丰富一点的个人助理Agent它具备查询时间、计算、模拟搜索、查询天气、甚至记录备忘录模拟的能力。import json import datetime import random from typing import Dict, List class PersonalAssistantAgent: def __init__(self, llm_client, system_promptNone): self.client llm_client self.messages [] if system_prompt: self.messages.append({role: system, content: system_prompt}) self.tool_registry ToolRegistry() self._register_builtin_tools() def _register_builtin_tools(self): 注册内置工具 self.tool_registry.register( description获取当前日期和时间, parameters{type: object, properties: {}, required: []} ) def get_current_time(): return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) self.tool_registry.register( description执行数学计算支持加减乘除及math库常用函数, parameters{ type: object, properties: { expression: {type: string, description: 数学表达式如 sqrt(9) 5 * 2} }, required: [expression] } ) def calculator(expression: str): # 使用一个更安全的计算方式示例仅支持有限操作 # 实际应用中应使用更完善的库如 numexpr 或自定义安全解析器 allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} allowed_names.update({abs: abs, round: round}) try: # 警告此方法仍非绝对安全仅用于演示 compiled_expr compile(expression, string, eval) for name in compiled_expr.co_names: if name not in allowed_names: raise ValueError(f禁止使用的名称: {name}) result eval(compiled_expr, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误: {e} self.tool_registry.register( description模拟网络搜索返回几条模拟结果摘要, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词} }, required: [query] } ) def search_web(query: str): mock_data { Python: [Python是一种高级编程语言。, 以简洁易读著称。, 广泛用于Web开发和数据分析。], 天气: [天气是大气状态的短期变化。, 通常用温度、湿度、降水等描述。, 天气预报依赖于气象模型。], AI: [人工智能是模拟人类智能的技术。, 包括机器学习、深度学习等领域。, 正在改变许多行业。] } for key, results in mock_data.items(): if key.lower() in query.lower(): return \n.join([f{i1}. {r} for i, r in enumerate(results[:3])]) return f未找到与{query}高度相关的模拟结果。这里是一些通用信息...模拟 self.tool_registry.register( description查询指定城市的模拟天气信息, parameters{ type: object, properties: { city: {type: string, description: 城市名称如北京、上海} }, required: [city] } ) def get_weather(city: str): conditions [晴, 多云, 阴, 小雨, 中雨, 大雨, 雪] condition random.choice(conditions) temp random.randint(0, 35) return json.dumps({ city: city, condition: condition, temperature: temp, unit: 摄氏度, tip: 记得带伞 if 雨 in condition else 天气不错 }, ensure_asciiFalse) # 一个简单的“备忘录”模拟存储在内存中 self.notes: List[Dict] [] self.tool_registry.register( description添加一条文本备忘录, parameters{ type: object, properties: { note: {type: string, description: 备忘录内容} }, required: [note] } ) def add_note(note: str): note_id len(self.notes) 1 self.notes.append({id: note_id, content: note, time: datetime.datetime.now().isoformat()}) return f已添加备忘录 (ID: {note_id}): {note} self.tool_registry.register( description列出所有备忘录, parameters{type: object, properties: {}, required: []} ) def list_notes(): if not self.notes: return 当前没有备忘录。 return \n.join([f{n[id]}. [{n[time]}] {n[content]} for n in self.notes]) def chat(self, user_input: str, max_turns8): 与Agent进行一轮对话可能触发多步工具调用 self.messages.append({role: user, content: user_input}) for turn in range(max_turns): # 调用LLM response self.client.chat.completions.create( modelgpt-3.5-turbo, messagesself.messages, toolsself.tool_registry.get_tools_schema(), tool_choiceauto, ) msg response.choices[0].message self.messages.append(msg) # 如果没有工具调用返回最终回复 if not msg.tool_calls: return msg.content # 执行工具调用 for tool_call in msg.tool_calls: func_name tool_call.function.name try: func_args json.loads(tool_call.function.arguments) except json.JSONDecodeError: tool_result 错误工具参数格式无效。 else: try: tool_result self.tool_registry.execute(func_name, **func_args) except Exception as e: tool_result f工具{func_name}执行出错: {str(e)} # 将结果反馈给LLM self.messages.append({ role: tool, tool_call_id: tool_call.id, content: str(tool_result), }) return 对话轮次已达上限请简化您的问题。 # 使用示例 if __name__ __main__: client openai.OpenAI(api_keyyour-api-key) # 请替换为你的API Key system_prompt 你是一个乐于助人的个人助理。请逐步思考利用可用工具解决问题。 每次尽量只调用一个最必要的工具。根据工具返回的结果决定下一步是继续调用工具还是给出友好、清晰的最终答案。 agent PersonalAssistantAgent(client, system_prompt) queries [ 现在几点了顺便帮我计算一下(12.5 4.3) * 2 等于多少。, 北京天气怎么样如果下雨就搜索一下‘室内活动推荐’。, 帮我记一下明天下午三点有个团队会议。然后再列出我所有的备忘录。, ] for query in queries: print(f\n用户: {query}) answer agent.chat(query) print(f助理: {answer}) print(- * 40) # 重置对话历史避免上下文过长影响下一个独立问题 agent.messages [agent.messages[0]] if agent.messages else []这个PersonalAssistantAgent类集成了工具注册、对话管理、错误处理等核心功能。你可以通过registry.register装饰器轻松扩展新的工具。它已经能够处理包含多个步骤和条件判断的复杂查询。7. 常见问题与排查技巧实录在实际开发和调试手写Agent的过程中你肯定会遇到各种各样的问题。下面是我踩过的一些坑和总结的排查思路。7.1 LLM不调用工具症状无论你怎么问LLM都只用自然语言回答从不触发tool_calls。排查检查工具描述这是最常见的原因。description是否清晰、准确地描述了工具的功能和适用场景LLM是根据描述来判断是否调用的。试着把描述写得更具体、更具指向性。检查tool_choice参数你设置成“auto”了吗如果设成“none”LLM就不会调用工具。“required”会强制LLM调用工具但可能导致它调用不合适的工具。检查系统提示系统提示system消息是否鼓励或要求LLM使用工具例如可以加上“请优先使用我提供的工具来获取信息或执行操作。”问题是否太简单对于“你好吗”这种问题LLM认为无需工具即可回答。尝试问一个明确需要外部信息的问题如“纽约现在几点”模型能力确保你使用的模型支持Function Calling。GPT-3.5-turbo和GPT-4系列都支持。7.2 工具调用参数错误症状LLM调用了工具但参数不对比如类型错误、缺少必填参数、或者参数值毫无意义。排查细化参数描述在parameters的properties里为每个参数提供清晰的description。例如“city”参数可以描述为“城市中文名如‘北京市’、‘上海市’”。提供示例在description或对话上下文中通过示例展示参数的格式。LLM会学习这些模式。在系统提示中约束可以在系统提示中说明“请确保为工具调用提供准确、完整的参数”。后置校验与重试在代码中校验参数如果发现明显错误如城市名称为空可以将错误信息反馈给LLM并要求它重新生成调用。这相当于一个简单的自我修正循环。7.3 Agent陷入死循环或逻辑混乱症状Agent反复调用同一个工具或者在几个工具间无效切换无法达成目标。排查与解决设置最大步数这是最基本的防护如我们的max_steps或max_turns。检查工具返回结果工具返回的结果是否清晰、结构化一个混乱或冗长的结果可能导致LLM无法理解进而做出错误决策。尽量让工具返回简洁、关键的信息。增强系统提示的规划性在系统提示中明确要求“逐步思考”、“先制定计划”、“每次只解决当前最核心的子问题”。这能一定程度上提升LLM的规划能力。引入“任务状态”跟踪对于复杂任务可以设计一个简单的状态机或者让LLM在每一步后输出当前任务的完成状态摘要帮助它保持方向。7.4 上下文长度爆炸症状对话进行到后面LLM回复变慢、成本增高甚至开始遗忘最初的指令。解决策略定期摘要这是最有效的策略。每进行若干步或当上下文token数接近阈值时用LLM对之前的对话历史特别是早期的用户指令和关键中间结果进行总结生成一段简短的“背景摘要”然后用这个摘要替换掉旧的长篇历史。选择性记忆只将与当前步骤最相关的几条历史消息保留在上下文中。这需要你定义“相关性”实现起来较复杂。使用长上下文模型如果成本允许直接使用支持128K或更长上下文的模型如GPT-4 Turbo。但这只是缓解不是根治。7.5 安全与成本控制成本每次工具调用和LLM回复都消耗token。复杂任务链可能消耗大量token。务必在代码中记录和监控usage并为API设置用量告警或硬性限制。工具安全再次强调永远不要相信来自LLM的未经验证的输入去执行危险操作。像calculator中使用eval是极端危险的示例。对于文件操作、数据库查询、系统命令等必须进行严格的输入验证、权限检查和沙箱隔离。内容安全LLM可能被诱导生成或工具可能返回不受控制的内容。需要对最终输出给用户的内容进行必要的过滤和审查尤其是在面向公众的应用中。手写一个AI Agent的过程是一个不断与LLM“沟通”和“协作”的过程。你通过工具描述和系统提示来塑造它的行为它通过推理和调用工具来解决问题。这个过程充满了挑战但也极具乐趣和启发性。当你看到几行简单的代码串联起一个能自动完成多步任务的智能体时那种成就感是无可替代的。希望这篇长文能为你打下坚实的基础让你有能力去探索更广阔的AI Agent世界。