AI Agent工具生态与Function Calling完全教程
一、概述
在大语言模型（LLM）的发展历程中，一个关键的转折点是模型从"只读"的文本生成器进化为能够与外部世界交互的智能体（AI Agent）。这一进化的技术基础就是**Function Calling（函数调用）和Tool Use（工具使用）**能力。
传统的LLM只能基于训练数据生成文本回答，无法获取实时信息、执行计算、操作数据库或调用外部API。Function Calling的出现打破了这一限制，使得LLM能够：
- 获取实时数据（天气、股价、新闻等）
- 执行精确计算
- 操作外部系统（发送邮件、创建文档、管理数据库等）
- 与第三方API交互
- 执行多步骤的复杂任务
本教程将从Function Calling的基本原理出发，系统讲解OpenAI和Anthropic两大平台的实现方式，自定义工具开发规范，工具描述设计最佳实践，MCP协议，以及LangChain和CrewAI等框架的工具集成。最后通过一个完整的实战案例，带你构建一个多功能AI Agent。
二、Function Calling / Tool Use 原理与架构
2.1 核心概念
Function Calling的核心思想是让LLM在需要外部信息或操作时，生成一个结构化的函数调用请求，由外部系统执行该函数，并将结果返回给LLM继续处理。
核心组件：
- 工具定义（Tool Definition）：描述可用工具的名称、功能、参数和返回值
- 工具调用请求（Tool Call）：LLM生成的结构化调用指令，包含函数名和参数
- 工具执行（Tool Execution）：外部系统执行实际的函数调用
- 工具结果（Tool Result）：将执行结果返回给LLM
2.2 完整调用流程
用户输入 → LLM分析 → 是否需要工具？
├── 否 → 直接生成回答
└── 是 → 生成工具调用请求（JSON格式）
→ 外部系统解析并执行
→ 将结果返回给LLM
→ LLM基于结果生成最终回答
# Function Calling的完整流程示意
def function_calling_loop(user_message, tools, llm_client):
"""Function Calling完整循环"""
messages = [{"role": "user", "content": user_message}]
while True:
# 1. 调用LLM，传入工具定义
response = llm_client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto" # 让模型自动决定是否调用工具
)
message = response.choices[0].message
# 2. 检查是否有工具调用
if not message.tool_calls:
# 没有工具调用，返回最终回答
return message.content
# 3. 执行所有工具调用
messages.append(message) # 将助手消息（包含tool_calls）添加到对话
for tool_call in message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
# 执行函数
result = execute_function(function_name, arguments)
# 4. 将工具结果添加到对话
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
# LLM基于工具结果生成最终回答
final_response = llm_client.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
2.3 架构模式
同步调用模式：LLM发起调用，等待结果后继续。适用于简单的单工具调用。
异步调用模式：LLM发起调用后不阻塞，可以同时发起多个工具调用。适用于并行执行多个独立工具。
链式调用模式：一个工具的输出作为另一个工具的输入，形成调用链。适用于多步骤任务。
递归调用模式：LLM根据工具结果决定是否需要进一步调用其他工具。适用于复杂的推理任务。
三、OpenAI Function Calling 详解
3.1 基础用法
OpenAI的Function Calling是最早也是最广泛使用的工具调用实现方式。
import openai
import json
# 定义工具
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，默认为摄氏度"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网获取最新信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
},
"num_results": {
"type": "integer",
"description": "返回结果数量，默认为5",
"default": 5
}
},
"required": ["query"]
}
}
}
]
# 工具执行函数
def execute_tool(tool_name, arguments):
if tool_name == "get_weather":
return get_weather(**arguments)
elif tool_name == "search_web":
return search_web(**arguments)
else:
return {"error": f"Unknown tool: {tool_name}"}
def get_weather(city, unit="celsius"):
# 实际实现中这里会调用天气API
return {
"city": city,
"temperature": 25,
"unit": unit,
"condition": "晴",
"humidity": 45
}
def search_web(query, num_results=5):
# 实际实现中这里会调用搜索API
return {
"query": query,
"results": [
{"title": f"搜索结果{i}", "url": f"https://example.com/{i}"}
for i in range(num_results)
]
}
# 使用Function Calling
def chat_with_tools(user_message):
messages = [
{"role": "system", "content": "你是一个有用的AI助手，可以使用工具来获取信息。"},
{"role": "user", "content": user_message}
]
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto"
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
# 有工具调用
messages.append(assistant_message)
for tool_call in assistant_message.tool_calls:
result = execute_tool(
tool_call.function.name,
json.loads(tool_call.function.arguments)
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
# 获取最终回答
final_response = openai.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
return assistant_message.content
# 测试
answer = chat_with_tools("北京今天天气怎么样？")
print(answer)
3.2 并行工具调用
OpenAI支持在单次响应中返回多个工具调用，实现并行执行。
def parallel_tool_calls(user_message):
"""并行工具调用示例"""
messages = [
{"role": "user", "content": user_message}
]
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto",
parallel_tool_calls=True # 启用并行工具调用
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
messages.append(assistant_message)
# 并行执行所有工具调用
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = {}
for tool_call in assistant_message.tool_calls:
future = executor.submit(
execute_tool,
tool_call.function.name,
json.loads(tool_call.function.arguments)
)
futures[tool_call.id] = future
# 收集结果
for tool_call in assistant_message.tool_calls:
result = futures[tool_call.id].result()
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
final_response = openai.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
return assistant_message.content
# 示例：同时查询多个城市天气
answer = parallel_tool_calls("请告诉我北京、上海和深圳今天的天气")
print(answer)
3.3 强制工具调用
通过tool_choice参数可以控制工具调用行为：
# 自动决定是否调用工具（默认）
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="auto"
)
# 强制调用某个特定工具
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice={"type": "function", "function": {"name": "get_weather"}}
)
# 强制调用工具（任意一个）
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="required"
)
# 禁止调用工具
response = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice="none"
)
3.4 流式处理中的Function Calling
def streaming_with_tools(user_message):
"""流式处理中的Function Calling"""
messages = [{"role": "user", "content": user_message}]
# 流式调用
stream = openai.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
stream=True
)
# 收集流式响应
tool_calls = []
content = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
content += delta.content
print(delta.content, end="", flush=True)
if delta.tool_calls:
for tc in delta.tool_calls:
if len(tool_calls) <= tc.index:
tool_calls.append({
"id": "",
"function": {"name": "", "arguments": ""}
})
if tc.id:
tool_calls[tc.index]["id"] = tc.id
if tc.function:
if tc.function.name:
tool_calls[tc.index]["function"]["name"] = tc.function.name
if tc.function.arguments:
tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments
# 如果有工具调用，执行并获取最终结果
if tool_calls:
# 执行工具调用...
pass
return content
四、Anthropic Tool Use 详解
4.1 基础用法
Anthropic的Claude模型通过不同的API格式支持工具使用。
import anthropic
import json
client = anthropic.Anthropic()
# 定义工具（Anthropic格式）
tools = [
{
"name": "get_weather",
"description": "获取指定城市的当前天气信息",
"input_schema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["city"]
}
}
]
def execute_tool(name, input_data):
if name == "get_weather":
return {"temperature": 25, "condition": "晴"}
return {"error": "Unknown tool"}
def chat_with_claude_tools(user_message):
messages = [{"role": "user", "content": user_message}]
while True:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=tools,
messages=messages
)
# 检查是否有工具使用
if response.stop_reason == "tool_use":
# 提取工具调用
tool_results = []
for content_block in response.content:
if content_block.type == "tool_use":
result = execute_tool(content_block.name, content_block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": content_block.id,
"content": json.dumps(result, ensure_ascii=False)
})
# 将助手响应和工具结果添加到消息
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
# 没有工具调用，返回最终回答
return response.content[0].text
answer = chat_with_claude_tools("北京今天天气怎么样？")
print(answer)
4.2 OpenAI兼容层
Anthropic也提供了OpenAI兼容的API端点：
from openai import OpenAI
# 使用Anthropic的OpenAI兼容端点
client = OpenAI(
api_key="your-anthropic-api-key",
base_url="https://api.anthropic.com/v1/"
)
# 使用与OpenAI相同的工具格式
response = client.chat.completions.create(
model="claude-sonnet-4-20250514",
messages=[{"role": "user", "content": "北京天气怎么样？"}],
tools=tools # 使用OpenAI格式的工具定义
)
4.3 Anthropic特有功能
工具选择控制：
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=tools,
tool_choice={"type": "auto"}, # 自动决定
# tool_choice={"type": "any"}, # 强制使用工具
# tool_choice={"type": "tool", "name": "get_weather"}, # 强制使用特定工具
messages=messages
)
Extended Thinking与工具结合：
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=16000,
thinking={
"type": "enabled",
"budget_tokens": 10000
},
tools=tools,
messages=messages
)
# 处理响应中的thinking和tool_use
for block in response.content:
if block.type == "thinking":
print(f"思考过程: {block.thinking}")
elif block.type == "tool_use":
print(f"工具调用: {block.name}({block.input})")
elif block.type == "text":
print(f"回答: {block.text}")
五、自定义工具开发规范
5.1 工具设计原则
- 单一职责：每个工具只做一件事，保持功能单一
- 清晰命名：工具名应准确描述其功能，使用小写字母和下划线
- 完善的描述：为工具和参数提供详细、准确的描述
- 合理的参数：参数类型明确，必填/选填标注清晰
- 错误处理：工具应优雅地处理异常情况，返回有意义的错误信息
- 幂等性：尽可能使工具操作具有幂等性
5.2 工具基类设计
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
import json
class ToolParameter(BaseModel):
"""工具参数定义"""
name: str
type: str
description: str
required: bool = True
enum: Optional[list] = None
default: Any = None
class ToolResult(BaseModel):
"""工具执行结果"""
success: bool
data: Any = None
error: Optional[str] = None
metadata: Dict[str, Any] = Field(default_factory=dict)
class BaseTool(ABC):
"""工具基类"""
@property
@abstractmethod
def name(self) -> str:
"""工具名称"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具描述"""
pass
@property
@abstractmethod
def parameters(self) -> Dict[str, Any]:
"""参数JSON Schema"""
pass
@abstractmethod
def execute(self, **kwargs) -> ToolResult:
"""执行工具"""
pass
def to_openai_tool(self) -> Dict[str, Any]:
"""转换为OpenAI工具格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
def to_anthropic_tool(self) -> Dict[str, Any]:
"""转换为Anthropic工具格式"""
return {
"name": self.name,
"description": self.description,
"input_schema": self.parameters
}
def validate_arguments(self, arguments: Dict[str, Any]) -> bool:
"""验证参数"""
required = self.parameters.get("required", [])
properties = self.parameters.get("properties", {})
# 检查必填参数
for param in required:
if param not in arguments:
raise ValueError(f"缺少必填参数: {param}")
# 检查参数类型
for param_name, param_value in arguments.items():
if param_name in properties:
expected_type = properties[param_name].get("type")
if expected_type == "string" and not isinstance(param_value, str):
raise TypeError(f"参数 {param_name} 应为字符串类型")
elif expected_type == "integer" and not isinstance(param_value, int):
raise TypeError(f"参数 {param_name} 应为整数类型")
elif expected_type == "number" and not isinstance(param_value, (int, float)):
raise TypeError(f"参数 {param_name} 应为数字类型")
return True
def safe_execute(self, **kwargs) -> ToolResult:
"""安全执行工具（带验证和错误处理）"""
try:
self.validate_arguments(kwargs)
return self.execute(**kwargs)
except ValueError as e:
return ToolResult(success=False, error=f"参数错误: {str(e)}")
except TypeError as e:
return ToolResult(success=False, error=f"类型错误: {str(e)}")
except Exception as e:
return ToolResult(success=False, error=f"执行错误: {str(e)}")
5.3 具体工具实现示例
class WeatherTool(BaseTool):
"""天气查询工具"""
@property
def name(self) -> str:
return "get_weather"
@property
def description(self) -> str:
return "获取指定城市的当前天气信息，包括温度、湿度、天气状况等"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'、'New York'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，默认为摄氏度(celsius)"
}
},
"required": ["city"]
}
def execute(self, city: str, unit: str = "celsius") -> ToolResult:
"""执行天气查询"""
# 实际实现中调用天气API
import requests
try:
# 示例：使用wttr.in API
url = f"https://wttr.in/{city}?format=j1"
response = requests.get(url, timeout=10)
data = response.json()
current = data['current_condition'][0]
temp = current['temp_C'] if unit == 'celsius' else current['temp_F']
return ToolResult(
success=True,
data={
"city": city,
"temperature": int(temp),
"unit": unit,
"condition": current['weatherDesc'][0]['value'],
"humidity": int(current['humidity']),
"wind_speed": int(current['windspeedKmph'])
}
)
except Exception as e:
return ToolResult(success=False, error=str(e))
class CalculatorTool(BaseTool):
"""计算器工具"""
@property
def name(self) -> str:
return "calculate"
@property
def description(self) -> str:
return "执行数学计算，支持基本运算、三角函数、对数等"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 '2+3*4'、'sqrt(16)'、'sin(3.14)'"
}
},
"required": ["expression"]
}
def execute(self, expression: str) -> ToolResult:
"""执行数学计算"""
import math
try:
# 安全的数学函数白名单
safe_dict = {
"abs": abs, "round": round, "min": min, "max": max,
"sum": sum, "pow": pow,
"sqrt": math.sqrt, "log": math.log, "log10": math.log10,
"sin": math.sin, "cos": math.cos, "tan": math.tan,
"pi": math.pi, "e": math.e
}
result = eval(expression, {"__builtins__": {}}, safe_dict)
return ToolResult(
success=True,
data={"expression": expression, "result": result}
)
except Exception as e:
return ToolResult(success=False, error=f"计算错误: {str(e)}")
class DatabaseQueryTool(BaseTool):
"""数据库查询工具"""
def __init__(self, connection_string: str):
self.connection_string = connection_string
@property
def name(self) -> str:
return "query_database"
@property
def description(self) -> str:
return "执行SQL查询，获取数据库中的数据。仅支持SELECT查询。"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "SQL查询语句，仅支持SELECT语句"
},
"limit": {
"type": "integer",
"description": "返回结果的最大行数，默认100",
"default": 100
}
},
"required": ["sql"]
}
def execute(self, sql: str, limit: int = 100) -> ToolResult:
"""执行SQL查询"""
import sqlite3
# 安全检查：只允许SELECT语句
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
return ToolResult(
success=False,
error="安全限制：仅支持SELECT查询"
)
# 防止SQL注入
forbidden_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
for keyword in forbidden_keywords:
if keyword in sql_upper:
return ToolResult(
success=False,
error=f"安全限制：不允许使用{keyword}语句"
)
try:
conn = sqlite3.connect(self.connection_string)
cursor = conn.cursor()
# 添加LIMIT限制
if "LIMIT" not in sql_upper:
sql = f"{sql.rstrip(';')} LIMIT {limit}"
cursor.execute(sql)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
conn.close()
return ToolResult(
success=True,
data={
"columns": columns,
"rows": rows,
"count": len(rows)
}
)
except Exception as e:
return ToolResult(success=False, error=str(e))
六、工具描述与Schema设计最佳实践
6.1 描述编写指南
工具描述是LLM决定是否以及如何使用工具的关键依据。好的描述应该：
# ❌ 差的描述
bad_tool = {
"name": "search",
"description": "搜索",
"parameters": {
"type": "object",
"properties": {
"q": {"type": "string"}
}
}
}
# ✅ 好的描述
good_tool = {
"name": "search_knowledge_base",
"description": "搜索内部知识库获取产品文档、FAQ和技术支持信息。当用户询问产品功能、使用方法或遇到技术问题时使用此工具。返回相关文档的标题、摘要和链接。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索查询词，建议使用具体的产品名称或功能关键词。例如：'如何配置SSO'、'API限流策略'"
},
"category": {
"type": "string",
"enum": ["product_docs", "faq", "troubleshooting", "api_reference"],
"description": "文档类别：product_docs(产品文档), faq(常见问题), troubleshooting(故障排除), api_reference(API参考)"
},
"max_results": {
"type": "integer",
"description": "返回结果数量，默认5，最大20",
"default": 5
}
},
"required": ["query"]
}
}
6.2 参数设计原则
- 使用有意义的参数名：
city比c好，start_date比d好 - 提供参数示例：在描述中给出具体示例值
- 使用枚举约束：对于有限选项使用
enum - 设置合理的默认值：减少必填参数数量
- 描述参数格式：对于日期、时间等特殊格式要明确说明
# 参数设计示例
{
"properties": {
"date": {
"type": "string",
"description": "日期，格式为YYYY-MM-DD，例如'2024-01-15'"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "urgent"],
"description": "任务优先级"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "标签列表，例如['bug', 'frontend', 'urgent']"
},
"options": {
"type": "object",
"properties": {
"verbose": {"type": "boolean", "description": "是否输出详细信息"},
"format": {"type": "string", "enum": ["json", "csv", "text"]}
},
"description": "可选配置项"
}
}
}
6.3 避免常见陷阱
# ❌ 描述过于模糊
{"description": "处理数据"}
# ✅ 明确说明功能
{"description": "将CSV文件转换为JSON格式，支持自定义字段映射和数据类型转换"}
# ❌ 参数名不直观
{"properties": {"q": {"type": "string"}, "n": {"type": "integer"}}}
# ✅ 参数名清晰
{"properties": {"query": {"type": "string"}, "max_results": {"type": "integer"}}}
# ❌ 没有说明何时使用
{"description": "查询天气"}
# ✅ 说明使用场景
{"description": "获取指定城市的实时天气信息。当用户询问天气、温度、是否需要带伞等问题时使用。"}
七、多工具编排与并行调用
7.1 工具编排模式
class ToolOrchestrator:
"""工具编排器"""
def __init__(self, tools: Dict[str, BaseTool], llm_client):
self.tools = tools
self.llm = llm_client
self.tool_schemas = [tool.to_openai_tool() for tool in tools.values()]
def execute_plan(self, user_message: str, max_iterations: int = 10):
"""执行工具编排计划"""
messages = [
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": user_message}
]
iteration = 0
while iteration < max_iterations:
iteration += 1
response = self.llm.chat.completions.create(
model="gpt-4",
messages=messages,
tools=self.tool_schemas,
tool_choice="auto"
)
message = response.choices[0].message
if not message.tool_calls:
# 任务完成
return {
"answer": message.content,
"iterations": iteration,
"tool_calls": self._extract_tool_history(messages)
}
# 执行工具调用
messages.append(message)
tool_results = self._execute_tool_calls(message.tool_calls)
for result in tool_results:
messages.append(result)
return {"error": "达到最大迭代次数", "iterations": iteration}
def _execute_tool_calls(self, tool_calls):
"""执行工具调用（支持并行）"""
results = []
# 分析依赖关系，独立的调用可以并行执行
independent_calls = []
dependent_calls = []
# 简化处理：全部并行执行
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
future_to_call = {}
for tool_call in tool_calls:
future = executor.submit(
self._execute_single_tool,
tool_call
)
future_to_call[future] = tool_call
for future in concurrent.futures.as_completed(future_to_call):
tool_call = future_to_call[future]
try:
result = future.result()
results.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
except Exception as e:
results.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps({"error": str(e)}, ensure_ascii=False)
})
return results
def _execute_single_tool(self, tool_call):
"""执行单个工具调用"""
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if name not in self.tools:
return {"error": f"未知工具: {name}"}
tool = self.tools[name]
result = tool.safe_execute(**args)
return result.dict()
def _build_system_prompt(self):
"""构建系统提示"""
tool_descriptions = []
for tool in self.tools.values():
tool_descriptions.append(f"- {tool.name}: {tool.description}")
return f"""你是一个智能助手，可以使用以下工具来完成任务：
{chr(10).join(tool_descriptions)}
使用工具的原则：
1. 只在必要时使用工具，简单问题直接回答
2. 可以同时调用多个独立的工具
3. 基于工具结果给出准确的回答
4. 如果工具返回错误，尝试其他方法或告知用户"""
def _extract_tool_history(self, messages):
"""提取工具调用历史"""
history = []
for msg in messages:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
for tc in msg.tool_calls:
history.append({
"tool": tc.function.name,
"arguments": json.loads(tc.function.arguments)
})
return history
7.2 条件工具调用
class ConditionalToolRouter:
"""条件工具路由器"""
def __init__(self):
self.routes = {}
def add_route(self, condition_fn, tool_name, transform_fn=None):
"""添加路由规则"""
self.routes[tool_name] = {
"condition": condition_fn,
"transform": transform_fn
}
def route(self, tool_name, arguments):
"""根据条件路由工具调用"""
if tool_name in self.routes:
route = self.routes[tool_name]
if route["condition"](arguments):
if route["transform"]:
arguments = route["transform"](arguments)
return True, arguments
return False, arguments
return True, arguments
# 使用示例
router = ConditionalToolRouter()
# 条件：当查询涉及敏感数据时，添加审计日志
def is_sensitive_query(args):
return "salary" in args.get("query", "").lower() or \
"ssn" in args.get("query", "").lower()
def add_audit_params(args):
args["audit"] = True
args["audit_reason"] = "sensitive_data_access"
return args
router.add_route(is_sensitive_query, "query_database", add_audit_params)
八、MCP协议与工具生态
8.1 MCP协议概述
MCP（Model Context Protocol） 是Anthropic推出的一个开放协议，旨在标准化AI模型与外部工具和数据源之间的交互方式。MCP定义了一套标准的通信协议，使得不同的AI模型和工具可以无缝集成。
MCP的核心组件：
- MCP Host：发起连接的AI应用（如Claude Desktop、IDE插件）
- MCP Client：与MCP Server建立连接的客户端
- MCP Server：提供工具和资源的服务端
8.2 MCP Server实现
# 使用mcp库实现MCP Server
from mcp.server import Server
from mcp.types import Tool, TextContent
import json
server = Server("my-tools-server")
@server.list_tools()
async def list_tools():
"""列出所有可用工具"""
return [
Tool(
name="get_weather",
description="获取指定城市的天气信息",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
),
Tool(
name="search_docs",
description="搜索文档",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
"""执行工具调用"""
if name == "get_weather":
city = arguments.get("city", "")
# 调用天气API
result = {"city": city, "temperature": 25, "condition": "晴"}
return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
elif name == "search_docs":
query = arguments.get("query", "")
# 搜索文档
results = [{"title": f"文档: {query}", "content": "相关内容..."}]
return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False))]
else:
raise ValueError(f"Unknown tool: {name}")
# 运行MCP Server
if __name__ == "__main__":
import asyncio
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
asyncio.run(main())
8.3 MCP Client使用
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client
async def use_mcp_tools():
"""使用MCP工具"""
# 连接到MCP Server
async with stdio_client("python", "my_mcp_server.py") as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 列出可用工具
tools = await session.list_tools()
print(f"可用工具: {[t.name for t in tools]}")
# 调用工具
result = await session.call_tool(
"get_weather",
arguments={"city": "北京"}
)
print(f"结果: {result}")
8.4 常用MCP Server
# 使用官方提供的MCP Server
# 文件系统MCP Server
from mcp.server.filesystem import FileSystemServer
fs_server = FileSystemServer(allowed_directories=["/home/user/documents"])
# GitHub MCP Server
from mcp.server.github import GitHubServer
github_server = GitHubServer(token="your-github-token")
# 数据库MCP Server
from mcp.server.database import DatabaseServer
db_server = DatabaseServer(connection_string="sqlite:///data.db")
九、LangChain / CrewAI 工具集成
9.1 LangChain工具集成
LangChain提供了丰富的工具集成框架。
from langchain.tools import BaseTool, StructuredTool, tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from pydantic import BaseModel, Field
from typing import Optional
# 方法1: 使用@tool装饰器
@tool
def search_web(query: str) -> str:
"""搜索互联网获取最新信息"""
# 实现搜索逻辑
return f"搜索结果: {query}"
@tool
def calculate(expression: str) -> str:
"""执行数学计算"""
try:
result = eval(expression)
return f"计算结果: {expression} = {result}"
except Exception as e:
return f"计算错误: {str(e)}"
# 方法2: 使用StructuredTool
class EmailInput(BaseModel):
"""发送邮件的输入"""
to: str = Field(description="收件人邮箱地址")
subject: str = Field(description="邮件主题")
body: str = Field(description="邮件正文")
cc: Optional[str] = Field(default=None, description="抄送邮箱地址")
def send_email(to: str, subject: str, body: str, cc: Optional[str] = None) -> str:
"""发送邮件"""
# 实际实现中调用邮件API
return f"邮件已发送至 {to}，主题: {subject}"
email_tool = StructuredTool.from_function(
func=send_email,
name="send_email",
description="发送电子邮件",
args_schema=EmailInput
)
# 方法3: 继承BaseTool
class DatabaseQueryTool(BaseTool):
name: str = "query_database"
description: str = "查询数据库获取信息"
def _run(self, query: str) -> str:
"""同步执行"""
# 实际实现
return f"查询结果: {query}"
async def _arun(self, query: str) -> str:
"""异步执行"""
return self._run(query)
# 创建Agent
def create_agent():
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search_web, calculate, email_tool, DatabaseQueryTool()]
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有用的AI助手，可以使用各种工具来完成任务。"),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
return agent_executor
# 使用Agent
agent = create_agent()
result = agent.invoke({"input": "帮我搜索一下最新的AI新闻，然后计算一下2的10次方"})
print(result["output"])
9.2 CrewAI工具集成
CrewAI是一个多Agent协作框架，支持复杂的任务编排。
from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
# 定义工具
class ResearchInput(BaseModel):
query: str = Field(description="研究主题")
class ResearchTool(BaseTool):
name: str = "research"
description: str = "深度研究某个主题，收集相关信息和数据"
args_schema: Type[BaseModel] = ResearchInput
def _run(self, query: str) -> str:
# 实际实现
return f"关于'{query}'的研究结果：\n1. 要点1\n2. 要点2\n3. 要点3"
class WritingInput(BaseModel):
topic: str = Field(description="写作主题")
research_data: str = Field(description="研究数据")
class WritingTool(BaseTool):
name: str = "write_article"
description: str = "基于研究数据撰写文章"
args_schema: Type[BaseModel] = WritingInput
def _run(self, topic: str, research_data: str) -> str:
return f"# {topic}\n\n基于研究数据撰写的文章内容..."
# 创建Agent
researcher = Agent(
role="资深研究员",
goal="深入研究指定主题，收集全面准确的信息",
backstory="你是一位经验丰富的研究员，擅长信息收集和分析。",
tools=[ResearchTool()],
verbose=True,
allow_delegation=False
)
writer = Agent(
role="技术作家",
goal="基于研究结果撰写高质量的技术文章",
backstory="你是一位优秀的技术作家，能够将复杂的技术概念用通俗易懂的语言表达。",
tools=[WritingTool()],
verbose=True,
allow_delegation=False
)
# 定义任务
research_task = Task(
description="研究AI Agent工具生态的最新发展",
expected_output="详细的研究报告，包含主要趋势、关键技术和代表性项目",
agent=researcher
)
writing_task = Task(
description="基于研究结果撰写一篇技术博客文章",
expected_output="一篇结构清晰、内容详实的技术博客文章",
agent=writer,
context=[research_task] # 依赖研究任务的输出
)
# 创建团队
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential, # 顺序执行
verbose=True
)
# 执行
result = crew.kickoff()
print(result)
9.3 自定义LangChain工具适配器
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel
class MCPToolAdapter(BaseTool):
"""将MCP工具适配为LangChain工具"""
def __init__(self, mcp_tool, mcp_session):
super().__init__()
self._mcp_tool = mcp_tool
self._mcp_session = mcp_session
self.name = mcp_tool.name
self.description = mcp_tool.description
def _run(self, **kwargs) -> str:
"""同步执行MCP工具"""
import asyncio
async def _execute():
result = await self._mcp_session.call_tool(
self.name,
arguments=kwargs
)
return result
return asyncio.run(_execute())
async def _arun(self, **kwargs) -> str:
"""异步执行MCP工具"""
result = await self._mcp_session.call_tool(
self.name,
arguments=kwargs
)
return result
class OpenAIToolAdapter:
"""将自定义工具适配为OpenAI Function Calling格式"""
@staticmethod
def from_function(func, name=None, description=None):
"""从函数创建OpenAI工具"""
import inspect
sig = inspect.signature(func)
doc = func.__doc__ or ""
properties = {}
required = []
for param_name, param in sig.parameters.items():
param_type = "string" # 默认类型
if param.annotation != inspect.Parameter.empty:
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
param_type = type_map.get(param.annotation, "string")
properties[param_name] = {
"type": param_type,
"description": f"参数 {param_name}"
}
if param.default == inspect.Parameter.empty:
required.append(param_name)
return {
"type": "function",
"function": {
"name": name or func.__name__,
"description": description or doc,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
},
"execute": func
}
十、工具安全与权限控制
10.1 安全设计原则
from enum import Enum
from typing import Set, List, Optional
from dataclasses import dataclass
import hashlib
import time
class Permission(Enum):
"""权限类型"""
READ = "read"
WRITE = "write"
EXECUTE = "execute"
ADMIN = "admin"
class RiskLevel(Enum):
"""风险等级"""
LOW = "low" # 只读查询
MEDIUM = "medium" # 写入操作
HIGH = "high" # 删除/修改操作
CRITICAL = "critical" # 涉及敏感数据或资金操作
@dataclass
class ToolPermission:
"""工具权限配置"""
tool_name: str
permissions: Set[Permission]
risk_level: RiskLevel
requires_confirmation: bool = False
rate_limit: Optional[int] = None # 每分钟最大调用次数
allowed_users: Optional[List[str]] = None
audit_log: bool = True
10.2 权限控制实现
class SecurityManager:
"""安全管理器"""
def __init__(self):
self.permissions = {}
self.rate_limits = {}
self.audit_logs = []
self.user_roles = {}
def register_tool(self, tool_name: str, permission: ToolPermission):
"""注册工具权限"""
self.permissions[tool_name] = permission
def check_permission(self, user_id: str, tool_name: str,
action: str = "execute") -> bool:
"""检查用户是否有权限使用工具"""
if tool_name not in self.permissions:
return False
perm = self.permissions[tool_name]
# 检查用户是否在允许列表中
if perm.allowed_users and user_id not in perm.allowed_users:
self._log_audit(user_id, tool_name, action, "denied", "用户不在允许列表")
return False
# 检查权限类型
required_perm = Permission(action)
if required_perm not in perm.permissions:
self._log_audit(user_id, tool_name, action, "denied", "权限不足")
return False
# 检查频率限制
if perm.rate_limit:
if not self._check_rate_limit(user_id, tool_name, perm.rate_limit):
self._log_audit(user_id, tool_name, action, "denied", "超过频率限制")
return False
# 高风险操作需要确认
if perm.requires_confirmation:
return self._request_confirmation(user_id, tool_name, action)
self._log_audit(user_id, tool_name, action, "allowed")
return True
def _check_rate_limit(self, user_id: str, tool_name: str,
limit: int) -> bool:
"""检查频率限制"""
key = f"{user_id}:{tool_name}"
current_time = time.time()
if key not in self.rate_limits:
self.rate_limits[key] = []
# 清理过期记录
self.rate_limits[key] = [
t for t in self.rate_limits[key]
if current_time - t < 60
]
if len(self.rate_limits[key]) >= limit:
return False
self.rate_limits[key].append(current_time)
return True
def _request_confirmation(self, user_id: str, tool_name: str,
action: str) -> bool:
"""请求用户确认"""
# 实际实现中可能通过UI或消息渠道请求确认
print(f"[安全确认] 用户 {user_id} 请求执行 {tool_name}.{action}")
print(f"风险等级: {self.permissions[tool_name].risk_level.value}")
# 简化实现：直接返回True
return True
def _log_audit(self, user_id: str, tool_name: str, action: str,
result: str, reason: str = ""):
"""记录审计日志"""
log_entry = {
"timestamp": time.time(),
"user_id": user_id,
"tool_name": tool_name,
"action": action,
"result": result,
"reason": reason
}
self.audit_logs.append(log_entry)
class SecureToolWrapper:
"""安全工具包装器"""
def __init__(self, tool: BaseTool, security_manager: SecurityManager):
self.tool = tool
self.security = security_manager
def execute(self, user_id: str, **kwargs):
"""安全执行工具"""
# 1. 权限检查
if not self.security.check_permission(user_id, self.tool.name):
return ToolResult(
success=False,
error="权限不足，无法执行此操作"
)
# 2. 参数验证和清洗
sanitized_args = self._sanitize_args(kwargs)
# 3. 执行工具
result = self.tool.safe_execute(**sanitized_args)
# 4. 记录执行结果
self._log_execution(user_id, sanitized_args, result)
return result
def _sanitize_args(self, args: dict) -> dict:
"""清洗参数，防止注入攻击"""
sanitized = {}
for key, value in args.items():
if isinstance(value, str):
# 移除潜在的注入字符
value = value.replace(";", "").replace("--", "")
value = value.replace("'", "\\'")
value = value[:1000] # 限制长度
sanitized[key] = value
return sanitized
def _log_execution(self, user_id: str, args: dict, result: ToolResult):
"""记录执行日志"""
log = {
"user": user_id,
"tool": self.tool.name,
"args": args,
"success": result.success,
"timestamp": time.time()
}
# 实际实现中写入日志系统
10.3 输入验证与沙箱
class InputValidator:
"""输入验证器"""
@staticmethod
def validate_sql(query: str) -> bool:
"""验证SQL查询安全性"""
query_upper = query.upper().strip()
# 禁止的SQL关键字
forbidden = [
"DROP", "DELETE", "TRUNCATE", "ALTER", "CREATE",
"INSERT", "UPDATE", "GRANT", "REVOKE", "EXEC",
"EXECUTE", "xp_", "sp_"
]
for keyword in forbidden:
if keyword in query_upper:
return False
# 只允许SELECT
if not query_upper.startswith("SELECT"):
return False
return True
@staticmethod
def validate_shell_command(command: str) -> bool:
"""验证Shell命令安全性"""
# 危险命令黑名单
dangerous_commands = [
"rm -rf", "mkfs", "dd if=", "> /dev/",
"chmod 777", "wget", "curl.*|.*sh",
"eval", "exec"
]
import re
for pattern in dangerous_commands:
if re.search(pattern, command, re.IGNORECASE):
return False
return True
@staticmethod
def validate_url(url: str) -> bool:
"""验证URL安全性"""
from urllib.parse import urlparse
parsed = urlparse(url)
# 只允许http和https
if parsed.scheme not in ["http", "https"]:
return False
# 禁止内网地址
blocked_patterns = [
r"127\.\d+\.\d+\.\d+",
r"10\.\d+\.\d+\.\d+",
r"172\.(1[6-9]|2\d|3[01])\.\d+\.\d+",
r"192\.168\.\d+\.\d+",
r"localhost",
r"0\.0\.0\.0"
]
import re
for pattern in blocked_patterns:
if re.search(pattern, parsed.hostname or ""):
return False
return True
class ToolSandbox:
"""工具沙箱"""
def __init__(self, timeout: int = 30, max_memory: int = 100):
self.timeout = timeout # 超时时间（秒）
self.max_memory = max_memory # 最大内存（MB）
def execute_in_sandbox(self, func, *args, **kwargs):
"""在沙箱中执行函数"""
import signal
import resource
def timeout_handler(signum, frame):
raise TimeoutError("执行超时")
# 设置超时
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(self.timeout)
try:
# 限制内存使用
resource.setrlimit(
resource.RLIMIT_AS,
(self.max_memory * 1024 * 1024, self.max_memory * 1024 * 1024)
)
result = func(*args, **kwargs)
return result
except TimeoutError:
return {"error": "执行超时"}
except MemoryError:
return {"error": "内存使用超限"}
finally:
signal.alarm(0) # 取消超时
十一、实战案例：构建多功能AI Agent
11.1 项目概述
我们将构建一个多功能AI助手Agent，它具备以下能力：
- 天气查询
- 网页搜索
- 文件操作
- 数据库查询
- 代码执行
- 邮件发送
11.2 完整实现
import json
import os
import sqlite3
import subprocess
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
# ==================== 工具基类 ====================
@dataclass
class ToolResult:
success: bool
data: Any = None
error: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
class BaseTool(ABC):
@property
@abstractmethod
def name(self) -> str: pass
@property
@abstractmethod
def description(self) -> str: pass
@property
@abstractmethod
def parameters(self) -> Dict[str, Any]: pass
@abstractmethod
def execute(self, **kwargs) -> ToolResult: pass
def to_openai_tool(self) -> Dict[str, Any]:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
def safe_execute(self, **kwargs) -> ToolResult:
try:
return self.execute(**kwargs)
except Exception as e:
return ToolResult(success=False, error=str(e))
# ==================== 具体工具实现 ====================
class WeatherTool(BaseTool):
@property
def name(self) -> str:
return "get_weather"
@property
def description(self) -> str:
return "获取指定城市的当前天气信息"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}
def execute(self, city: str) -> ToolResult:
# 模拟天气API
weather_data = {
"北京": {"temp": 25, "condition": "晴", "humidity": 40},
"上海": {"temp": 28, "condition": "多云", "humidity": 65},
"深圳": {"temp": 30, "condition": "阵雨", "humidity": 80},
}
data = weather_data.get(city, {"temp": 22, "condition": "未知", "humidity": 50})
return ToolResult(
success=True,
data={"city": city, **data, "unit": "celsius"}
)
class WebSearchTool(BaseTool):
@property
def name(self) -> str:
return "web_search"
@property
def description(self) -> str:
return "搜索互联网获取最新信息"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"num_results": {"type": "integer", "description": "结果数量", "default": 5}
},
"required": ["query"]
}
def execute(self, query: str, num_results: int = 5) -> ToolResult:
# 模拟搜索结果
results = [
{"title": f"关于'{query}'的搜索结果{i}",
"url": f"https://example.com/result{i}",
"snippet": f"这是关于{query}的详细信息..."}
for i in range(1, num_results + 1)
]
return ToolResult(success=True, data={"query": query, "results": results})
class FileTool(BaseTool):
@property
def name(self) -> str:
return "file_operation"
@property
def description(self) -> str:
return "执行文件操作，包括读取、写入和列出目录"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["read", "write", "list"],
"description": "操作类型"
},
"path": {"type": "string", "description": "文件路径"},
"content": {"type": "string", "description": "写入内容（仅write操作需要）"}
},
"required": ["operation", "path"]
}
def execute(self, operation: str, path: str, content: str = "") -> ToolResult:
try:
if operation == "read":
with open(path, 'r', encoding='utf-8') as f:
return ToolResult(success=True, data={"content": f.read()})
elif operation == "write":
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
return ToolResult(success=True, data={"message": f"文件已写入: {path}"})
elif operation == "list":
files = os.listdir(path)
return ToolResult(success=True, data={"files": files})
except Exception as e:
return ToolResult(success=False, error=str(e))
class CalculatorTool(BaseTool):
@property
def name(self) -> str:
return "calculate"
@property
def description(self) -> str:
return "执行数学计算"
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
def execute(self, expression: str) -> ToolResult:
import math
safe_dict = {
"abs": abs, "round": round, "min": min, "max": max,
"sqrt": math.sqrt, "log": math.log, "sin": math.sin,
"cos": math.cos, "pi": math.pi, "e": math.e
}
try:
result = eval(expression, {"__builtins__": {}}, safe_dict)
return ToolResult(success=True, data={"expression": expression, "result": result})
except Exception as e:
return ToolResult(success=False, error=str(e))
# ==================== Agent核心 ====================
class MultiToolAgent:
"""多功能AI Agent"""
def __init__(self, llm_client, tools: List[BaseTool]):
self.llm = llm_client
self.tools = {tool.name: tool for tool in tools}
self.tool_schemas = [tool.to_openai_tool() for tool in tools]
self.conversation_history = []
self.system_prompt = self._build_system_prompt()
def _build_system_prompt(self) -> str:
tool_list = "\n".join([
f"- {t.name}: {t.description}" for t in self.tools.values()
])
return f"""你是一个功能强大的AI助手，可以使用以下工具：
{tool_list}
使用原则：
1. 只在需要时使用工具，简单问题直接回答
2. 可以同时调用多个独立工具
3. 基于工具结果给出准确、有帮助的回答
4. 如果工具出错，尝试其他方法或告知用户
5. 保持友好、专业的语气"""
def chat(self, user_message: str) -> str:
"""与Agent对话"""
self.conversation_history.append({
"role": "user", "content": user_message
})
messages = [
{"role": "system", "content": self.system_prompt}
] + self.conversation_history
max_iterations = 5
iteration = 0
while iteration < max_iterations:
iteration += 1
response = self.llm.chat.completions.create(
model="gpt-4",
messages=messages,
tools=self.tool_schemas,
tool_choice="auto"
)
message = response.choices[0].message
if not message.tool_calls:
# 没有工具调用，返回最终回答
self.conversation_history.append({
"role": "assistant", "content": message.content
})
return message.content
# 执行工具调用
messages.append(message)
self.conversation_history.append(message)
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f" 🔧 调用工具: {tool_name}({arguments})")
if tool_name in self.tools:
result = self.tools[tool_name].safe_execute(**arguments)
result_data = result.dict()
else:
result_data = {"error": f"未知工具: {tool_name}"}
tool_message = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result_data, ensure_ascii=False)
}
messages.append(tool_message)
self.conversation_history.append(tool_message)
return "抱歉，处理过程中出现问题。请尝试简化您的请求。"
def reset(self):
"""重置对话历史"""
self.conversation_history = []
# ==================== 使用示例 ====================
def main():
import openai
# 初始化工具
tools = [
WeatherTool(),
WebSearchTool(),
FileTool(),
CalculatorTool()
]
# 创建Agent
agent = MultiToolAgent(openai, tools)
# 测试对话
test_messages = [
"你好！",
"北京今天天气怎么样？",
"帮我搜索一下最新的AI新闻",
"计算一下 (23 + 45) * 67 的结果",
"帮我创建一个文件 test.txt，内容写Hello World"
]
for msg in test_messages:
print(f"\n👤 用户: {msg}")
response = agent.chat(msg)
print(f"🤖 助手: {response}")
print("-" * 50)
if __name__ == "__main__":
main()
11.3 增强功能：对话记忆与上下文管理
class ConversationMemory:
"""对话记忆管理"""
def __init__(self, max_tokens: int = 4000):
self.max_tokens = max_tokens
self.messages = []
self.summaries = []
def add_message(self, role: str, content: str):
"""添加消息"""
self.messages.append({"role": role, "content": content})
self._trim_if_needed()
def get_messages(self, include_summary: bool = True) -> List[Dict]:
"""获取消息列表"""
if include_summary and self.summaries:
summary_text = "之前的对话摘要：" + " ".join(self.summaries[-3:])
return [{"role": "system", "content": summary_text}] + self.messages
return self.messages
def _trim_if_needed(self):
"""如果超出限制，压缩旧消息"""
total_chars = sum(len(m["content"]) for m in self.messages)
if total_chars > self.max_tokens * 4: # 粗略估算
# 将前半部分消息压缩为摘要
old_messages = self.messages[:len(self.messages)//2]
summary = self._summarize(old_messages)
self.summaries.append(summary)
self.messages = self.messages[len(self.messages)//2:]
def _summarize(self, messages: List[Dict]) -> str:
"""生成消息摘要"""
# 实际实现中使用LLM生成摘要
topics = set()
for msg in messages:
if msg["role"] == "user":
topics.add(msg["content"][:50])
return f"讨论了: {', '.join(list(topics)[:3])}"
class EnhancedAgent(MultiToolAgent):
"""增强版Agent，支持记忆和上下文管理"""
def __init__(self, llm_client, tools: List[BaseTool]):
super().__init__(llm_client, tools)
self.memory = ConversationMemory()
self.context = {} # 会话上下文
def chat(self, user_message: str) -> str:
"""增强版对话"""
self.memory.add_message("user", user_message)
messages = [
{"role": "system", "content": self._build_enhanced_system_prompt()}
] + self.memory.get_messages()
# ... 其余逻辑与父类相同
response_text = self._process_messages(messages)
self.memory.add_message("assistant", response_text)
return response_text
def _build_enhanced_system_prompt(self) -> str:
base_prompt = self.system_prompt
# 添加上下文信息
context_info = ""
if self.context:
context_info = "\n当前上下文：\n"
for key, value in self.context.items():
context_info += f"- {key}: {value}\n"
return base_prompt + context_info
def _process_messages(self, messages):
"""处理消息（与父类逻辑相同）"""
# ... 实现细节省略
pass
十二、最佳实践总结
12.1 工具设计最佳实践
- 单一职责：每个工具只做一件事
- 清晰命名：工具名和参数名要直观易懂
- 完善描述：帮助LLM理解何时以及如何使用工具
- 错误处理：优雅处理异常，返回有意义的错误信息
- 幂等设计：重复调用不产生副作用
12.2 安全最佳实践
- 最小权限原则：只授予必要的权限
- 输入验证：验证和清洗所有输入参数
- 审计日志：记录所有工具调用
- 频率限制：防止滥用
- 沙箱执行：高风险操作在沙箱中执行
12.3 性能最佳实践
- 并行调用：独立的工具调用并行执行
- 缓存结果：缓存频繁查询的结果
- 超时控制：设置合理的超时时间
- 批量处理：支持批量操作减少调用次数
12.4 开发流程最佳实践
- 先设计Schema：先定义工具的输入输出Schema
- 单元测试：为每个工具编写单元测试
- 集成测试：测试工具与LLM的集成
- 渐进式开发：从简单工具开始，逐步增加复杂度
- 文档先行：为工具编写清晰的文档
十三、常见问题
Q1: Function Calling不触发怎么办？
可能原因：
- 工具描述不够清晰
- 用户问题不需要工具就能回答
tool_choice设置为none
解决方案：
- 改进工具描述，明确使用场景
- 使用
tool_choice="required"强制调用 - 在系统提示中明确要求使用工具
Q2: 工具参数解析错误怎么办？
可能原因：
- JSON Schema定义不准确
- 参数描述不够清晰
- LLM生成的参数格式错误
解决方案：
- 使用Pydantic模型验证参数
- 提供参数示例
- 添加参数验证和错误提示
Q3: 如何处理工具调用失败？
def robust_tool_execution(tool_call, max_retries=3):
"""健壮的工具执行，支持重试"""
for attempt in range(max_retries):
try:
result = execute_tool(tool_call)
if result.success:
return result
if attempt < max_retries - 1:
print(f"工具调用失败，重试 {attempt + 1}/{max_retries}")
except Exception as e:
if attempt == max_retries - 1:
return ToolResult(success=False, error=f"重试{max_retries}次后仍失败: {str(e)}")
return ToolResult(success=False, error="达到最大重试次数")
Q4: 如何优化工具调用的延迟？
- 使用并行调用
- 缓存常用查询结果
- 使用更快的模型进行简单任务
- 预加载常用数据
Q5: MCP和直接API调用有什么区别？
MCP提供了标准化的协议和接口，使得工具可以跨平台复用。直接API调用更灵活但缺乏标准化。MCP适合构建可复用的工具生态系统，直接API调用适合特定场景的定制化需求。
十四、总结
本教程系统地介绍了AI Agent工具生态与Function Calling的核心技术和实践方法。关键要点：
- Function Calling是LLM与外部世界交互的桥梁，使得AI能够获取实时信息、执行操作
- OpenAI和Anthropic都提供了完善的工具调用支持，但API格式有所不同
- 工具设计需要遵循单一职责、清晰描述、完善错误处理等原则
- MCP协议为工具生态提供了标准化的交互方式
- LangChain和CrewAI等框架简化了工具集成和多Agent协作
- 安全控制是生产环境中不可或缺的部分，包括权限控制、输入验证、审计日志等
随着AI Agent技术的发展，工具生态将变得更加丰富和标准化。掌握这些技术，将帮助你构建更加智能、可靠的AI应用。
