当前位置：首页 > java >正文

第六章：Tool and LLM Integration

java 2025/5/2 18:13:07

Chapter 6: Tool and LLM Integration

从执行流到工具集成：如何让AI“调用真实世界的技能”？

在上一章的执行流框架中，我们已经能让多个代理协作完成复杂任务。但你是否想过：如果用户要求“查询实时天气”或“打开网页搜索”，AI如何调用真实世界的工具（比如天气API、浏览器）？就像给指挥家配备更多乐器，让整个“交响乐团”能力更强！本章将介绍如何通过工具与LLM集成层，让AI无缝调用外部服务。

核心问题：如何让AI“听懂”不同服务商的语言？

想象你是一位需要跨国合作的翻译官：

亚马逊Bedrock要求用“工具使用ID”格式说话
OpenAI要求用“函数调用”格式说话
工具API需要特定参数格式

工具集成层就像这位翻译官，通过以下机制实现：
✅ 将不同LLM的格式统一转换
✅ 在调用失败时重试或降级
✅ 支持实时流式输出

核心概念分解

1. 协议转换器：像翻译官一样转换“语言”

就像把中文翻译成英文，协议转换器负责：

# 将OpenAI的函数调用格式转为Bedrock格式（简化示例）
def convert_openai_to_bedrock(tools):bedrock_tools = []for tool in tools:bedrock_tool = {"name": tool["name"],"description": tool["description"],"inputSchema": tool["parameters"]  # 统一参数格式}bedrock_tools.append(bedrock_tool)return bedrock_tools

2. 错误处理机制：像安全网一样兜底

当调用失败时自动：

尝试其他服务商（如Bedrock→OpenAI）
返回友好提示而非崩溃

# 简化版错误处理逻辑
try:response = call_external_api()
except APIError:return "当前服务繁忙，请稍后再试"

3. 流式与非流式API：像直播与录播两种模式

流式模式：实时返回文字（适合聊天）
非流式模式：一次性返回完整结果（适合计算任务）

# 流式调用示例
async def stream_api_call():for chunk in response_stream:yield chunk  # 逐块返回内容

如何用集成层解决问题？

场景：调用天气API获取实时数据

步骤1：定义工具配置

# config.toml中添加天气工具配置
[tools.weather_api]
name = "天气查询"
description = "获取实时天气数据"
parameters = {"location": "城市名称","unit": "返回温度单位（C/F）"
}

步骤2：创建工具调用类

class WeatherTool:async def call(self, location: str, unit: str = "C"):# 模拟调用外部APIreturn f"{location}当前温度25°C，天气晴朗"

步骤3：在代理中集成

# 在代理的run方法中调用
tool_response = await WeatherTool().call(user_query.location)
memory.add_message(Message.tool_message(content=tool_response,name="天气工具",tool_call_id="tool_001"
))

此时，用户就能在对话中看到类似这样的回答：

用户：上海今天天气如何？
AI：上海当前温度25°C，天气晴朗（数据来自天气工具）

内部实现揭秘：工具调用的“翻译过程”

流程分解（用简单序列图演示）

关键代码解析（app/bedrock.py片段）

# 将Bedrock响应转为OpenAI格式的简化示例
def _convert_bedrock_response(self, bedrock_response):openai_response = {"choices": [{"message": {"content": bedrock_response["output"]["content"],"role": "assistant"}}]}return OpenAIResponse(openai_response)