在大模型技术迅猛发展的当下，AI Agent已成为连接语言模型与现实世界的关键桥梁。本文将手把手带你使用纯Python构建一个能够自主决策、调度多种工具并完成复杂任务的智能Agent系统。我们不会依赖重型框架，而是从底层原理出发，让你透彻理解Agent的核心工作机制。
一、什么是多工具AI Agent
多工具AI Agent是一个智能体程序，它能够接收用户的自然语言指令，自主分析任务需求，从可用的工具集中选择合适的工具（如搜索引擎、计算器、文件读写器、API调用器等），按正确的顺序调用它们，并将各工具返回的结果整合成最终答案。与传统的”一问一答”模式不同，Agent具备规划能力、工具选择能力和反思纠错能力。
举个实际场景：用户问”帮我查一下旧金山今天的天气，然后换算成摄氏度，如果温度高于25度就帮我生成一封提醒带防晒霜的邮件草稿”。这个任务涉及天气查询、温度转换、条件判断和邮件生成四个环节，单个LLM调用无法完成，需要一个Agent来协调多个工具协同工作。
二、系统架构设计
我们的Agent系统采用模块化架构，核心包含以下组件：
- LLM推理引擎：负责理解用户意图、生成工具调用计划和整合最终回答
- 工具注册中心：管理所有可用工具的定义、参数规范和调用接口
- 任务规划器：将复杂任务分解为可执行的步骤序列
- 执行引擎：按计划依次调用工具，处理中间结果和异常
- 记忆模块：保存对话上下文和中间执行状态
下面是整体架构的流程示意：用户输入经过LLM推理引擎解析后，任务规划器生成执行计划，执行引擎从工具注册中心获取对应工具并调用，结果经LLM整合后返回给用户。如果某一步执行失败，Agent会自动尝试替代方案或调整计划。
三、完整代码实现
我们将使用Python从零实现这个Agent。这里以OpenAI兼容接口作为LLM后端（你也可以替换为本地模型如Ollama），并集成四个实用工具：网页搜索模拟器、数学计算器、日期时间工具和文件操作工具。
3.1 工具定义与注册中心
首先定义一个统一的工具接口，让所有工具遵循相同的调用规范：
import json
import math
from datetime import datetime, timedelta
from typing import Any, Callable
import re
class Tool:
"""工具基类，定义统一的调用接口"""
def __init__(self, name: str, description: str, func: Callable, parameters: dict):
self.name = name
self.description = description
self.func = func
self.parameters = parameters # JSON Schema格式的参数定义
def execute(self, **kwargs) -> str:
"""执行工具并返回字符串结果"""
try:
result = self.func(**kwargs)
return str(result)
except Exception as e:
return f"工具执行出错: {str(e)}"
def to_openai_format(self) -> dict:
"""转换为OpenAI函数调用格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}3.2 实现四个实用工具
下面是四个具体工具的实现，每个工具都有明确的职责和参数定义：
# 工具1: 数学计算器
def calculator_func(expression: str) -> str:
"""安全地计算数学表达式"""
allowed_names = {"abs": abs, "round": round, "max": max, "min": min,
"pow": pow, "sqrt": math.sqrt, "sin": math.sin,
"cos": math.cos, "pi": math.pi, "e": math.e}
# 只允许安全的数学函数和数字
sanitized = expression.replace(" ", "")
if not re.match(r'^[d+-*/().,%sw]+$', sanitized):
return "错误: 表达式包含不允许的字符"
try:
result = eval(sanitized, {"__builtins__": {}}, allowed_names)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
calculator_tool = Tool(
name="calculator",
description="执行数学计算，支持加减乘除、三角函数、开方等运算",
func=calculator_func,
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，例如: '2+3*4', 'sqrt(144)', 'sin(pi/2)'"
}
},
"required": ["expression"]
}
)
# 工具2: 日期时间工具
def datetime_func(operation: str, date_str: str = "", days_offset: int = 0) -> str:
"""日期时间相关操作"""
now = datetime.now()
if operation == "get_current":
return f"当前日期时间: {now.strftime('%Y-%m-%d %H:%M:%S')}, 星期{['一','二','三','四','五','六','日'][now.weekday()]}"
elif operation == "add_days":
target_date = now + timedelta(days=days_offset)
return f"{days_offset}天后的日期: {target_date.strftime('%Y-%m-%d')}"
elif operation == "diff_days":
if date_str:
target = datetime.strptime(date_str, "%Y-%m-%d")
diff = (target - now).days
return f"距离{date_str}还有{diff}天"
return "未知操作"
datetime_tool = Tool(
name="datetime_util",
description="获取当前时间、计算日期差值、推算未来日期",
func=datetime_func,
parameters={
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["get_current", "add_days", "diff_days"],
"description": "操作类型"
},
"date_str": {"type": "string", "description": "目标日期，格式YYYY-MM-DD"},
"days_offset": {"type": "integer", "description": "天数偏移量"}
},
"required": ["operation"]
}
)
# 工具3: 模拟网页搜索
def web_search_func(query: str, num_results: int = 3) -> str:
"""模拟搜索引擎返回结果（实际项目中可接入真实搜索API）"""
knowledge_base = {
"旧金山天气": "旧金山今日多云转晴，气温18-28摄氏度，湿度65%，风力3级",
"python最新版本": "Python最新稳定版本为3.12.4，于2024年6月发布，主要改进了性能和多线程支持",
"langchain": "LangChain是一个用于构建LLM应用的开源框架，支持Python和JavaScript",
"openai": "OpenAI是一家AI研究公司，开发了GPT系列大语言模型和DALL-E图像生成模型",
}
results = []
for key, value in knowledge_base.items():
if query.lower() in key.lower() or any(w in key.lower() for w in query.lower().split()):
results.append(f"📎 {key}: {value}")
if not results:
results.append(f"🔍 关于'{query}'的搜索结果: 未找到精确匹配，建议尝试更具体的关键词。")
return "n".join(results[:num_results])
search_tool = Tool(
name="web_search",
description="搜索互联网获取信息，适合查找事实性知识、实时数据等",
func=web_search_func,
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"num_results": {"type": "integer", "description": "返回结果数量，默认3"}
},
"required": ["query"]
}
)
# 工具4: 文件读写工具
def file_ops_func(operation: str, filepath: str, content: str = "") -> str:
"""安全的文件读写操作"""
import os
safe_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "agent_workspace")
os.makedirs(safe_dir, exist_ok=True)
full_path = os.path.join(safe_dir, os.path.basename(filepath))
if operation == "write":
with open(full_path, "w", encoding="utf-8") as f:
f.write(content)
return f"文件已写入: {full_path} (大小: {len(content)}字符)"
elif operation == "read":
if os.path.exists(full_path):
with open(full_path, "r", encoding="utf-8") as f:
data = f.read()
return f"文件内容:n{data[:2000]}" if len(data) > 2000 else f"文件内容:n{data}"
return f"文件不存在: {full_path}"
return "未知操作"
file_tool = Tool(
name="file_operations",
description="读写文件，支持创建新文件和读取已有文件",
func=file_ops_func,
parameters={
"type": "object",
"properties": {
"operation": {"type": "string", "enum": ["read", "write"], "description": "操作类型"},
"filepath": {"type": "string", "description": "文件名或路径"},
"content": {"type": "string", "description": "要写入的内容（仅write操作需要）"}
},
"required": ["operation", "filepath"]
}
)
# 工具注册中心
TOOL_REGISTRY = {
"calculator": calculator_tool,
"datetime_util": datetime_tool,
"web_search": search_tool,
"file_operations": file_tool
}3.3 Agent核心引擎
这是Agent的大脑，负责与LLM交互、解析工具调用请求并管理整个执行循环：
import os
from openai import OpenAI
class MultiToolAgent:
"""多工具AI Agent核心引擎"""
def __init__(self, api_key: str = None, model: str = "gpt-3.5-turbo"):
self.client = OpenAI(
api_key=api_key or os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
self.model = model
self.tools = TOOL_REGISTRY
self.conversation_history = []
self.max_iterations = 8 # 防止无限循环
# 系统提示词，定义Agent的行为规范
self.system_prompt = """你是一个智能AI助手，可以调用多种工具来完成任务。
请遵循以下原则：
1. 仔细分析用户需求，确定需要哪些工具
2. 如果任务复杂，将其分解为多个步骤
3. 每次只调用一个工具，等待结果后再决定下一步
4. 如果工具返回错误，尝试其他方法或向用户说明
5. 最终用自然语言将结果清晰地呈现给用户
6. 在不需要调用工具时，直接给出你的回答
7. 在回答中引用工具返回的具体数据，让用户知道信息来源可靠"""
def _build_messages(self):
"""构建发送给LLM的消息列表"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.conversation_history[-20:]) # 保留最近20条消息
return messages
def _execute_tool_call(self, tool_name: str, arguments: dict) -> str:
"""执行指定的工具调用"""
if tool_name not in self.tools:
return f"错误: 未知工具 '{tool_name}'，可用工具: {list(self.tools.keys())}"
tool = self.tools[tool_name]
print(f" 🔧 调用工具: {tool_name} | 参数: {arguments}")
result = tool.execute(**arguments)
print(f" ✅ 工具返回: {result[:150]}...")
return result
def run(self, user_query: str) -> str:
"""运行Agent处理用户查询"""
print(f"n{'='*60}")
print(f"🚀 Agent启动 | 用户查询: {user_query}")
print(f"{'='*60}")
self.conversation_history.append({"role": "user", "content": user_query})
iteration = 0
while iteration < self.max_iterations:
iteration += 1
print(f"n📍 第{iteration}轮推理...")
try:
response = self.client.chat.completions.create(
model=self.model,
messages=self._build_messages(),
tools=[t.to_openai_format() for t in self.tools.values()],
tool_choice="auto",
temperature=0.3
)
except Exception as e:
return f"LLM调用失败: {str(e)}"
assistant_message = response.choices[0].message
# 检查是否需要调用工具
if assistant_message.tool_calls:
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
# 将Assistant的工具调用请求加入历史
self.conversation_history.append({
"role": "assistant",
"content": None,
"tool_calls": [{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_name,
"arguments": tool_call.function.arguments
}
}]
})
# 执行工具
tool_result = self._execute_tool_call(tool_name, arguments)
# 将工具结果加入历史
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result
})
else:
# 没有工具调用，说明任务完成
final_answer = assistant_message.content or "Agent已完成任务，但未生成文本回复。"
self.conversation_history.append({
"role": "assistant",
"content": final_answer
})
print(f"n🎯 最终回答:n{final_answer}")
return final_answer
return "达到最大迭代次数，Agent未能完成所有任务。请尝试简化您的请求。"
def reset(self):
"""重置对话历史"""
self.conversation_history = []
print("🔄 Agent对话历史已重置")3.4 Agent运行示例
现在让我们实际运行这个Agent，看看它如何处理需要多个工具协作的复杂任务：
# 初始化Agent
agent = MultiToolAgent(model="gpt-3.5-turbo")
# 示例1: 需要搜索+计算器的复合任务
response1 = agent.run("帮我查一下Python最新版本的发布时间，然后计算从那时到今天过了多少天")
print(f"n📋 任务1完成: {response1[:300]}")
# 重置对话
agent.reset()
# 示例2: 需要日期工具+文件操作
response2 = agent.run("获取当前日期时间，然后将其保存到名为current_time.txt的文件中")
print(f"n📋 任务2完成: {response2[:300]}")
# 重置对话
agent.reset()
# 示例3: 复杂多步骤任务
response3 = agent.run(
"先搜索旧金山的天气信息，然后计算sqrt(2024)的值，"
"最后把这两个结果汇总写入一个叫summary.txt的文件"
)
print(f"n📋 任务3完成: {response3[:300]}")四、深入理解Agent的执行循环
Agent的核心是一个ReAct（推理+行动）循环。每一轮迭代中，LLM会收到完整的对话历史（包括之前的工具调用和结果），然后做出决策：是继续调用工具，还是给出最终回答。这个设计有几个关键优势：
- 上下文感知：LLM能看到所有已执行步骤的结果，避免重复操作
- 错误恢复：如果某个工具返回错误，LLM可以尝试替代方案
- 动态规划：根据中间结果灵活调整后续步骤，而非死板地执行预设流程
- 可追踪性：所有工具调用都有记录，便于调试和审计
执行流程可以概括为：用户输入 → LLM分析 → 判断是否需要工具 → [是] 调用工具 → 获取结果 → 反馈给LLM → 继续判断 → [否] 生成最终回答。这个循环会持续到LLM认为任务完成或达到最大迭代次数。
五、生产环境优化建议
上述实现是一个可工作的原型，但要投入生产环境，还需要考虑以下优化：
- 异步工具调用：当多个工具调用之间没有依赖关系时，可以并行执行以提升效率。使用
asyncio和asyncio.gather实现并发调用。 - 工具结果缓存：对相同参数的调用结果进行缓存，避免重复请求外部API（尤其是付费API）。
- 流式输出：使用SSE（Server-Sent Events）将Agent的思考和执行过程实时推送给前端，改善用户体验。
- 安全沙箱：代码执行类工具应在隔离的Docker容器中运行，防止恶意代码影响宿主机。
- 工具权限分级：根据用户角色限制可调用的工具集，敏感操作（如数据库写入）需要额外确认。
- 监控与日志：记录每次工具调用的耗时、成功率和错误信息，建立告警机制。
- 成本控制：设置每日Token消耗上限，对长对话自动压缩历史消息以节约成本。
六、扩展：接入真实外部服务
将模拟工具替换为真实服务非常简单，只需修改工具的执行函数。例如接入真实的搜索引擎API：
# 接入SerpAPI实现真实网页搜索
def real_web_search(query: str, num_results: int = 3) -> str:
import requests
api_key = os.getenv("SERPAPI_KEY")
url = "https://serpapi.com/search"
params = {"q": query, "api_key": api_key, "num": num_results, "engine": "google"}
resp = requests.get(url, params=params)
data = resp.json()
results = []
for item in data.get("organic_results", [])[:num_results]:
results.append(f"📎 {item['title']}: {item['snippet']}n 🔗 {item['link']}")
return "nn".join(results) if results else "未找到相关结果"
# 直接替换工具定义中的func即可
search_tool_v2 = Tool(
name="web_search",
description="使用Google搜索引擎搜索互联网信息",
func=real_web_search,
parameters={...} # 参数定义保持不变
)同样的模式适用于接入数据库查询、邮件发送、Slack通知、GitHub操作等任何外部服务。这种统一接口+插件化的设计让Agent的能力可以无限扩展。
七、总结与展望
本文我们从零搭建了一个完整的多工具AI Agent系统，核心代码不到200行，但已经具备了处理复杂多步骤任务的能力。回顾关键要点：
- ✅ 统一的工具接口设计是Agent可扩展性的基石
- ✅ ReAct循环让LLM能够在推理和行动之间灵活切换
- ✅ 对话历史管理确保Agent拥有完整的上下文感知能力
- ✅ 模块化架构使得替换LLM后端或添加新工具变得非常简单
随着大模型能力的持续提升，AI Agent正在从概念验证走向生产应用。掌握本文介绍的底层原理和实现模式，你将能够构建出适应各种业务场景的智能自动化系统。下一步可以探索的方向包括：多Agent协作（使用CrewAI或AutoGen）、Agent的记忆增强（向量数据库+RAG）、以及Agent的自我反思与学习机制。
动手实践是最好的学习方式，建议你复制本文代码，替换为自己业务场景中的工具，亲身感受AI Agent带来的效率变革。
