从零搭建AI Agent工具链：手写Python智能体实现ReAct与Function Calling

最近业务中要做一个内部提效工具让大模型根据自然语言自动查数据库、读文件、调内部接口最后产出一份分析报告。刚开始我直接用脚本拼 Prompt发现完全跑不通模型经常答非所问工具调用也总是出格式错误。后来把整个链路拆成“模型调用 工具注册 记忆管理 编排策略”一套工具链才逐渐稳定。这篇文章就把我从零搭建智能体工具链的完整过程分享出来。不依赖 Dify、Coze 这类平台而是用 Python 从模型接入层开始手写一个轻量但可扩展的 Agent 框架最终实现带 Function Calling、工具注册、记忆裁剪和多 Agent 编排的完整体系。1. Agent 到底是什么概念拆解与工具链全景1.1 什么是 AI AgentAI Agent智能体是当前大模型应用里最热的方向之一。通俗地解释Agent 是一个“能自己决定下一步做什么”的 AI 程序。普通的大模型调用是这样的用户输入 - 模型 - 文本输出模型只能“说”不能“做”。而 Agent 是用户输入 - 模型规划 - 调用工具 - 观察结果 - 继续规划 - 最终输出模型不仅能理解你的问题还能主动决定要不要调用某个工具、调用哪个工具、传什么参数。调用完后把结果继续喂给模型模型再判断下一步是继续调用还是给最终答案。这个循环就是 Agent 最核心的机制业界通常叫它ReActReasoning Acting也就是“推理 行动”交替进行。1.2 为什么需要一套完整的工具链只给模型一个工具函数不叫 Agent。真正的 Agent 开发涉及以下几个环节模型接入层不同厂商的模型接口、不同模型版本需要统一封装。工具系统工具如何定义、注册、校验参数、执行、返回结果。记忆模块保持多轮对话上下文必要时支持长期记忆。编排策略单个 Agent 能力有限多个 Agent 如何分工协作。安全边界工具权限控制、敏感操作确认、输出内容过滤。这些环节共同构成一套“智能体工具链”。没有这套工具链开发 Agent 就是写一次性脚本有了这套工具链你可以在不同项目里复用同一套能力。1.3 从零搭建 vs 使用现成平台现在的选择其实挺多Dify、Coze 都能快速搭建 Agent也有一些开源的 Agent 框架。但如果你需要深度定制、离线部署、学习底层机制从零搭建反而是更稳的选择可控性强每一层逻辑都知道是干什么的出了问题好排查。依赖少只依赖模型 API不受平台规则限制。学习价值高能彻底理解 Agent 内部的消息流转机制。本文从零搭建不是“重复造轮子”而是先理解原理再决定要不要引入更重的框架。2. 整体架构设计Agent 工具链的分层模型2.1 五层架构模型一套完整的 Agent 工具链我习惯分成五层层级职责核心组件应用层面向用户提供具体功能命令行工具、Web 服务、群机器人编排层决定任务分给哪个 Agent、执行顺序Orchestrator、多 Agent 协作策略工具层定义和执行工具调用ToolRegistry、Function Calling记忆层管理短期对话历史和长期知识Memory、向量库模型层封装大模型接口统一调用方式ChatClient、Prompt 管理每层只负责自己的事情。例如工具层不关心模型是 GPT 还是 DeepSeek只要模型能输出标准的工具调用格式就行编排层也不关心工具内部怎么实现只看工具返回的结果。2.2 核心流程感知—规划—行动—观察Agent 的单次运行循环可以拆成这样第 1 步接收用户输入 第 2 步把当前对话历史 工具清单发给模型 第 3 步模型决定直接回答或者调用某个工具 第 4 步执行工具拿到结果 第 5 步把结果追加到对话历史 第 6 步回到第 2 步直到模型给出最终结果或达到最大步数注意第 5 步特别关键每次工具调用结果都必须以tool角色追加到消息列表里模型才能看到执行结果。2.3 项目目录结构规划本文实战项目叫min-agent目录结构如下min-agent/ ├── agent.py # Agent 执行器ReAct 循环 ├── llm.py # 模型接入层封装 ├── tools.py # 工具注册与定义 ├── memory.py # 记忆管理模块 ├── orchestration.py # 多 Agent 编排 ├── config.py # 配置读取 ├── main.py # 入口演示 └── requirements.txt # 依赖后面的章节会按这个结构逐个文件实现。3. 环境准备与依赖说明3.1 技术选型本文实现基于 Python 3.10模型接入使用 OpenAI 兼容接口。之所以选 OpenAI 兼容协议是因为国内很多模型服务商DeepSeek、通义千问、Moonshot 等都提供该协议代码可以无缝切换。你需要准备Python 3.10 或更高版本。一个支持 Function Calling / Tools 的大模型 API Key。能用 pip 安装依赖的网络环境。3.2 安装依赖新建虚拟环境后安装下面几个依赖pip install openai python-dotenv这里不引入任何 Agent 框架只用最基础的 SDK。3.3 配置管理在项目根目录建.env文件LLM_BASE_URLhttps://api.example.com/v1 LLM_API_KEYsk-your-key LLM_MODELgpt-4o-mini注意LLM_BASE_URL请填写你实际使用的模型服务地址。如果用 DeepSeek就填 DeepSeek 的接口地址如果用通义就填通义的兼容地址。具体参数以各家服务商文档为准。config.py读取配置import os from dotenv import load_dotenv load_dotenv() class Config: LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.example.com/v1) LLM_API_KEY os.getenv(LLM_API_KEY, ) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) MAX_STEPS int(os.getenv(MAX_STEPS, 5)) MAX_HISTORY int(os.getenv(MAX_HISTORY, 20)) config Config()4. 模型接入层与消息协议4.1 统一的消息格式所有模型调用都围绕一个消息列表messages展开。消息有三种常见角色system系统提示词定义 Agent 的行为。user用户输入。assistant模型回复。当模型发起工具调用时这个回复里会包含tool_calls字段。tool工具执行结果。llm.py封装一个ChatClient屏蔽不同模型的差异from openai import OpenAI from config import config class ChatClient: def __init__(self): self.client OpenAI( base_urlconfig.LLM_BASE_URL, api_keyconfig.LLM_API_KEY, ) self.model config.LLM_MODEL def chat(self, messages, toolsNone): kwargs { model: self.model, messages: messages, } if tools: kwargs[tools] tools response self.client.chat.completions.create(**kwargs) return response.choices[0].message这段代码最核心的是当传入tools时模型具备了“决定是否调用工具”的能力。工具清单会在 API 请求里以 JSON Schema 的形式传给模型。4.2 系统提示词的设计系统提示词决定了 Agent 的“人设”和“行为边界”。下面是一个基础模板SYSTEM_PROMPT 你是一个智能助手你可以通过调用工具来完成用户的任务。 使用规则 1. 当用户的问题需要实时数据或执行操作时使用工具。 2. 当工具返回结果后基于结果回答用户。 3. 如果不需要工具直接回答。 4. 所有回答使用中文。 这个提示词要放在messages列表的第一条。5. 工具系统ToolRegistry 的设计与实现5.1 为什么需要工具注册中心一个 Agent 会挂载很多工具。如果直接在代码里写 if-else 判断工具名称工具一多就会乱。工具注册中心解决几个问题统一管理工具清单。自动生成模型要求的 JSON Schema。按名称快速找到并执行工具。5.2 核心代码实现tools.py完整实现import json class Tool: def __init__(self, name, description, schema, func): self.name name self.description description self.schema schema # JSON Schema self.func func def run(self, **kwargs): return self.func(**kwargs) class ToolRegistry: def __init__(self): self._tools {} def register(self, tool: Tool): self._tools[tool.name] tool def unregister(self, name: str): self._tools.pop(name, None) def get(self, name: str): tool self._tools.get(name) if not tool: raise KeyError(f工具未注册: {name}) return tool def schemas(self): schemas [] for tool in self._tools.values(): schemas.append({ type: function, function: { name: tool.name, description: tool.description, parameters: tool.schema, }, }) return schemas registry ToolRegistry()5.3 内置工具示例实现两个工具获取当前时间、写文件。import datetime def get_current_time(): 获取当前时间 now datetime.datetime.now() return {time: now.strftime(%Y-%m-%d %H:%M:%S)} def save_to_file(path, content): 写入内容到指定文件 with open(path, w, encodingutf-8) as f: f.write(content) return {status: ok, path: path, length: len(content)} registry.register(Tool( nameget_current_time, description获取当前的系统时间当用户询问日期或时间时调用, schema{type: object, properties: {}}, funcget_current_time, )) registry.register(Tool( namesave_to_file, description将文本内容保存到本地文件当用户需要记录备忘或生成文件时调用, schema{ type: object, properties: { path: {type: string, description: 保存路径}, content: {type: string, description: 文件内容}, }, required: [path, content], }, funcsave_to_file, ))注意工具的参数 Schema 必须写严格。模型会根据这个 Schema 生成参数如果字段缺失或类型错误工具执行时就会报错。5.4 工具系统的扩展思路实际项目中你可以在Tool.run()里加入参数校验、超时控制、权限检查也可以在工具返回时统一包一层结构def safe_run(self, **kwargs): try: result self.func(**kwargs) return {success: True, result: result} except Exception as e: return {success: False, error: str(e)}这样工具内部即使报错也不会把 Agent 整个流程打挂。6. Agent 执行器实现 ReAct 循环6.1 执行器的完整代码agent.py是整套工具链的心脏实现前面讲的“推理—行动—观察”循环import json from llm import ChatClient from tools import registry from memory import Memory from config import config class Agent: def __init__(self, system_prompt, nameagent): self.name name self.client ChatClient() self.system_prompt system_prompt self.max_steps config.MAX_STEPS self.memory Memory(max_messagesconfig.MAX_HISTORY) self.tools registry def run(self, user_input): self.memory.add_user(user_input) messages [{role: system, content: self.system_prompt}] messages.extend(self.memory.history) for step in range(1, self.max_steps 1): print(f[{self.name}] 第 {step} 步: 调用模型) message self.client.chat(messages, toolsself.tools.schemas()) if message.tool_calls is None: # 模型没有要求调用工具直接作为最终回答 self.memory.add_assistant(message.content) return message.content # 模型要求调用工具 messages.append(message) for tool_call in message.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) print(f[{self.name}] 调用工具: {fn_name}, 参数: {fn_args}) tool self.tools.get(fn_name) result tool.run(**fn_args) # 工具结果追加回消息列表模型在下一次迭代中可以看到 messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) self.memory.add_tool_result(fn_name, result) return 已达到最大执行步数任务可能未完成。这个循环的关键在于每轮模型返回结果有两种可能要么是普通文字要么是tool_calls。只要模型返回tool_calls就执行工具把结果塞回messages然后继续下一轮。当模型不再要求调用工具时说明它已经拿到足够信息可以输出最终答案了。6.2 每一步都在做什么以“现在几点了帮我记录到笔记.txt”为例第 1 轮模型看到工具列表判断“现在几点”需要 get_current_time返回工具调用。 执行 get_current_time拿到时间。 第 2 轮模型看到时间结果判断“记录到笔记.txt”需要 save_to_file。 执行 save_to_file拿到保存结果。 第 3 轮模型看到保存结果不再要求工具返回最终回答。这就是 ReAct 循环的直观理解。7. 记忆模块短期记忆与上下文管理7.1 为什么 Agent 需要记忆Agent 是多轮对话系统。如果没有记忆模型每轮只能看到当前输入之前聊过的内容全部丢失。比如用户先说“帮我查下北京的天气”再说“那上海呢”模型必须记得上一轮聊的是天气。memory.py实现一个简单的短期记忆class Memory: def __init__(self, max_messages20): self.history [] self.max_messages max_messages def add_user(self, content): self.history.append({role: user, content: content}) self._trim() def add_assistant(self, content): self.history.append({role: assistant, content: content}) self._trim() def add_tool_result(self, tool_name, result): text f工具 {tool_name} 返回{result} self.history.append({role: system, content: text}) self._trim() def clear(self): self.history [] def _trim(self): if len(self.history) self.max_messages: self.history self.history[-self.max_messages:]注意_trim()每次追加消息后都会检查长度。对于上下文窗口有限的模型这个裁剪策略很关键。7.2 记忆裁剪策略的选择裁剪策略有很多种保留最近 N 条最简单但可能丢失早期关键信息。摘要历史每轮对话后让模型生成摘要保留摘要 最近对话。向量检索把历史存入向量库按相关性召回。实际项目里摘要和向量检索更实用。本文先实现前两种策略中的“最近 N 条”后续可以平滑替换为向量检索。7.3 长期记忆的简要思路如果要做长期记忆可以引入一个简单的 JSON 文件作为存储。每次 Agent 启动时加载对话结束后写入。比如import json import os class JsonMemory: def __init__(self, store_pathmemory_store.json): self.store_path store_path self.data {} self._load() def _load(self): if os.path.exists(self.store_path): with open(self.store_path, r, encodingutf-8) as f: self.data json.load(f) def save(self): with open(self.store_path, w, encodingutf-8) as f: json.dump(self.data, f, ensure_asciiFalse, indent2) def put(self, key, value): self.data[key] value self.save() def get(self, key, defaultNone): return self.data.get(key, default)这个模块适合存储用户偏好、历史任务记录等结构化信息。8. 多 Agent 编排让多个智能体协同工作8.1 为什么需要多 Agent单一 Agent 适合简单任务但复杂问题最好拆解。例如“写一篇技术博客”这个任务可以拆成“研究主题”和“撰写文章”两个角色。多 Agent 编排就是让多个各司其职的 Agent 配合。8.2 简单编排实现orchestration.pyfrom agent import Agent from tools import registry class Orchestrator: def __init__(self, agents: dict): self.agents agents def route(self, user_input: str) - str: # 简单的路由规则根据关键词选择 Agent if any(kw in user_input for kw in [写, 生成, 总结]): return writer if any(kw in user_input for kw in [时间, 几点, 日期]): return time return general def run(self, user_input: str): agent_name self.route(user_input) print(f[orchestrator] 路由到 {agent_name}) agent self.agents[agent_name] return agent.run(user_input)这段代码是演示性质真实场景中可以用模型做路由判断也可以用更复杂的规则引擎。重点是编排层只负责“派任务”不关心 Agent 内部实现。8.3 多个 Agent 的执行流程假设创建了time_agent、writer_agent、general_agent三个 Agent分别挂载不同工具。用户输入“现在几点顺便帮我生成一个今天的工作总结”编排器先路由再决定是顺序调用还是并行调用。实际项目可以根据任务依赖关系灵活设计。9. 完整运行与验证9.1 入口文件main.pyfrom agent import Agent from orchestration import Orchestrator from tools import registry import tools # noqa: F401 确保工具注册执行 def create_agents(): time_agent Agent( system_prompt你是一个时间助手负责查询时间。, nametime_agent, ) general_agent Agent( system_prompt你是一个通用助手基于工具结果回答用户问题。, namegeneral_agent, ) return {time: time_agent, general: general_agent} if __name__ __main__: agents create_agents() orch Orchestrator(agents) while True: user_input input(你: ) if user_input.strip().lower() in {exit, quit, q}: break response orch.run(user_input) print(fAgent: {response})运行python main.py预期交互你: 现在几点 [orchestrator] 路由到 time [time_agent] 第 1 步: 调用模型 [time_agent] 调用工具: get_current_time, 参数: {} [time_agent] 第 2 步: 调用模型 Agent: 当前时间是 2025-06-01 14:23:05。9.2 验证工具调用逻辑如果你想不依赖真实模型也可以用一段模拟函数来验证流程。用 mock 数据测试时重点观察消息列表中role是否交替正确。这是 Agent 开发中最容易出错的地方。10. 常见问题与排查思路问题现象常见原因解决思路模型不调用工具工具 Schema 写得不规范或系统提示词没说明检查schemas()输出是否符合 OpenAI 格式工具参数解析失败模型返回的 JSON 里字段和 Schema 不一致在tool.run()外层做异常捕获并打印原始参数多轮循环后输出混乱没有把tool_calls消息 append 回messages确认assistant消息和tool消息都完整保留达到最大步数还没有结果任务太复杂或模型反复调用同一工具加大MAX_STEPS或在提示词里禁止重复调用上下文超长历史消息累积未裁剪调整Memory._trim()的值增加摘要或向量召回API 返回 401API Key 错误或 base_url 不匹配检查.env配置确认接口协议是 OpenAI 兼容排查 Agent 问题有一个万能套路把传给模型的messages原样打印出来看每一轮的消息是否完整、顺序是否正确。90% 的问题都出在消息列表上。11. 最佳实践与工程建议11.1 工具设计原则每个工具只做一件事不要写“万能工具”。工具描述要具体说明什么情况下使用模型才能正确选择。参数 Schema 的description写清楚模型才能生成正确参数。工具必须有超时和异常处理不能因为一个工具失败拖垮整个 Agent。11.2 提示词工程建议系统提示词里明确“工具优先还是直接回答”。控制模型的自由发挥空间用“必须基于工具结果回答”这类约束。不要把所有规则都堆进提示词能写在代码里的逻辑不要写进提示词。11.3 生产环境安全边界权限最小化Agent 挂载的文件写入工具应该限制可写目录而不是任意路径。敏感操作确认删除文件、发送消息、执行命令等工具需要二次确认。日志记录每个工具调用都要记录调用方、参数、结果、耗时便于审计。速率限制Agent 循环会频繁调用模型要考虑 API 限流和成本控制。11.4 可观测性开发 Agent 最容易遇到“黑盒问题”。建议在关键节点打日志[time_agent] 第 1 步: 调用模型 [time_agent] 调用工具: get_current_time, 参数: {} [time_agent] 工具返回: {time: 2025-06-01 14:23:05}这套日志能直观看到 Agent 每一步在做什么排查问题效率会高很多。12. 总结与学习路线这套工具链从模型接入层、工具注册中心、ReAct 执行器、记忆模块到多 Agent 编排已经覆盖了 Agent 开发的核心链路。建议按下面的路线继续深入先把本文代码完整跑通理解messages的流转过程。给 ToolRegistry 增加参数校验和权限控制。引入向量数据库实现长期记忆。尝试把编排策略从规则升级为模型判断。扩展 Agent 的工具比如 HTTP 请求、数据库查询、代码执行。下一步想深入的话重点研究两个方向一是 Function Calling 在不同模型上的差异二是多 Agent 协作中的任务拆分与结果合并。这两个方向是 Agent 开发走向工程化的必经之路。