当前位置：首页 > news >正文

从零手写AI Agent：基于Function Calling与任务链的智能体构建实践

news 2026/8/14 4:41:39

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 Calling？

Function 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_choice="auto"或tool_choice="required"）。

import openai import json # 假设你已经设置了OPENAI_API_KEY client = openai.OpenAI() def chat_with_llm(messages, tools): """发送消息给LLM，并允许其调用工具""" response = client.chat.completions.create( model="gpt-3.5-turbo", # 或 "gpt-4" messages=messages, tools=tools, tool_choice="auto", # 让模型自行决定是否调用工具 ) return response.choices[0].message # 初始化对话 messages = [{"role": "user", "content": "现在几点了？"}] llm_response_message = chat_with_llm(messages, tools) print(f"LLM的回复消息对象: {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(f"LLM直接回复：{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_steps=10): """ 运行一个简单的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 = f"Error: 未知工具 '{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等于多少，然后基于这个数字搜索一下相关的结果。", tools=tools, max_steps=5 ) print(f"\n=== 最终答案 ===\n{final_answer}")

这个Agent会先调用calculator计算(15+7)*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_ascii=False) # 更新工具列表，添加get_weather # ... (将get_weather的描述加入tools列表) # 测试条件任务 final_answer = run_agent( user_query="查询一下北京的天气，如果下雨，就搜索‘北京 室内游泳馆’。", tools=tools, # 假设tools已包含get_weather和search_web max_steps=5 )

在这个循环中，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: 计算服务暂时不可用”）。这个错误信息会被反馈给LLM，LLM有可能尝试其他方法或向用户报告问题。
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_prompt=None): 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"{i+1}. {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_ascii=False) # 一个简单的“备忘录”模拟，存储在内存中 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_turns=8): """与Agent进行一轮对话（可能触发多步工具调用）""" self.messages.append({"role": "user", "content": user_input}) for turn in range(max_turns): # 调用LLM response = self.client.chat.completions.create( model="gpt-3.5-turbo", messages=self.messages, tools=self.tool_registry.get_tools_schema(), tool_choice="auto", ) 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_key="your-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。
排查：
1. 检查工具描述：这是最常见的原因。description是否清晰、准确地描述了工具的功能和适用场景？LLM是根据描述来判断是否调用的。试着把描述写得更具体、更具指向性。
2. 检查tool_choice参数：你设置成“auto”了吗？如果设成“none”，LLM就不会调用工具。“required”会强制LLM调用工具，但可能导致它调用不合适的工具。
3. 检查系统提示：系统提示（system消息）是否鼓励或要求LLM使用工具？例如，可以加上“请优先使用我提供的工具来获取信息或执行操作。”
4. 问题是否太简单：对于“你好吗？”这种问题，LLM认为无需工具即可回答。尝试问一个明确需要外部信息的问题，如“纽约现在几点？”
5. 模型能力：确保你使用的模型支持Function Calling。GPT-3.5-turbo和GPT-4系列都支持。

7.2 工具调用参数错误

症状：LLM调用了工具，但参数不对，比如类型错误、缺少必填参数、或者参数值毫无意义。
排查：
1. 细化参数描述：在parameters的properties里，为每个参数提供清晰的description。例如，“city”参数可以描述为“城市中文名，如‘北京市’、‘上海市’”。
2. 提供示例：在description或对话上下文中，通过示例展示参数的格式。LLM会学习这些模式。
3. 在系统提示中约束：可以在系统提示中说明“请确保为工具调用提供准确、完整的参数”。
4. 后置校验与重试：在代码中校验参数，如果发现明显错误（如城市名称为空），可以将错误信息反馈给LLM，并要求它重新生成调用。这相当于一个简单的自我修正循环。

7.3 Agent陷入死循环或逻辑混乱

症状：Agent反复调用同一个工具，或者在几个工具间无效切换，无法达成目标。
排查与解决：
1. 设置最大步数：这是最基本的防护，如我们的max_steps或max_turns。
2. 检查工具返回结果：工具返回的结果是否清晰、结构化？一个混乱或冗长的结果可能导致LLM无法理解，进而做出错误决策。尽量让工具返回简洁、关键的信息。
3. 增强系统提示的规划性：在系统提示中明确要求“逐步思考”、“先制定计划”、“每次只解决当前最核心的子问题”。这能一定程度上提升LLM的规划能力。
4. 引入“任务状态”跟踪：对于复杂任务，可以设计一个简单的状态机，或者让LLM在每一步后输出当前任务的完成状态摘要，帮助它保持方向。

7.4 上下文长度爆炸

症状：对话进行到后面，LLM回复变慢、成本增高，甚至开始遗忘最初的指令。
解决策略：
1. 定期摘要：这是最有效的策略。每进行若干步或当上下文token数接近阈值时，用LLM对之前的对话历史（特别是早期的用户指令和关键中间结果）进行总结，生成一段简短的“背景摘要”，然后用这个摘要替换掉旧的长篇历史。
2. 选择性记忆：只将与当前步骤最相关的几条历史消息保留在上下文中。这需要你定义“相关性”，实现起来较复杂。
3. 使用长上下文模型：如果成本允许，直接使用支持128K或更长上下文的模型（如GPT-4 Turbo）。但这只是缓解，不是根治。

7.5 安全与成本控制

成本：每次工具调用和LLM回复都消耗token。复杂任务链可能消耗大量token。务必在代码中记录和监控usage，并为API设置用量告警或硬性限制。
工具安全：再次强调，永远不要相信来自LLM的未经验证的输入去执行危险操作。像calculator中使用eval是极端危险的示例。对于文件操作、数据库查询、系统命令等，必须进行严格的输入验证、权限检查和沙箱隔离。
内容安全：LLM可能被诱导生成或工具可能返回不受控制的内容。需要对最终输出给用户的内容进行必要的过滤和审查，尤其是在面向公众的应用中。

手写一个AI Agent的过程，是一个不断与LLM“沟通”和“协作”的过程。你通过工具描述和系统提示来塑造它的行为，它通过推理和调用工具来解决问题。这个过程充满了挑战，但也极具乐趣和启发性。当你看到几行简单的代码串联起一个能自动完成多步任务的智能体时，那种成就感是无可替代的。希望这篇长文能为你打下坚实的基础，让你有能力去探索更广阔的AI Agent世界。

查看全文

http://www.jsqmd.com/news/1389408/