前言
前面四篇我们搭建了 AI 应用的基础设施——对话框架、Prompt 工程、多轮记忆、RAG 知识库。这一篇做最后的拼图：让 AI 不仅能"说话"，还能"做事"。
AI Agent 的核心能力就三个字：调用工具。当 AI 能调用搜索引擎查实时信息、调用计算器算数、调用 API 操作数据、调用代码解释器执行脚本时，它就不再只是一个聊天机器人，而是一个数字员工。
本篇从零构建一个 AI Agent 系统，涵盖：
1. 工具定义与注册机制
2. Function Calling 的完整实现
3. Agent 循环（Think → Act → Observe）
4. 多工具协作与错误恢复
1. 什么是 AI Agent？
传统 LLM： 用户问 → LLM 回答（一次对话，无外部能力）
AI Agent： 用户问 → LLM 思考 → 调用工具 → 观察结果 → 再次思考 → ... → 最终回答
核心区别在于 Agent 可以自主决策调用什么工具、何时调用，并根据工具返回的结果调整后续行动。
┌─────────────────────────────────────────────────┐
│ Agent 循环 │
│ │
│ Think → Act → Observe → Think → Act → ... → Done │
│ 思考 行动 观察 再思考 │
│ │
│ Think: 分析当前状态，决定下一步做什么 │
│ Act: 调用工具或返回最终答案 │
│ Observe: 获取工具执行结果 │
│ Done: 达到目标或无法继续，返回最终答案 │
└─────────────────────────────────────────────────┘
2. 工具定义与注册
2.1 工具接口
首先定义工具的通用接口：
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
import json
class BaseTool(ABC):
"""工具的基类。"""
@property
@abstractmethod
def name(self) -> str:
"""工具名称（在 Agent 中唯一标识）。"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具描述（Agent 根据描述决定是否调用此工具）。"""
pass
@property
@abstractmethod
def parameters(self) -> Dict:
"""工具参数 Schema（OpenAI Function Calling 格式）。"""
pass
@abstractmethod
def execute(self, **kwargs) -> str:
"""执行工具，返回结果字符串。"""
pass
class ToolRegistry:
"""工具注册中心。"""
def __init__(self):
self._tools: Dict[str, BaseTool] = {}
def register(self, tool: BaseTool):
"""注册工具。"""
self._tools[tool.name] = tool
print(f" Registered tool: {tool.name}")
def get(self, name: str) -> Optional[BaseTool]:
"""获取工具。"""
return self._tools.get(name)
def get_all_tools(self) -> List[BaseTool]:
"""获取所有已注册的工具。"""
return list(self._tools.values())
def get_openai_tools(self) -> List[Dict]:
"""获取 OpenAI Function Calling 格式的工具定义。"""
return [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters,
},
}
for tool in self._tools.values()
]
2.2 实现具体工具
计算器工具：
import math
import re
class CalculatorTool(BaseTool):
"""计算器工具——执行数学运算。"""
@property
def name(self) -> str:
return "calculator"
@property
def description(self) -> str:
return "执行数学计算，支持加减乘除、幂运算、三角函数等。适用于需要精确计算的场景。"
@property
def parameters(self) -> Dict:
return {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 '12 * 48 + 36 / 2'",
}
},
"required": ["expression"],
}
def execute(self, expression: str = "") -> str:
try:
# 安全计算：只允许数学表达式
allowed = re.sub(r"[0-9+\-*/.()% ]", "", expression)
if allowed:
return f"Error: Invalid characters: {allowed}"
result = eval(expression, {"__builtins__": {}}, {"math": math})
return f"{expression} = {result}"
except Exception as e:
return f"Calculation error: {e}"
搜索工具（需要搜索引擎 API）：
import requests
import os
class WebSearchTool(BaseTool):
"""搜索引擎工具——搜索实时信息。"""
@property
def name(self) -> str:
return "web_search"
@property
def description(self) -> str:
return "搜索互联网获取实时信息。当需要最新新闻、实时数据、或不确定的事实时应使用此工具。"
@property
def parameters(self) -> Dict:
return {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，应具体且包含核心业务术语",
},
"count": {
"type": "integer",
"description": "返回结果数量（默认 5）",
"default": 5,
},
},
"required": ["query"],
}
def execute(self, query: str = "", count: int = 5) -> str:
try:
api_key = os.getenv("SEARCH_API_KEY", "")
if not api_key:
return "Search API not configured"
# 这里使用 Bing Search API 为例
headers = {"Ocp-Apim-Subscription-Key": api_key}
params = {"q": query, "count": count, "mkt": "zh-CN"}
resp = requests.get(
"https://api.bing.microsoft.com/v7.0/search",
headers=headers,
params=params,
timeout=10,
)
results = resp.json()
snippets = []
for r in results.get("webPages", {}).get("value", [])[:count]:
snippets.append(f"- [{r['name']}]({r['url']}): {r['snippet']}")
return "\n".join(snippets) if snippets else "No results found"
except Exception as e:
return f"Search error: {e}"
文件读写工具：
class FileReadTool(BaseTool):
"""文件读取工具——读取本地文件内容。"""
@property
def name(self) -> str:
return "file_read"
@property
def description(self) -> str:
return "读取本地文件内容。支持 txt、py、md、json 等文本格式。"
@property
def parameters(self) -> Dict:
return {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "文件路径（绝对路径或相对路径）",
}
},
"required": ["file_path"],
}
def execute(self, file_path: str = "") -> str:
try:
# 安全检查：防止读取敏感文件
forbidden = ["/etc/passwd", "/etc/shadow", ".env"]
if any(f in file_path for f in forbidden):
return "Error: Access denied"
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
content = f.read()
if len(content) > 5000:
content = content[:5000] + "\n\n... (truncated)"
return content
except Exception as e:
return f"Read error: {e}"
当前时间工具：
import time
class CurrentTimeTool(BaseTool):
"""时间工具——获取当前日期和时间。"""
@property
def name(self) -> str:
return "current_time"
@property
def description(self) -> str:
return "获取当前日期和时间。当需要知道当前时间、日期、星期几时使用。"
@property
def parameters(self) -> Dict:
return {
"type": "object",
"properties": {
"format": {
"type": "string",
"description": "时间格式（可选：date/time/datetime/timestamp）",
"default": "datetime",
}
},
}
def execute(self, format: str = "datetime") -> str:
now = time.time()
formats = {
"date": time.strftime("%Y-%m-%d"),
"time": time.strftime("%H:%M:%S"),
"datetime": time.strftime("%Y-%m-%d %H:%M:%S"),
"timestamp": str(int(now)),
}
return formats.get(format, formats["datetime"])
2.3 注册工具
registry = ToolRegistry()
registry.register(CalculatorTool())
registry.register(CurrentTimeTool())
registry.register(WebSearchTool())
registry.register(FileReadTool())
# 查看所有已注册工具
for t in registry.get_all_tools():
print(f" {t.name}: {t.description[:50]}...")
3. Agent 核心循环
3.1 单步 Agent
最简单的 Agent——只做一次工具调用决策：
from openai import OpenAI
class SimpleAgent:
"""单步 Agent：接收用户输入，决定是否调用工具，返回结果。"""
def __init__(self, registry: ToolRegistry, llm_client: OpenAI):
self.registry = registry
self.client = llm_client
self.system_prompt = """你是一个 AI 助手，可以使用工具来帮助用户。
在回答问题时，如果需要实时信息或精确计算，请使用对应工具。
如果不需要工具，直接回答即可。"""
def run(self, user_input: str) -> str:
"""处理用户输入。"""
messages = [{"role": "system", "content": self.system_prompt}]
messages.append({"role": "user", "content": user_input})
response = self.client.chat.completions.create(
model=os.getenv("LLM_MODEL", "deepseek-chat"),
messages=messages,
tools=self.registry.get_openai_tools(),
tool_choice="auto",
)
msg = response.choices[0].message
# 如果模型决定调用工具
if msg.tool_calls:
for tc in msg.tool_calls:
tool_name = tc.function.name
tool_args = json.loads(tc.function.arguments)
tool = self.registry.get(tool_name)
if tool:
# 执行工具
result = tool.execute(**tool_args)
messages.append(msg)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
# 调用 LLM 生成最终回答
final = self.client.chat.completions.create(
model=os.getenv("LLM_MODEL", "deepseek-chat"),
messages=messages,
)
return final.choices[0].message.content
# 不需要工具，直接返回
return msg.content
3.2 多步 Agent（完整 Agent 循环）
真正的 Agent 需要多步循环——一次工具调用的结果可能触发下一次工具调用：
class Agent:
"""完整 Agent：支持多步工具调用循环。"""
def __init__(
self,
registry: ToolRegistry,
llm_client: OpenAI,
max_iterations: int = 10,
):
self.registry = registry
self.client = llm_client
self.max_iterations = max_iterations
self.system_prompt = """你是一个能使用工具的 AI 助手。
## 行为规则
1. 分析用户需求，决定是否使用工具
2. 一次调用一个工具，等待结果后再决定下一步
3. 工具返回结果后，分析结果是否满足需求
4. 如果满足，给出最终答案
5. 如果不满足或需要更多信息，继续调用其他工具
6. 如果尝试多次仍无法解决，告知用户当前进展和遇到的困难
7. 所有数字计算必须使用 calculator 工具，不要自己算
## 可用工具
{tools_description}
请按以下格式思考：
Thought: 分析当前情况，决定下一步做什么
Action: 工具名称（如果不需工具则输出 Final Answer）
Action Input: 工具参数 JSON
"""
def run(self, user_input: str, verbose: bool = True) -> str:
"""运行 Agent。"""
tools_desc = "\n".join([
f"- {t.name}: {t.description}（参数：{json.dumps(t.parameters, ensure_ascii=False)}）"
for t in self.registry.get_all_tools()
])
messages = [
{"role": "system", "content": self.system_prompt.format(
tools_description=tools_desc
)},
{"role": "user", "content": user_input},
]
iteration = 0
while iteration str:
"""安全执行工具，捕获所有异常。"""
try:
return tool.execute(**args)
except TypeError as e:
# 参数错误——Agent 传了错误的参数
return f"ParameterError: {e}. Please check the required parameters and types."
except TimeoutError:
return "TimeoutError: The tool execution timed out."
except PermissionError:
return "PermissionError: Access denied."
except Exception as e:
return f"UnexpectedError: {type(e).__name__}: {e}"
def _handle_tool_error(self, error_msg: str, messages: list) -> bool:
"""判断工具错误是否可恢复。"""
recoverable = [
"ParameterError",
"TimeoutError",
]
return any(e in error_msg for e in recoverable)
6. 与前面几篇的集成
将 Agent 集成到系列第一篇的 AI Chat 应用中：
# agent_app.py — 在 AI Chat 中添加 Agent 模式
from agent_core import Agent, ToolRegistry
from tools import (
CalculatorTool, CurrentTimeTool,
WebSearchTool, FileReadTool,
)
from openai import OpenAI
import os
# 初始化 LLM 客户端
client = OpenAI(
api_key=os.getenv("LLM_API_KEY"),
base_url=os.getenv("LLM_BASE_URL"),
)
# 注册工具
registry = ToolRegistry()
registry.register(CalculatorTool())
registry.register(CurrentTimeTool())
registry.register(WebSearchTool())
registry.register(FileReadTool())
# 创建 Agent
agent = Agent(registry, client)
@app.post("/api/agent/chat")
async def agent_chat(request: Request):
"""Agent 对话接口。"""
body = await request.json()
user_message = body.get("message", "")
session_id = body.get("session_id", "default")
if not user_message.strip():
return JSONResponse({"error": "Message required"}, status_code=400)
# 使用 Agent 处理
try:
result = agent.run(user_message, verbose=False)
return JSONResponse({
"reply": result,
"mode": "agent",
})
except Exception as e:
return JSONResponse({
"reply": f"Agent error: {e}",
"mode": "agent",
})
@app.get("/api/agent/tools")
async def list_tools():
"""查看可用的 Agent 工具列表。"""
tools = [
{"name": t.name, "description": t.description}
for t in registry.get_all_tools()
]
return JSONResponse({"tools": tools})
7. Agent 应用的三大模式
根据使用场景，Agent 有三种常见模式：
模式 1：Chat Agent（对话式）
用户 Agent Tools
适合：客服、助手、知识问答
模式 2：Workflow Agent（工作流式）
用户 -> Agent -> Tool1 -> Agent -> Tool2 -> Agent -> 结果
适合：数据处理、报告生成、多步骤操作
模式 3：Orchestrator Agent（编排式）
┌─ Agent A ─ Tool A
用户 ── Orchestrator ── Agent B ─ Tool B ── 最终结果
└─ Agent C ─ Tool C
适合：复杂任务分解、多 Agent 协作
本文实现的是模式 1（Chat Agent），它是最通用、最稳定的模式。
8. 注意事项与最佳实践
8.1 Token 消耗
Agent 循环会消耗大量 Token。每次迭代都包含：历史消息 + 工具定义 + 工具调用 + 工具结果。
# 每次 Agent 调用的 Token 构成
total_tokens = (
system_prompt_tokens # 包含所有工具定义
+ history_tokens # 历史对话
+ tool_call_tokens # 模型生成的 tool call
+ tool_result_tokens # 工具返回的结果
+ final_output_tokens # 最终回答
)
# 估算：一轮 Agent 循环约 2000-5000 tokens
# 5 轮循环约 10000-25000 tokens
优化建议：
- 限制 max_iterations（建议 5-10 轮）
- 工具结果只保留关键字段（用摘要代替完整输出）
- 定期清理历史消息（参考第3篇的上下文截断策略）
8.2 安全注意事项
# 工具调用的安全清单
SAFETY_CHECKLIST = """
□ 文件读写工具：限制可访问的目录范围
□ 网络工具：限制可访问的域名/IP
□ 执行工具：禁止执行 shell 命令（除非明确需要）
□ 敏感操作：需要用户确认才能执行
□ 数据隐私：工具结果中不要泄露敏感信息
"""
8.3 何时应该用 Agent？
| 适合 Agent | 不适合 Agent |
|---|---|
| 需要搜索实时信息 | 纯知识问答（RAG 更适合） |
| 需要精确计算 | 简单对话 |
| 多步骤操作 | 单步任务 |
| 需要操作外部系统 | 需要极低延迟 |
| 不确定的复杂任务 | 高安全/高可靠性要求 |
总结
AI Agent 的核心是 "Think → Act → Observe" 循环：
- 工具是关键——Agent 的能力边界由工具决定，不是由模型决定
- 描述是灵魂——工具描述的质量直接影响 Agent 的决策准确性
- 循环是引擎——多步循环让 Agent 能完成复杂任务
- 容错是保障——真实环境中错误必然发生，优雅处理很重要
至此，《AI 应用开发实战》系列五篇全部完成 🎉
| 篇目 | 标题 |
|---|---|
| 第1篇 | 从零搭建你的第一个 AI 应用 |
| 第2篇 | Prompt 工程实战 |
| 第3篇 | 多轮对话进阶 |
| 第4篇 | 从零实现 RAG 系统 |
| 第5篇 | AI Agent——工具调用与自主决策 ← 你在这里 |
本文是 《AI 应用开发实战》系列 的第 5 篇。
本文由 Zyentor（智元界） 原创发布