1. 项目概述：构建基于OpenAI工具调用的AI代理系统
在当今AI技术快速发展的背景下，如何构建一个能够可靠执行复杂任务的AI代理系统成为了许多开发者的关注焦点。本文将详细介绍如何从零开始构建一个基于OpenAI工具调用的AI代理系统，特别聚焦于费用跟踪这一具体应用场景。
这个项目源于一个实际需求：许多小型企业和个人都需要一个智能助手来帮助他们管理日常财务交易。传统的方式要么需要手动记录，要么依赖复杂的财务软件，而我们的目标是开发一个能够通过自然语言交互就能完成这些任务的AI代理。
1.1 核心需求解析
这个AI代理系统需要满足以下几个核心需求：
- 自然语言交互：用户能够用日常语言描述财务交易，如"今天买咖啡花了5美元，税率0.2"。
- 自动数据补全：系统能够自动计算缺失的财务数据（如总金额）并补充日期等信息。
- 错误处理机制：当信息不完整时，能够优雅地处理并请求补充信息，而不是虚构数据。
- 可扩展架构：系统设计应该能够方便地扩展到其他类型的任务，而不仅限于费用跟踪。
1.2 技术选型考量
在技术选型上，我们主要基于以下几个考量：
- OpenAI API：选择OpenAI的GPT-3.5-turbo模型作为核心，因其在自然语言理解和生成方面的卓越表现。
- Pydantic：用于数据验证和模型定义，确保输入数据的结构正确性。
- LangChain：虽然本项目没有完全使用LangChain框架，但借鉴了其工具调用转换的思想。
- Python：作为主要开发语言，因其丰富的AI和数据科学生态系统。
提示：在实际开发中，我们建议从最简单的可行方案开始，然后逐步添加复杂性，而不是一开始就设计过于复杂的架构。
2. 核心架构设计与实现
2.1 数据模型定义
首先，我们需要定义核心的数据模型。对于费用跟踪系统，我们定义了两个主要的Pydantic模型：
python复制from pydantic.v1 import BaseModel
from datetime import datetime
class Expense(BaseModel):
description: str
net_amount: float
gross_amount: float
tax_rate: float
date: datetime
class Report(BaseModel):
report: str
这些模型不仅定义了数据结构，还将在后续的工具调用中作为类型提示和验证机制。
2.2 工具调用机制
OpenAI的工具调用功能允许模型在对话过程中决定何时以及如何调用外部工具。我们使用convert_to_openai_tool函数将Pydantic模型转换为OpenAI工具定义：
python复制from langchain_core.utils.function_calling import convert_to_openai_tool
add_expense_tool = convert_to_openai_tool(Expense)
report_tool = convert_to_openai_tool(Report)
这个转换过程会自动生成工具的名称、描述和参数定义，使模型能够理解如何调用这些工具。
2.3 系统消息设计
系统消息对于引导模型行为至关重要。我们设计了如下系统消息：
python复制SYSTEM_MESSAGE = """You are tasked with completing specific objectives and
must report the outcomes. At your disposal, you have a variety of tools,
each specialized in performing a distinct type of task.
For successful task completion:
Thought: Consider the task at hand and determine which tool is best suited
based on its capabilities and the nature of the work.
Use the report_tool with an instruction detailing the results of your work.
If you encounter an issue and cannot complete the task:
Use the report_tool to communicate the challenge or reason for the
task's incompletion.
You will receive feedback based on the outcomes of
each tool's task execution or explanations for any tasks that
couldn't be completed. This feedback loop is crucial for addressing
and resolving any issues by strategically deploying the available tools.
"""
这条系统消息明确了代理的角色、可用的工具以及预期的行为模式，特别是强调了错误处理和反馈机制。
3. 核心功能实现与优化
3.1 初始实现与问题发现
我们的初始实现直接调用OpenAI API并处理响应：
python复制from openai import OpenAI
client = OpenAI()
model_name = "gpt-3.5-turbo-0125"
messages = [
{"role":"system", "content": SYSTEM_MESSAGE},
{"role":"user", "content": "I have spend 5$ on a coffee today please track my expense. The tax rate is 0.2."}
]
response = client.chat.completions.create(
model=model_name,
messages=messages,
tools=[add_expense_tool, report_tool]
)
然而，测试中发现了两个主要问题：
- gross_amount未被自动计算
- 日期被虚构而非使用当前日期
3.2 工具类设计与优化
为了解决这些问题，我们设计了更强大的Tool类：
python复制from pydantic import BaseModel
from typing import Type, Callable, Dict, Any, List
class ToolResult(BaseModel):
content: str
success: bool
class Tool(BaseModel):
name: str
model: Type[BaseModel]
function: Callable
validate_missing: bool = False
def run(self, **kwargs) -> ToolResult:
if self.validate_missing:
missing_values = self.validate_input(**kwargs)
if missing_values:
content = f"Missing values: {', '.join(missing_values)}"
return ToolResult(content=content, success=False)
result = self.function(**kwargs)
return ToolResult(content=str(result), success=True)
def validate_input(self, **kwargs) -> List[str]:
missing_values = []
for key in self.model.__fields__.keys():
if key not in kwargs:
missing_values.append(key)
return missing_values
@property
def openai_tool_schema(self) -> Dict[str, Any]:
schema = convert_to_openai_tool(self.model)
if "required" in schema["function"]["parameters"]:
del schema["function"]["parameters"]["required"]
return schema
这个Tool类提供了几个关键功能：
- 输入验证（可选）
- 工具执行
- 结果标准化
- OpenAI工具模式生成（自动移除required约束）
3.3 工具实现示例
我们实现了几个具体的工具：
python复制def add_expense_func(**kwargs):
return f"Added expense: {kwargs} to the database."
add_expense_tool = Tool(
name="add_expense_tool",
model=Expense,
function=add_expense_func
)
def report_func(report: str = None):
return f"Reported: {report}"
report_tool = Tool(
name="report_tool",
model=Report,
function=report_func
)
def get_current_date():
return datetime.now().strftime("%Y-%m-%d")
get_date_tool = Tool(
name="get_current_date",
model=BaseModel,
function=get_current_date,
validate_missing=False
)
4. 代理工作流实现
4.1 代理类设计
我们设计了OpenAIAgent类来管理整个工作流：
python复制class StepResult(BaseModel):
event: str
content: str
success: bool
class OpenAIAgent:
def __init__(self, tools, client, system_message=SYSTEM_MESSAGE,
model_name="gpt-3.5-turbo-0125", max_steps=5, verbose=True):
self.tools = tools
self.client = client
self.model_name = model_name
self.system_message = system_message
self.step_history = []
self.max_steps = max_steps
self.verbose = verbose
def run(self, user_input: str):
openai_tools = [tool.openai_tool_schema for tool in self.tools]
self.step_history = [
{"role":"system", "content":self.system_message},
{"role":"user", "content":user_input}
]
step_result = None
i = 0
while i < self.max_steps:
step_result = self.run_step(self.step_history, openai_tools)
if step_result.event == "finish":
break
elif step_result.event == "error":
print(f"Error: {step_result.content}")
else:
print(f"Tool Result: {step_result.content}")
i += 1
return step_result.content
def run_step(self, messages, tools):
response = self.client.chat.completions.create(
model=self.model_name,
messages=messages,
tools=tools
)
self.step_history.append(response.choices[0].message)
if not response.choices[0].message.tool_calls:
return StepResult(event="Error", content="No tool calls were returned.", success=False)
tool_name = response.choices[0].message.tool_calls[0].function.name
tool_kwargs = parse_function_args(response)
tool_result = run_tool_from_response(response, tools=self.tools)
tool_result_msg = self.tool_call_message(response, tool_result)
self.step_history.append(tool_result_msg)
if tool_result.success:
return StepResult(event="tool_result", content=tool_result.content, success=True)
else:
return StepResult(event="error", content=tool_result.content, success=False)
def tool_call_message(self, response, tool_result):
tool_call = response.choices[0].message.tool_calls[0]
return {
"tool_call_id": tool_call.id,
"role": "tool",
"name": tool_call.function.name,
"content": tool_result.content,
}
4.2 工作流执行示例
让我们看一个完整的执行示例：
python复制tools = [add_expense_tool, report_tool, get_date_tool]
agent = OpenAIAgent(tools, client)
result = agent.run("I have spent 5$ on a coffee today please track my expense. The tax rate is 0.2.")
执行过程会输出类似以下内容：
code复制Tool Call: get_current_date
Args: {}
Tool Result: 2024-03-15
Tool Call: add_expense_tool
Args: {'description': 'Coffee', 'net_amount': 5, 'tax_rate': 0.2, 'date': '2024-03-15'}
Error: Missing values: gross_amount
Tool Call: add_expense_tool
Args: {'description': 'Coffee', 'net_amount': 5, 'tax_rate': 0.2, 'date': '2024-03-15', 'gross_amount': 6}
Tool Result: Added expense: {'description': 'Coffee', 'net_amount': 5, 'tax_rate': 0.2, 'date': '2024-03-15', 'gross_amount': 6} to the database.
Final Result: Reported: Expense successfully tracked for coffee purchase.
5. 关键技术与经验分享
5.1 处理缺失信息的策略
在开发过程中，我们发现处理缺失信息是一个关键挑战。我们的解决方案包括：
- 将工具参数设为可选：通过从工具模式中移除required字段，我们允许模型只提供它知道的信息。
- 显式验证：Tool类可以配置为验证输入，明确报告缺失的字段。
- 渐进式补全：代理可以通过多次工具调用来逐步补全缺失的信息。
5.2 工具调用模式设计
有效的工具调用模式需要考虑以下几点：
- 工具粒度：每个工具应该专注于一个具体的任务，而不是试图做太多事情。
- 错误处理：工具应该能够明确报告失败原因，而不仅仅是返回失败状态。
- 结果标准化：所有工具应该返回统一格式的结果，便于代理处理。
5.3 代理工作流优化
在实现代理工作流时，我们总结了以下经验：
- 限制最大步骤数：防止无限循环，设置合理的max_steps。
- 完整的上下文维护：确保每一步的工具调用和结果都添加到对话历史中。
- 清晰的终止条件：定义明确的成功和失败条件。
6. 实际应用与扩展
6.1 费用跟踪场景的完整实现
在实际的费用跟踪应用中，我们可以扩展这个基础架构：
- 数据库集成：替换add_expense_func为实际的数据库操作。
- 数据验证：在工具执行前添加更复杂的数据验证逻辑。
- 用户确认：在重要操作前添加确认步骤。
6.2 扩展到其他领域
这个架构可以轻松扩展到其他领域：
- 日历管理：添加创建、修改和查询日历事件的工具。
- 任务管理：实现任务创建、分配和状态更新的工具。
- 客户关系管理：开发客户信息管理和交互记录的工具。
6.3 性能优化考虑
对于生产环境，还需要考虑：
- 异步执行：将工具调用改为异步以提高性能。
- 缓存机制：缓存常用工具的结果。
- 限流处理：实现API调用限流以避免超额费用。
7. 常见问题与解决方案
在实际开发和测试过程中，我们遇到了以下典型问题及解决方案：
| 问题描述 | 根本原因 | 解决方案 |
|---|---|---|
| 模型虚构缺失字段 | 工具模式中将字段标记为required | 从工具模式中移除required字段 |
| 工具调用顺序不合理 | 系统消息中缺乏明确的指导 | 在系统消息中添加工具使用优先级说明 |
| 循环调用工具无法终止 | 缺乏明确的终止条件 | 设置最大步骤数并检测重复操作 |
| 计算错误（如税额） | 模型数学计算能力有限 | 在工具中实现计算逻辑而非依赖模型 |
| 时区处理不一致 | 未明确指定时区 | 在日期相关工具中强制使用时区感知的datetime |
8. 开发经验与最佳实践
基于这个项目的开发经验，我们总结了以下最佳实践：
- 从简单开始：先实现最小可行产品，然后逐步添加功能。
- 模块化设计：保持工具和代理逻辑分离，便于维护和扩展。
- 全面日志记录：记录完整的交互历史以便调试和分析。
- 自动化测试：为各种边界情况编写测试用例。
- 用户反馈循环：在实际使用中收集反馈并持续改进。
注意：在实际部署时，务必添加适当的错误处理和日志记录，这对调试和维护至关重要。
9. 未来发展方向
这个基础架构可以朝多个方向发展：
- 多代理协作：实现多个专业代理协同完成复杂任务。
- 长期记忆：添加记忆机制使代理能够从历史交互中学习。
- 动态工具加载：支持运行时添加和移除工具而不重启代理。
- 可视化监控：开发仪表板实时监控代理状态和性能。
- 安全增强：添加身份验证和授权机制。
10. 项目总结与资源
通过本项目，我们实现了一个基于OpenAI工具调用的可靠AI代理系统，特别针对费用跟踪场景进行了优化。关键成就包括：
- 设计了灵活的工具调用架构
- 实现了优雅的错误处理机制
- 构建了可扩展的代理工作流
- 解决了实际应用中的多个挑战
项目完整代码已开源，开发者可以参考并扩展这个基础架构来实现自己的AI代理应用。这个项���不仅展示了当前AI技术的应用潜力，也为未来更复杂的AI系统奠定了基础。