OpenAI Agents Python 项目中的手动执行工具调用方案解析

在开发基于OpenAI Agents Python项目的智能代理时，我们经常会遇到需要将工具调用委托给第三方系统执行的场景。本文将深入探讨如何实现这一需求的技术方案。

核心问题背景

在标准工作流程中，当代理遇到被@function_tool装饰器标记的工具函数时，系统会自动执行这些函数。然而，在某些企业级应用场景下，开发者可能需要：

1. 将工具调用请求转发至外部系统处理
2. 等待第三方系统返回执行结果
3. 手动将结果注入到消息流中

技术解决方案

方案一：使用FunctionTool类直接创建工具

相比使用装饰器，我们可以通过FunctionTool类更灵活地控制工具调用行为：

def run_my_tool(name: str, json_args: str) -> str:
   # 调用第三方系统接口
   return call_3rd_party(name, json) 

# 动态构建工具列表
tool_list = []
for tool in third_party_service.tools():
    async def on_invoke_tool(ctx: RunContextWrapper[Any], input: str) -> str:
        return run_my_tool(ctx.tool_name, input)
 
    tool_list.append(FunctionTool(
        name=tool.name, 
        params_json_schema=tool.json_schema(),
        on_invoke_tool=on_invoke_tool,
    ))

方案二：自定义工具调用拦截器

对于更复杂的场景，可以实现一个调用拦截层：

class ThirdPartyToolDispatcher:
    def __init__(self, original_tools):
        self.original_tools = original_tools
        
    async def dispatch(self, tool_name, arguments):
        # 转发到第三方系统
        result = await third_party_client.execute(tool_name, arguments)
        return result

# 在代理初始化时注入
agent = Agent(
    tools=ThirdPartyToolDispatcher(original_tools),
    ...
)

实现要点

1. 上下文保持：模型期望在发出工具调用后，能在下一个交互轮次中获取执行结果

2. 异步处理：考虑到第三方系统调用的延迟，建议使用异步模式实现

3. 错误处理：需要妥善处理第三方系统调用失败的情况

4. 结果格式化：确保返回结果符合模型预期的格式

最佳实践建议

1. 对于简单的转发需求，直接使用FunctionTool方案即可

2. 对于企业级集成，建议实现中间件层处理认证、重试等复杂逻辑

3. 考虑添加日志记录功能，便于调试和审计

4. 实现结果缓存机制，优化性能

通过以上方案，开发者可以灵活地将OpenAI代理的工具调用与现有企业系统集成，实现更复杂的业务场景。