AI 应用开发实战（5）：AI Agent —— 工具调用、自主决策、多步循环完整实现

前言

前面四篇我们搭建了 AI 应用的基础设施——对话框架、Prompt 工程、多轮记忆、RAG 知识库。这一篇做最后的拼图：让 AI 不仅能"说话"，还能"做事"。

AI Agent 的核心能力就三个字：调用工具。当 AI 能调用搜索引擎查实时信息、调用计算器算数、调用 API 操作数据、调用代码解释器执行脚本时，它就不再只是一个聊天机器人，而是一个数字员工。

本篇从零构建一个 AI Agent 系统，涵盖：

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？

总结

AI Agent 的核心是 “Think → Act → Observe” 循环：

至此，《AI 应用开发实战》系列五篇全部完成 🎉

本文是 《AI 应用开发实战》系列 的第 5 篇。 本文由 Zyentor（智元界） 原创发布

本文发布于 Zyentor（智元界） —— AI 开发者社区 原文链接：https://www.zyentor.com/news/3361