AI Agent工具调用最小实现:从零搭建消息闭环

AI Agent工具调用最小实现:从零搭建消息闭环 AI 安全讨论得再多落到 AI Agent 应用开发上仍然要回到最基础的工具调用闭环模型决定调用哪个函数代码执行并校验结果审计日志记录全部过程。一个没有权限边界、没有结果校验、没有日志的 Agent只是把更多不可控动作交给了大模型。本文用最小可运行代码实现一个基于工具调用的 AI Agent 小项目适合刚接触 AI 应用开发的读者。项目会包含两类执行方式没有 API Key 时用本地模拟决策器跑通消息闭环有 API Key 时切换到大模型决策器让模型根据用户问题选择工具。最终你会理解tools、tool_calls、roletool这套消息结构并能在此基础上继续扩展真实业务工具。1. 先理解 AI Agent 为什么需要工具调用机制1.1 普通对话与 Agent 的分水岭普通 Chat Completion 的调用方式非常简单把用户输入放到messages里发给模型模型返回一段纯文本。这个过程适合问答、翻译、摘要但模型无法主动查询数据库、读取文件、调用内部接口也无法获得训练数据之后出现的信息。AI Agent 的核心区别在于模型不再直接给出最终答案而是先输出一个“调用意图”。例如用户问“北京现在是几点”模型可能返回一个结构化内容调用get_city_time工具参数是{city: 北京}。这是模型“决定”的部分。真正执行时间查询动作的是代码不是模型。执行完工具后代码把结果回传给模型模型再基于这个结果组织成自然语言回答。模型负责推理和表达代码负责执行和校验两者各司其职。这也是 Agent 比普通聊天机器人可靠的原因实时数据、计算逻辑、外部系统交互都由确定性的代码完成而不是让模型凭记忆生成。1.2 工具调用的完整闭环工具调用在接口层面有一套固定消息流程理解这个流程比记住某个 SDK 方法更重要。开发者在请求中声明工具列表tools。每个工具包含名称、描述和参数 JSON Schema。用户问题放进messages像普通对话一样发送给模型。模型返回两种结果之一最终文本或者tool_calls数组。如果模型返回的是最终文本说明它认为不需要调用工具直接把这个文本返回给用户。如果模型返回tool_calls程序遍历数组执行对应函数并把每个执行结果追加为一条roletool的消息。带着追加后的完整messages再次请求模型。模型看到工具结果后可能生成最终回答也可能继续调用下一个工具。这个循环会一直重复直到模型不再返回tool_calls或者达到开发者设置的最大轮数。最大轮数非常关键否则遇到模型反复调用同一个工具时会一直循环下去既浪费 token也拖慢响应。为什么工具结果必须是roletool而不是简单拼接到下一轮 user 消息里因为接口层面的角色体系要求每个工具结果必须关联到对应的tool_call_id。这样模型可以区分哪次调用产生了哪个结果尤其当一次请求里同时调用多个工具时消息结构依然不会乱。1.3 本地模拟版和真实 API 版的思路差异很多新手一开始就想接真实大模型结果卡在 API Key、网络、模型权限上。更务实的做法是先把“执行循环”写出来再用一个最简单的规则决策器替代模型让消息闭环先跑通。真实大模型版与本地模拟版共用的部分包括工具定义、工具执行函数、工具结果回填、最大轮数控制。区别只在“决策器”真实决策器调用 OpenAI 兼容接口由模型根据用户问题选择工具。本地模拟决策器根据问题文本里的关键词返回相同结构的tool_calls。这样设计可以分两步学习先理解消息结构再理解模型行为。项目本身也支持通过环境变量切换两种模式方便本地调试。2. 环境准备和最小项目结构2.1 Python 环境与依赖建议使用 Python 3.10 及以上版本不需要 GPU普通笔记本即可运行。需要安装两个 Python 库pip install openai python-dotenvopenai用于调用兼容接口python-dotenv用于读取.env文件中的环境变量。依赖版本以官方发布为准落地到生产项目前建议锁定具体版本号避免升级造成行为变化。如果当前环境已经有其他版本的openai可以先查看版本pip show openai新版 SDK 的客户端写法通常是OpenAI()老版本的ChatCompletion.create写法已经过时。如果代码报unexpected keyword argument tools大概率是 SDK 版本过旧需要升级。2.2 API Key 的安全使用方式API Key 是身份凭证不能硬编码到源码中也不能出现在提交到 Git 的配置文件里。本地开发推荐放在.env文件中然后由python-dotenv加载。.env文件示例OPENAI_API_KEY你的key OPENAI_MODELgpt-4o-mini AGENT_MODEmock使用load_dotenv()后代码里通过os.getenv(OPENAI_API_KEY)读取。.gitignore中必须加入.env还要注意三点不要打印完整 Key。调试时最多显示前几位用于确认是否加载成功。不要把 Key 分享到公开仓库、聊天群、论坛或博客示例中。生产环境建议使用云厂商的密钥管理服务例如环境变量注入、KMS、Vault并按需配置 Key 的权限范围。如果暂时没有 API Key没关系。项目默认使用本地模拟模式不需要联网也能跑通整个 Agent 消息循环。等流程理解清楚后再切换到真实大模型模式。2.3 项目目录结构推荐使用一个最小目录结构后续扩展工具时也更清晰my_agent/ ├── .env.example ├── .gitignore ├── agent.py └── tools.py文件职责文件职责.env.example环境变量样例只写变量名和占位值不写真实 Key.gitignore忽略.env、缓存目录、虚拟环境目录tools.py所有工具函数的定义和实现agent.py工具描述、消息循环、决策器、命令行入口在实际项目里可以把工具描述、工具实现、Agent 循环拆成更细的模块但这个小项目保持两个文件足够。3. 实现一个最小 Agent查询时间和数学计算3.1 定义工具描述工具描述是模型选择工具的依据。描述越清晰模型越不容易用错。在agent.py中定义两个工具TOOLS [ { type: function, function: { name: get_city_time, description: 获取指定城市的当前时间。城市名称用中文例如北京。注意本工具返回的是代码运行所在服务器的本地时间。, parameters: { type: object, properties: { city: { type: string, description: 目标城市名称例如 北京 } }, required: [city] } } }, { type: function, function: { name: calculator, description: 执行一个只包含数字、加、减、乘、除和括号的数学表达式返回计算结果。, parameters: { type: object, properties: { expression: { type: string, description: 数学表达式例如 2100 / 7 3 } }, required: [expression] } } } ]关键点在于description。模型不会读源码它只能根据这段描述决定是否调用工具。所以描述里要说明函数做什么、参数格式是什么、边界条件是什么。parameters使用 JSON Schema 格式。typeobject表示参数是一个对象properties描述每个字段required声明必填字段。如果某个工具没有必填参数required可以留空数组。常见错误是把所有参数都塞进一个字符串里让模型自己解析。结果往往是格式不稳定参数偶尔缺字段。更稳妥的做法是先定义清晰的结构化参数再在工具执行层校验。3.2 实现安全的工具函数工具执行层要独立处理异常不能假设模型返回的参数一定合法。第一个工具返回城市时间。这里不查真实时区数据库只返回服务器本地时间用于演示流程def get_city_time(city: str) - str: now datetime.now() return f{city}的当前时间是 {now.strftime(%Y-%m-%d %H:%M:%S)}第二个工具是数学计算。新手最容易直接写eval(expression)这非常危险。用户输入或模型参数可能包含恶意代码直接eval会造成任意代码执行。这里用一个基于 AST 的安全计算器只允许数字、加减乘除、括号和一元正负号import ast import operator def _eval_expr(node): if isinstance(node, ast.Constant): if isinstance(node.value, (int, float)) and not isinstance(node.value, bool): return node.value raise ValueError(只支持整数或小数) if isinstance(node, ast.BinOp): left _eval_expr(node.left) right _eval_expr(node.right) operations { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, } op_func operations.get(type(node.op)) if op_func is None: raise ValueError(不支持的运算符) return op_func(left, right) if isinstance(node, ast.UnaryOp): operand _eval_expr(node.operand) if isinstance(node.op, ast.USub): return -operand if isinstance(node.op, ast.UAdd): return operand raise ValueError(不支持的一元运算符) raise ValueError(表达式包含不支持的内容) def calculator(expression: str) - str: try: tree ast.parse(expression, modeeval) return str(_eval_expr(tree.body)) except ZeroDivisionError: return 计算失败除数不能为 0 except Exception as e: return f计算失败{e}用 AST 解析后逐节点计算完全不调用eval和exec可以从源头避免注入。工具返回的永远是字符串这样回填到消息时类型统一模型也更容易理解。实现完成后把工具名称和函数映射到一个字典方便后面的执行循环查找TOOL_IMPL { get_city_time: get_city_time, calculator: calculator, }3.3 实现执行与回填循环工具调用循环是整个 Agent 的骨架。先实现execute_tool_calls负责解析模型返回的工具调用参数并执行函数import json def execute_tool_calls(tool_calls): results [] for call in tool_calls: fn_name call[function][name] raw_args call[function].get(arguments) or {} try: args json.loads(raw_args) except json.JSONDecodeError: args {} fn TOOL_IMPL.get(fn_name) if fn is None: output f未知工具{fn_name} else: try: output fn(**args) except TypeError as e: output f工具参数不正确{e} except Exception as e: output f工具执行失败{e} results.append({ call_id: call[id], output: output }) return results这里有一个容易踩的坑模型返回的arguments是 JSON 字符串必须先json.loads再作为关键字参数传入。如果 JSON 解析失败不能直接让程序崩溃要返回一条错误信息给模型模型看到错误后可能会自动修正参数。接着实现run_agent它负责维护整个消息历史MAX_TURNS 5 def run_agent(question, decider): messages [{role: user, content: question}] for turn in range(MAX_TURNS): decision decider(messages) if isinstance(decision, str): return decision if decision is None: return 决策器没有返回工具调用或最终回答 messages.append({ role: assistant, content: , tool_calls: decision }) results execute_tool_calls(decision) for item in results: messages.append({ role: tool, tool_call_id: item[call_id], content: item[output] }) return 超过最大工具调用轮数没有生成最终回答run_agent接受一个decider参数它负责根据当前messages返回“最终回答字符串”或“工具调用列表”。这样真实模型和本地模拟可以复用同一份循环代码。tool_call_id必须与 assistant 消息中某个tool_calls的id一一对应。如果回填时tool_call_id对不上接口会报格式错误。3.4 接入模拟决策器和真实 API 决策器本地模拟决策器不调用任何外部接口只根据问题文本里的关键词返回工具调用def mock_decider(messages): last messages[-1] if last[role] tool: return f工具返回结果{last[content]} question last[content] if 时间 in question or 几点 in question: city 北京 if 北京 in question else 当前城市 return [{ id: call_mock_time, type: function, function: { name: get_city_time, arguments: json.dumps({city: city}, ensure_asciiFalse) } }] if 计算 in question: expression question.split(计算, 1)[1].strip() return [{ id: call_mock_calc, type: function, function: { name: calculator, arguments: json.dumps({expression: expression}, ensure_asciiFalse) } }] return f模拟模式只支持时间查询和数学计算当前问题{question}真实模型决策器调用 OpenAI 兼容接口from openai import OpenAI def openai_decider(messages): client OpenAI() model os.getenv(OPENAI_MODEL, gpt-4o-mini) response client.chat.completions.create( modelmodel, messagesmessages, toolsTOOLS, temperature0.2, ) message response.choices[0].message if not message.tool_calls: return message.content or return [{ id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments } } for tc in message.tool_calls]这里把 SDK 返回的对象转成普通 dict 列表再交给run_agent。好处是run_agent不依赖具体 SDK 类型以后换其他兼容接口时只需要改openai_decider内部实现。3.5 入口和运行命令在agent.py末尾添加命令行入口if __name__ __main__: import sys question sys.argv[1] if len(sys.argv) 1 else 现在北京几点 mode os.getenv(AGENT_MODE, mock) if mode mock: answer run_agent(question, mock_decider) else: answer run_agent(question, openai_decider) print(answer)运行模拟模式cd my_agent AGENT_MODEmock python agent.py 现在北京几点 AGENT_MODEmock python agent.py 计算 2100 / 7 3运行真实模型模式AGENT_MODEopenai OPENAI_MODELgpt-4o-mini python agent.py 计算 2100 / 7 3如果OPENAI_API_KEY已经配置在.env文件中会自动加载。如果没有 Key真实模式会报