AI 应用开发实战（5）：AI Agent —— 工具调用、自主决策、多步循环完整实现
·
前言
前面四篇我们搭建了 AI 应用的基础设施——对话框架、Prompt 工程、多轮记忆、RAG 知识库。这一篇做最后的拼图：让 AI 不仅能"说话"，还能"做事"。
AI Agent 的核心能力就三个字：调用工具。当 AI 能调用搜索引擎查实时信息、调用计算器算数、调用 API 操作数据、调用代码解释器执行脚本时，它就不再只是一个聊天机器人，而是一个数字员工。
本篇从零构建一个 AI Agent 系统，涵盖：
- 工具定义与注册机制
- Function Calling 的完整实现
- Agent 循环（Think → Act → Observe）
- 多工具协作与错误恢复
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 < self.max_iterations:
iteration += 1
if verbose:
print(f"\n{'='*40} Iteration {iteration} {'='*40}")
# 调用 LLM
response = self.client.chat.completions.create(
model=os.getenv("LLM_MODEL", "deepseek-chat"),
messages=messages,
tools=self.registry.get_openai_tools(),
tool_choice="auto",
temperature=0.3, # Agent 场景用低温度，提高确定性
)
msg = response.choices[0].message
if verbose:
if msg.content:
print(f" Thought: {msg.content[:200]}")
if msg.tool_calls:
for tc in msg.tool_calls:
print(f" Action: {tc.function.name}({tc.function.arguments})")
# 检查是否最终回答
if not msg.tool_calls:
return msg.content or "I'm not sure how to proceed."
# 执行工具调用
messages.append(msg)
for tc in msg.tool_calls:
tool_name = tc.function.name
try:
tool_args = json.loads(tc.function.arguments)
except json.JSONDecodeError:
tool_args = {}
tool = self.registry.get(tool_name)
if tool:
if verbose:
print(f" Executing: {tool_name}...")
result = tool.execute(**tool_args)
if verbose:
print(f" Result: {result[:150]}...")
else:
result = f"Error: Unknown tool '{tool_name}'"
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": str(result)[:2000],
})
# 超过最大迭代次数
return "I've reached the maximum number of steps. Here's what I've found so far."
3.3 Agent 执行示例
用户：2023 年诺贝尔物理学奖得主是谁？他现在多少岁了？
Agent 执行过程：
Iteration 1:
Thought: 我需要搜索 2023 年诺贝尔物理学奖的信息
Action: web_search({"query": "2023 Nobel Prize in Physics winner"})
Result: 2023 年诺贝尔物理学奖授予皮埃尔·阿戈斯蒂尼、费伦茨·克劳斯和安娜·吕利耶...
Iteration 2:
Thought: 我需要知道获奖者的出生年份来算年龄
Action: web_search({"query": "Pierre Agostini birth date"})
Result: Pierre Agostini born 1941
Iteration 3:
Thought: 知道了出生年份，计算当前年龄
Action: calculator({"expression": "2026 - 1941"})
Result: 2026 - 1941 = 85
Final Answer: 2023 年诺贝尔物理学奖得主之一是皮埃尔·阿戈斯蒂尼，他出生于 1941 年，截至 2026 年已经 85 岁。
4. 工具描述的艺术
工具描述决定了 Agent 能不能正确调用工具。这是整个 Agent 系统最关键的设计点。
4.1 好的描述 vs 差的描述
# 差的描述——太模糊
{
"name": "search",
"description": "搜索信息",
}
# 好的描述——明确说清楚什么时候该用、什么时候不该用
{
"name": "web_search",
"description": "搜索互联网获取实时信息。当用户问及最新新闻、实时数据、\
当前事件、或你无法确定的事实时应使用此工具。\
不要用于简单的常识问题（如"中国的首都是什么"），直接回答即可。",
}
4.2 参数描述的关键
# 差的参数描述
{
"query": {
"type": "string",
"description": "搜索关键词",
}
}
# 好的参数描述——告诉 Agent 应该传什么内容、什么风格
{
"query": {
"type": "string",
"description": "搜索关键词。应该包含核心业务术语，\
如搜索「Python 异步编程最佳实践」而非「Python 怎么用 async」。\
使用精确的关键词组合，避免过于宽泛的词汇。",
}
}
4.3 工具描述原则总结
| 原则 | 说明 | 示例 |
|---|---|---|
| 说清用途 | 明确工具能做什么 | “执行数学计算，支持加减乘除、幂运算” |
| 说清边界 | 明确工具不能做什么 | “不要用于文本处理或逻辑判断” |
| 说清时机 | 明确什么场景应该调用 | “当需要实时信息时应使用此工具” |
| 说清参数 | 参数描述要具体 | “搜索关键词，要求精确术语” |
5. 错误处理与容错
Agent 在真实环境中一定会遇到各种错误。好的 Agent 应该能优雅地处理：
class RobustAgent(Agent):
"""带错误处理的 Agent。"""
def _execute_tool_safely(self, tool: BaseTool, args: dict) -> 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（智元界） 原创发布
本文发布于 Zyentor（智元界） —— AI 开发者社区
原文链接：https://www.zyentor.com/news/3361
更多推荐
11
0- 0
已为社区贡献20条内容
所有评论(0)