18231 字
约 60 分钟
1
第3章 你的第一个Agent-天气查询助手
2026-09-18
第3章 你的第一个Agent-天气查询助手
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch03-weather-agent.html 所属：第一篇-Agent基础
第一篇：Agent 基础 📌 本章目标
不看理论，先动手！用 50 行代码构建一个能查天气的 Agent，在实战中理解"感知→决策→行动→观察"循环。后续章节会逐步改进这个 Agent。 🚀 小白先完成一个最小闭环
如果你是第一次写 Agent，不要一上来追求"多工具、多轮对话、异常重试"。先只完成下面四步：
- 安装依赖并准备 API Key
- 定义一个
get_weather工具 - 把工具传给模型，让模型决定是否调用
- 看到终端里打印出工具入参与最终回答
只要这四步跑通，你就已经真正做出了第一个 Agent。后面的多轮记忆、异常处理、调用链分析，都是在这个最小闭环上渐进增强。 🎤 这一章的真实面试追问
- 为什么要限制 Agent 的最大循环次数，而不是让它一直试？
- 工具调用失败时，应该重试工具、重试整轮，还是直接降级回答？
- 如果模型没有触发工具，而是直接胡说八道，你会怎么排查？
3.1 从一个问题开始
假设你想问："北京明天天气怎么样？"
传统程序怎么做？写一个函数：get_weather("北京", "明天")，返回结果。但你还需要：解析用户输入、判断意图、处理"明天是几号"、格式化输出……
Agent怎么做？你只需要给 AI 一个工具（get_weather），AI 自己判断意图、组装参数、调用工具、格式化结果。
3.2 准备工作
3.2.1 安装 Python 环境
# 安装 OpenAI SDK
pip install openai
# 确认 Python 版本（需要 3.8+）
python --version
3.2.2 获取 API Key
你需要一个 LLM API Key。本章用 OpenAI 兼容接口为例（支持 OpenAI、DeepSeek、Qwen 等）：
# 环境变量设置（选一个你有的）
export OPENAI_API_KEY="sk-xxx" # OpenAI
export OPENAI_BASE_URL="https://api.openai.com/v1"
# 或者用 DeepSeek（更便宜）
export OPENAI_API_KEY="sk-xxx"
export OPENAI_BASE_URL="https://api.deepseek.com/v1"
3.3 定义工具：给 AI 一双手
Agent 的核心是"AI + 工具"。我们先定义一个天气查询工具：
import json
from datetime import datetime, timedelta
def get_weather(city: str, date: str = "today") -> dict:
"""
查询指定城市的天气
Args:
city: 城市名称，如 "北京"、"上海"
date: 日期，"today"、"tomorrow" 或 "YYYY-MM-DD" 格式
Returns:
包含天气信息的字典
"""
# 模拟天气数据（实际项目调用天气 API）
weather_data = {
"北京": {"today": ("晴", 25), "tomorrow": ("多云", 23)},
"上海": {"today": ("小雨", 28), "tomorrow": ("阴", 27)},
"广州": {"today": ("雷阵雨", 31), "tomorrow": ("晴", 33)},
}
# 处理日期
if date == "today":
date_key = "today"
elif date == "tomorrow":
date_key = "tomorrow"
else:
# 如果是具体日期，简化为 today
date_key = "today"
if city not in weather_data:
return {"error": f"暂不支持查询 {city} 的天气"}
weather, temp = weather_data[city][date_key]
return {
"city": city,
"date": date,
"weather": weather,
"temperature": f"{temp}°C",
}
# 测试工具
print(get_weather("北京", "tomorrow"))
# {'city': '北京', 'date': 'tomorrow', 'weather': '多云', 'temperature': '23°C'}
import { config } from 'dotenv';
config();
// 模拟天气数据
const weatherData: Record> = {
"北京": { today: ["晴", 25], tomorrow: ["多云", 23] },
"上海": { today: ["小雨", 28], tomorrow: ["阴", 27] },
"广州": { today: ["雷阵雨", 31], tomorrow: ["晴", 33] },
};
interface WeatherResult {
city: string;
date: string;
weather: string;
temperature: string;
error?: string;
}
function getWeather(city: string, date: string = "today"): WeatherResult | { error: string } {
if (!weatherData[city]) {
return { error: `暂不支持查询 ${city} 的天气` };
}
const dateKey = date === "tomorrow" ? "tomorrow" : "today";
const [weather, temp] = weatherData[city][dateKey];
return {
city,
date,
weather,
temperature: `${temp}°C`,
};
}
// 测试工具
console.log(getWeather("北京", "tomorrow"));
// { city: '北京', date: 'tomorrow', weather: '多云', temperature: '23°C' }
package main
import (
"encoding/json"
"fmt"
)
// WeatherResult 天气查询结果
type WeatherResult struct {
City string `json:"city"`
Date string `json:"date"`
Weather string `json:"weather"`
Temperature string `json:"temperature"`
Error string `json:"error,omitempty"`
}
// 模拟天气数据
var weatherData = map[string]map[string][2]interface{}{
"北京": {"today": {"晴", 25}, "tomorrow": {"多云", 23}},
"上海": {"today": {"小雨", 28}, "tomorrow": {"阴", 27}},
"广州": {"today": {"雷阵雨", 31}, "tomorrow": {"晴", 33}},
}
func getWeather(city string, date string) WeatherResult {
if date == "" {
date = "today"
}
data, ok := weatherData[city]
if !ok {
return WeatherResult{Error: fmt.Sprintf("暂不支持查询 %s 的天气", city)}
}
dateKey := "today"
if date == "tomorrow" {
dateKey = "tomorrow"
}
entry := data[dateKey]
weather := entry[0].(string)
temp := entry[1].(int)
return WeatherResult{
City: city,
Date: date,
Weather: weather,
Temperature: fmt.Sprintf("%d°C", temp),
}
}
func main() {
result, _ := json.MarshalIndent(getWeather("北京", "tomorrow"), "", " ")
fmt.Println(string(result))
}
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
public class WeatherAgent {
// 模拟天气数据
private static final Map> weatherData = new HashMap<>();
static {
weatherData.put("北京", Map.of("today", new String[]{"晴", "25"}, "tomorrow", new String[]{"多云", "23"}));
weatherData.put("上海", Map.of("today", new String[]{"小雨", "28"}, "tomorrow", new String[]{"阴", "27"}));
weatherData.put("广州", Map.of("today", new String[]{"雷阵雨", "31"}, "tomorrow", new String[]{"晴", "33"}));
}
public static Map getWeather(String city, String date) {
if (date == null || date.isEmpty()) {
date = "today";
}
Map cityData = weatherData.get(city);
if (cityData == null) {
return Map.of("error", "暂不支持查询 " + city + " 的天气");
}
String dateKey = date.equals("tomorrow") ? "tomorrow" : "today";
String[] entry = cityData.get(dateKey);
return Map.of(
"city", city,
"date", date,
"weather", entry[0],
"temperature", entry[1] + "°C"
);
}
public static void main(String[] args) throws Exception {
Map result = getWeather("北京", "tomorrow");
System.out.println(new ObjectMapper().writeValueAsString(result));
// {"city":"北京","date":"tomorrow","weather":"多云","temperature":"23°C"}
}
}
3.4 定义工具 Schema：让 AI 知道工具怎么用
AI 怎么知道有这个工具？怎么知道参数格式？答案是工具 Schema——一份 JSON 描述：
tools_schema = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市指定日期的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如 北京、上海、广州"
},
"date": {
"type": "string",
"description": "日期，可以是 today、tomorrow 或 YYYY-MM-DD 格式",
"default": "today"
}
},
"required": ["city"]
}
}
}
]
Schema 的三个关键字段：
- name：工具名，AI 调用时用这个名字
- description：工具描述，AI 靠这段文字判断"该不该用这个工具"
- parameters：参数定义，AI 靠这个组装正确的参数 💡 description 是最重要的字段
AI 是否调用工具，很大程度上取决于 description 写得好不好。不要写"查询天气"，要写"查询指定城市指定日期的天气信息"——越具体越好。
3.5 组装 Agent：让 AI 用工具
现在把 AI 和工具连起来，形成完整的 Agent 循环：
import json
from openai import OpenAI
client = OpenAI() # 自动读取环境变量
def run_agent(user_message: str) -> str:
"""运行 Agent：感知→决策→行动→观察"""
messages = [
{"role": "system", "content": "你是一个天气助手，帮用户查询天气。用自然语言回答。"},
{"role": "user", "content": user_message},
]
# === Agent 循环 ===
while True:
# 1. 感知 + 决策：AI 决定是否调用工具
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools_schema,
)
msg = response.choices[0].message
# 2. 判断：AI 没调用工具 → 任务完成，返回结果
if not msg.tool_calls:
return msg.content
# 3. 行动：AI 要调工具 → 执行工具
messages.append(msg) # 先把 AI 的消息加入历史
for tool_call in msg.tool_calls:
# 解析工具名和参数
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"🔧 调用工具: {func_name}({func_args})")
# 执行工具
if func_name == "get_weather":
result = get_weather(**func_args)
else:
result = {"error": f"未知工具: {func_name}"}
# 4. 观察：把工具结果返回给 AI
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 循环回去 → AI 看到工具结果，决定下一步
# === 测试 ===
print(run_agent("北京明天天气怎么样？"))
import OpenAI from 'openai';
const client = new OpenAI(); // 自动读取环境变量
const toolsSchema = [/* ...同上定义工具 Schema... */];
async function getWeather(city: string, date: string = "today") {
// ...同上定义工具实现...
}
async function runAgent(userMessage: string): Promise {
const messages: any[] = [
{ role: "system", content: "你是一个天气助手，帮用户查询天气。用自然语言回答。" },
{ role: "user", content: userMessage },
];
// === Agent 循环 ===
while (true) {
// 1. 感知 + 决策：AI 决定是否调用工具
const response = await client.chat.completions.create({
model: "gpt-4o-mini",
messages,
tools: toolsSchema,
});
const msg = response.choices[0].message;
// 2. 判断：AI 没调用工具 → 任务完成
if (!msg.tool_calls || msg.tool_calls.length === 0) {
return msg.content || "";
}
// 3. 行动：AI 要调工具 → 执行工具
messages.push(msg);
for (const toolCall of msg.tool_calls) {
const funcName = toolCall.function.name;
const funcArgs = JSON.parse(toolCall.function.arguments);
console.log(`🔧 调用工具: ${funcName}(${JSON.stringify(funcArgs)})`);
// 执行工具
let result: any;
if (funcName === "get_weather") {
result = await getWeather(funcArgs.city, funcArgs.date);
} else {
result = { error: `未知工具: ${funcName}` };
}
// 4. 观察：把工具结果返回给 AI
messages.push({
role: "tool",
tool_call_id: toolCall.id,
content: JSON.stringify(result),
});
}
// 循环回去 → AI 看到工具结果，决定下一步
}
}
// === 测试 ===
runAgent("北京明天天气怎么样？").then(console.log);
package main
import (
"encoding/json"
"fmt"
"os"
openai "github.com/sashabaranov/go-openai"
)
func runAgent(userMessage string) (string, error) {
client := openai.NewClient(os.Getenv("OPENAI_API_KEY"))
messages := []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: "你是一个天气助手，帮用户查询天气。用自然语言回答。"},
{Role: openai.ChatMessageRoleUser, Content: userMessage},
}
// === Agent 循环 ===
for {
// 1. 感知 + 决策
resp, err := client.CreateChatCompletion(
context.Background(),
openai.ChatCompletionRequest{
Model: "gpt-4o-mini",
Messages: messages,
Tools: toolsSchema, // 工具 Schema 定义
},
)
if err != nil {
return "", err
}
msg := resp.Choices[0].Message
// 2. 判断：没调用工具 → 返回结果
if len(msg.ToolCalls) == 0 {
return msg.Content, nil
}
// 3. 行动：执行工具
messages = append(messages, msg)
for _, toolCall := range msg.ToolCalls {
funcName := toolCall.Function.Name
var funcArgs map[string]interface{}
json.Unmarshal([]byte(toolCall.Function.Arguments), &funcArgs)
fmt.Printf("🔧 调用工具: %s(%v)\n", funcName, funcArgs)
var result interface{}
if funcName == "get_weather" {
city, _ := funcArgs["city"].(string)
date, _ := funcArgs["date"].(string)
result = getWeather(city, date)
} else {
result = map[string]string{"error": "未知工具: " + funcName}
}
// 4. 观察：工具结果返回给 AI
resultJSON, _ := json.Marshal(result)
messages = append(messages, openai.ChatCompletionMessage{
Role: openai.ChatMessageRoleTool,
Content: string(resultJSON),
ToolCallID: toolCall.ID,
})
}
}
}
import com.openai.client.OpenAIClient;
import com.openai.models.*;
import java.util.*;
public class WeatherAgentApp {
static OpenAIClient client = OpenAIClient.fromEnv();
public static String runAgent(String userMessage) {
List messages = new ArrayList<>();
messages.add(ChatCompletionMessageParam.ofSystem(
ChatCompletionSystemMessageParam.builder()
.systemMessage("你是一个天气助手，帮用户查询天气。用自然语言回答。")
.build()));
messages.add(ChatCompletionMessageParam.ofUser(
ChatCompletionUserMessageParam.builder()
.userMessage(userMessage)
.build()));
// === Agent 循环 ===
while (true) {
// 1. 感知 + 决策
ChatCompletionCreateParams params = ChatCompletionCreateParams.builder()
.model("gpt-4o-mini")
.messages(messages)
.tools(toolSchemas) // 工具 Schema 定义
.build();
ChatCompletion response = client.chat().completions().create(params);
ChatCompletionMessage msg = response.choices().get(0).message();
String content = msg.content().orElse("");
// 2. 判断：没调用工具 → 返回结果
if (msg.toolCalls().isEmpty()) {
return content;
}
// 3. 行动：执行工具
messages.add(ChatCompletionMessageParam.ofAssistant(msg));
for (var toolCall : msg.toolCalls().get()) {
String funcName = toolCall.function().name();
String funcArgsJson = toolCall.function().arguments();
System.out.println("🔧 调用工具: " + funcName + "(" + funcArgsJson + ")");
// 解析参数并执行
Map funcArgs = parseArgs(funcArgsJson);
Map result;
if ("get_weather".equals(funcName)) {
result = getWeather(funcArgs.get("city"), funcArgs.get("date"));
} else {
result = Map.of("error", "未知工具: " + funcName);
}
// 4. 观察：工具结果返回给 AI
messages.add(ChatCompletionMessageParam.ofTool(
ChatCompletionToolMessageParam.builder()
.toolMessage(new ObjectMapper().writeValueAsString(result))
.toolCallId(toolCall.id())
.build()));
}
}
}
}
运行结果：
**🔧 调用工具: get_weather({'city': '北京', 'date': 'tomorrow'})**
北京明天多云，气温23°C。
3.6 理解 Agent 循环
回顾上面的代码，Agent 的核心是一个 while True 循环：
这就是 ReAct 循环的简化版——下一章会详细讲解。现在你只需要理解：
- 感知：AI 接收用户消息和对话历史
- 决策：AI 判断是否需要调用工具
- 行动：如果需要，执行工具调用
- 观察：把工具结果给 AI，让它看到结果
- 循环：AI 看到结果后，可能继续调工具，也可能直接回答
3.7 增强：多轮对话
上面的 Agent 只能处理单个问题。加上多轮对话：
def run_agent_loop():
"""多轮对话 Agent"""
messages = [
{"role": "system", "content": "你是一个天气助手，帮用户查询天气。用自然语言回答。"},
]
print("🌤️ 天气助手已启动，输入 'quit' 退出\n")
while True:
user_input = input("👤 ")
if user_input.lower() in ["quit", "exit", "退出"]:
print("👋 再见！")
break
messages.append({"role": "user", "content": user_input})
# Agent 循环（和之前一样）
while True:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools_schema,
)
msg = response.choices[0].message
if not msg.tool_calls:
print(f"🤖 {msg.content}\n")
messages.append(msg)
break
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"🔧 调用: {func_name}({func_args})")
if func_name == "get_weather":
result = get_weather(**func_args)
else:
result = {"error": f"未知工具"}
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 启动
run_agent_loop()
import OpenAI from 'openai';
import * as readline from 'readline';
const client = new OpenAI();
const toolsSchema = [/* ...工具 Schema 定义同上... */];
async function getWeather(city: string, date: string = "today") {
// ...工具实现同上...
}
async function runAgentLoop(): Promise {
const messages: any[] = [
{ role: "system", content: "你是一个天气助手，帮用户查询天气。用自然语言回答。" },
];
console.log("🌤️ 天气助手已启动，输入 'quit' 退出\n");
const rl = readline.createInterface({
input: process.stdin,
output: process.stdout,
});
const ask = (prompt: string): Promise =>
new Promise((resolve) => rl.question(prompt, resolve));
while (true) {
const userInput = await ask("👤 ");
if (["quit", "exit", "退出"].includes(userInput.toLowerCase())) {
console.log("👋 再见！");
rl.close();
break;
}
messages.push({ role: "user", content: userInput });
// Agent 循环（和之前一样）
while (true) {
const response = await client.chat.completions.create({
model: "gpt-4o-mini",
messages,
tools: toolsSchema,
});
const msg = response.choices[0].message;
if (!msg.tool_calls || msg.tool_calls.length === 0) {
console.log(`🤖 ${msg.content}\n`);
messages.push(msg);
break;
}
messages.push(msg);
for (const toolCall of msg.tool_calls) {
const funcName = toolCall.function.name;
const funcArgs = JSON.parse(toolCall.function.arguments);
console.log(`🔧 调用: ${funcName}(${JSON.stringify(funcArgs)})`);
let result: any;
if (funcName === "get_weather") {
result = await getWeather(funcArgs.city, funcArgs.date);
} else {
result = { error: "未知工具" };
}
messages.push({
role: "tool",
tool_call_id: toolCall.id,
content: JSON.stringify(result),
});
}
}
}
}
// 启动
runAgentLoop();
package main
import (
"bufio"
"encoding/json"
"fmt"
"os"
"strings"
openai "github.com/sashabaranov/go-openai"
)
var client = openai.NewClient(os.Getenv("OPENAI_API_KEY"))
func runAgentLoop() {
messages := []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: "你是一个天气助手，帮用户查询天气。用自然语言回答。"},
}
fmt.Println("🌤️ 天气助手已启动，输入 'quit' 退出")
fmt.Println()
scanner := bufio.NewScanner(os.Stdin)
for {
fmt.Print("👤 ")
if !scanner.Scan() {
break
}
userInput := strings.TrimSpace(scanner.Text())
lower := strings.ToLower(userInput)
if lower == "quit" || lower == "exit" || lower == "退出" {
fmt.Println("👋 再见！")
return
}
messages = append(messages, openai.ChatCompletionMessage{
Role: openai.ChatMessageRoleUser,
Content: userInput,
})
// Agent 循环（和之前一样）
for {
resp, err := client.CreateChatCompletion(
context.Background(),
openai.ChatCompletionRequest{
Model: "gpt-4o-mini",
Messages: messages,
Tools: toolsSchema,
},
)
if err != nil {
fmt.Printf("❌ API 错误: %v\n", err)
break
}
msg := resp.Choices[0].Message
if len(msg.ToolCalls) == 0 {
fmt.Printf("🤖 %s\n\n", msg.Content)
messages = append(messages, msg)
break
}
messages = append(messages, msg)
for _, toolCall := range msg.ToolCalls {
funcName := toolCall.Function.Name
var funcArgs map[string]interface{}
json.Unmarshal([]byte(toolCall.Function.Arguments), &funcArgs)
fmt.Printf("🔧 调用: %s(%v)\n", funcName, funcArgs)
var result interface{}
if funcName == "get_weather" {
city, _ := funcArgs["city"].(string)
date, _ := funcArgs["date"].(string)
result = getWeather(city, date)
} else {
result = map[string]string{"error": "未知工具"}
}
resultJSON, _ := json.Marshal(result)
messages = append(messages, openai.ChatCompletionMessage{
Role: openai.ChatMessageRoleTool,
Content: string(resultJSON),
ToolCallID: toolCall.ID,
})
}
}
}
}
func main() {
runAgentLoop()
}
import com.openai.client.OpenAIClient;
import com.openai.models.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
import java.util.Scanner;
public class WeatherAgentLoop {
static OpenAIClient client = OpenAIClient.fromEnv();
static ObjectMapper objectMapper = new ObjectMapper();
public static void runAgentLoop() {
List messages = new ArrayList<>();
messages.add(ChatCompletionMessageParam.ofSystem(
ChatCompletionSystemMessageParam.builder()
.systemMessage("你是一个天气助手，帮用户查询天气。用自然语言回答。")
.build()));
System.out.println("🌤️ 天气助手已启动，输入 'quit' 退出");
System.out.println();
Scanner scanner = new Scanner(System.in);
while (true) {
System.out.print("👤 ");
String userInput = scanner.nextLine().trim();
String lower = userInput.toLowerCase();
if (lower.equals("quit") || lower.equals("exit") || lower.equals("退出")) {
System.out.println("👋 再见！");
return;
}
messages.add(ChatCompletionMessageParam.ofUser(
ChatCompletionUserMessageParam.builder()
.userMessage(userInput)
.build()));
// Agent 循环（和之前一样）
while (true) {
ChatCompletionCreateParams params = ChatCompletionCreateParams.builder()
.model("gpt-4o-mini")
.messages(messages)
.tools(toolSchemas)
.build();
ChatCompletion response = client.chat().completions().create(params);
ChatCompletionMessage msg = response.choices().get(0).message();
if (msg.toolCalls().isEmpty()) {
System.out.println("🤖 " + msg.content().orElse("") + "\n");
messages.add(ChatCompletionMessageParam.ofAssistant(msg));
break;
}
messages.add(ChatCompletionMessageParam.ofAssistant(msg));
for (var toolCall : msg.toolCalls().get()) {
String funcName = toolCall.function().name();
String funcArgsJson = toolCall.function().arguments();
System.out.println("🔧 调用: " + funcName + "(" + funcArgsJson + ")");
Map funcArgs = parseArgs(funcArgsJson);
Map result;
if ("get_weather".equals(funcName)) {
result = getWeather(funcArgs.get("city"), funcArgs.get("date"));
} else {
result = Map.of("error", "未知工具");
}
try {
messages.add(ChatCompletionMessageParam.ofTool(
ChatCompletionToolMessageParam.builder()
.toolMessage(objectMapper.writeValueAsString(result))
.toolCallId(toolCall.id())
.build()));
} catch (Exception e) {
messages.add(ChatCompletionMessageParam.ofTool(
ChatCompletionToolMessageParam.builder()
.toolMessage("{\"error\":\"序列化失败\"}")
.toolCallId(toolCall.id())
.build()));
}
}
}
}
}
public static void main(String[] args) {
runAgentLoop();
}
}
测试效果：
🌤️ 天气助手已启动，输入 'quit' 退出
👤 北京明天天气怎么样？
**🔧 调用: get_weather({'city': '北京', 'date': 'tomorrow'})**
**🤖 北京明天多云，气温23°C。**
👤 那上海呢？
**🔧 调用: get_weather({'city': '上海', 'date': 'tomorrow'})**
**🤖 上海明天阴天，气温27°C。**
👤 这两个城市哪个更热？
**🤖 上海明天更热一些，气温27°C，而北京是23°C，差了4度。**
注意第三次对话——AI 没有调用工具，而是从对话历史中推理出了答案。这就是 Agent 的"记忆"——第6章会详细讲。
3.8 工具异常处理
现实世界并不总是顺利的——API 会超时、参数会出错、网络会断开。一个健壮的 Agent 必须能优雅地处理异常，而不是直接崩溃。
前���我们写的 Agent 有一个致命缺陷：假设工具永远不会出错。如果 get_weather 抛异常，整个 Agent 就挂了。
⚠️ 现实中的异常场景
- API 超时：天气服务响应慢或不可达
- 参数错误：AI 生成了不合法的参数（如 city="纽约" 不在支持列表）
- JSON 解析失败：AI 返回的 arguments 不是有效 JSON
- 服务限流：429 Too Many Requests
3.8.1 三种异常处理策略
面对异常，我们有三种策略：
- 重试：对于临时性故障（超时、限流），重试可能成功
- 优雅降级：对于不可恢复的故障，返回有意义的提示，而不是崩溃
- 让 AI 知道：把错误信息告诉 AI，让 AI 自己决定下一步（换工具？改参数？道歉？） 💡 关键原则：把错误信息告诉 AI
不要吞掉异常！把错误信息作为 tool result 返回给 AI，AI 会根据错误信息调整策略。比如：如果 city 参数不被支持，AI 可能换一个城市名再试；如果 API 超时，AI 可能直接用已有信息回答。
3.8.2 完整异常处理代码
import json
import time
import logging
from openai import OpenAI
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather_agent")
client = OpenAI()
def get_weather(city: str, date: str = "today") -> dict:
"""带模拟异常的天气查询"""
weather_data = {
"北京": {"today": ("晴", 25), "tomorrow": ("多云", 23)},
"上海": {"today": ("小雨", 28), "tomorrow": ("阴", 27)},
"广州": {"today": ("雷阵雨", 31), "tomorrow": ("晴", 33)},
}
# 模拟：不支持的城市 → 参数错误
if city not in weather_data:
return {
"error": f"不支持查询 {city} 的天气，目前支持：北京、上海、广州",
"supported_cities": list(weather_data.keys()),
}
# 模拟：随机超时（20%概率）
import random
if random.random() dict:
"""
执行工具，带指数退避重试
Args:
func_name: 工具名
func_args: 工具参数
max_retries: 最大重试次数
Returns:
工具结果字典（成功或错误信息）
"""
for attempt in range(max_retries + 1):
try:
if func_name == "get_weather":
result = get_weather(**func_args)
else:
result = {"error": f"未知工具: {func_name}"}
return result
except TimeoutError as e:
if attempt str:
"""带异常处理的 Agent"""
messages = [
{"role": "system", "content": "你是一个天气助手。如果工具返回错误信息，请根据错误提示调整策略或如实告诉用户。"},
{"role": "user", "content": user_message},
]
while True:
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools_schema,
)
except Exception as e:
# LLM API 本身出错
logger.error(f"❌ LLM API 异常: {e}")
return "抱歉，AI 服务暂时不可用，请稍后再试。"
msg = response.choices[0].message
if not msg.tool_calls:
return msg.content
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
try:
func_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
# AI 返回了无效 JSON → 直接告诉 AI
func_args = {}
logger.warning(f"⚠️ AI 返回无效参数")
print(f"🔧 调用工具: {func_name}({func_args})")
# 使用带重试的工具执行
result = execute_tool_with_retry(func_name, func_args)
print(f"📦 结果: {result}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 测试各种异常场景
print("=== 正常查询 ===")
print(run_agent_robust("北京今天天气怎么样？"))
print("\n=== 不支持的城市 ===")
print(run_agent_robust("纽约今天天气怎么样？"))
print("\n=== 可能触发超时重试 ===")
print(run_agent_robust("上海明天天气怎么样？"))
import OpenAI from 'openai';
const client = new OpenAI();
const logger = {
info: (msg: string) => console.log(`[INFO] ${msg}`),
warning: (msg: string) => console.warn(`[WARN] ${msg}`),
error: (msg: string) => console.error(`[ERROR] ${msg}`),
};
interface WeatherResult {
city?: string;
date?: string;
weather?: string;
temperature?: string;
error?: string;
supported_cities?: string[];
retry_attempts?: number;
raw_args?: string;
expected_params?: string;
}
const weatherData: Record> = {
"北京": { today: ["晴", 25], tomorrow: ["多云", 23] },
"上海": { today: ["小雨", 28], tomorrow: ["阴", 27] },
"广州": { today: ["雷阵雨", 31], tomorrow: ["晴", 33] },
};
function getWeather(city: string, date: string = "today"): WeatherResult {
// 模拟：不支持的城市 → 参数错误
if (!weatherData[city]) {
return {
error: `不支持查询 ${city} 的天气，目前支持：北京、上海、广州`,
supported_cities: Object.keys(weatherData),
};
}
// 模拟：随机超时（20%概率）
if (Math.random() {
for (let attempt = 0; attempt setTimeout(resolve, waitTime * 1000));
} else {
logger.error(`❌ 超过最大重试次数 (${maxRetries})`);
return {
error: `天气服务暂时不可用，已重试 ${maxRetries} 次。建议稍后再试。`,
retry_attempts: maxRetries,
};
}
} else if (e instanceof SyntaxError) {
// JSON 解析失败 → 不可重试
logger.error(`❌ 参数解析失败（不可重试）: ${errMsg}`);
return {
error: `参数格式错误，无法解析: ${errMsg}`,
raw_args: JSON.stringify(funcArgs),
};
} else if (e instanceof TypeError) {
// 参数类型错误 → 不可重试
logger.error(`❌ 参数类型错误（不可重试）: ${errMsg}`);
return {
error: `工具参数错误: ${errMsg}`,
expected_params: "city (string), date (string, optional)",
};
} else {
// 未知异常 → 降级
logger.error(`❌ 未知异常: ${e.constructor.name}: ${errMsg}`);
return {
error: "工具执行遇到未知错误，请稍后再试或换个方式提问。",
};
}
}
}
return { error: "未知错误" };
}
async function runAgentRobust(userMessage: string): Promise {
const messages: any[] = [
{ role: "system", content: "你是一个天气助手。如果工具返回错误信息，请根据错误提示调整策略或如实告诉用户。" },
{ role: "user", content: userMessage },
];
while (true) {
try {
const response = await client.chat.completions.create({
model: "gpt-4o-mini",
messages,
tools: toolsSchema,
});
const msg = response.choices[0].message;
if (!msg.tool_calls || msg.tool_calls.length === 0) {
return msg.content || "";
}
messages.push(msg);
for (const toolCall of msg.tool_calls) {
const funcName = toolCall.function.name;
let funcArgs: any;
try {
funcArgs = JSON.parse(toolCall.function.arguments);
} catch {
funcArgs = {};
logger.warning("⚠️ AI 返回无效参数");
}
console.log(`🔧 调用工具: ${funcName}(${JSON.stringify(funcArgs)})`);
const result = await executeToolWithRetry(funcName, funcArgs);
console.log(`📦 结果: ${JSON.stringify(result)}`);
messages.push({
role: "tool",
tool_call_id: toolCall.id,
content: JSON.stringify(result),
});
}
} catch (e: any) {
logger.error(`❌ LLM API 异常: ${e.message}`);
return "抱歉，AI 服务暂时不可用，请稍后再试。";
}
}
}
// 测试各种异常场景
console.log("=== 正常查询 ===");
runAgentRobust("北京今天天气怎么样？").then(console.log);
console.log("\n=== 不支持的城市 ===");
runAgentRobust("纽约今天天气怎么样？").then(console.log);
console.log("\n=== 可能触发超时重试 ===");
runAgentRobust("上海明天天气怎么样？").then(console.log);
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"math"
"math/rand"
"os"
"time"
openai "github.com/sashabaranov/go-openai"
)
var client = openai.NewClient(os.Getenv("OPENAI_API_KEY"))
var logger = log.New(os.Stderr, "", log.LstdFlags)
func getWeatherRobust(city string, date string) map[string]interface{} {
weatherData := map[string]map[string][2]interface{}{
"北京": {"today": {"晴", 25}, "tomorrow": {"多云", 23}},
"上海": {"today": {"小雨", 28}, "tomorrow": {"阴", 27}},
"广州": {"today": {"雷阵雨", 31}, "tomorrow": {"晴", 33}},
}
if date == "" {
date = "today"
}
data, ok := weatherData[city]
if !ok {
return map[string]interface{}{
"error": fmt.Sprintf("不支持查询 %s 的天气，目前支持：北京、上海、广州", city),
"supported_cities": []string{"北京", "上海", "广州"},
}
}
// 模拟：随机超时（20%概率）
if rand.Float64() dict:
// Python: weather_data = {
// Python: "北京": {"today": ("晴", 25), "tomorrow": ("多云", 23)},
// Python: "上海": {"today": ("小雨", 28), "tomorrow": ("阴", 27)},
// Python: "广州": {"today": ("雷阵雨", 31), "tomorrow": ("晴", 33)},
// Python: }
// 模拟：不支持的城市 → 参数错误
if (city !in weather_data) {
return {;
// Python: "error": f"不支持查询 {city} 的天气，目前支持：北京、上海、广州",
// Python: "supported_cities": list(weather_data.keys()),
// Python: }
// 模拟：随机超时（20%概率）
// import random
if (random.random() dict:
// Python: 执行工具，带指数退避重试
// Python: Args:
// Python: func_name: 工具名
// Python: func_args: 工具参数
// Python: max_retries: 最大重试次数
// Python: Returns:
// Python: 工具结果字典（成功或错误信息）
for (var attempt : range(max_retries + 1)) {
// Python: try:
if (func_name == "get_weather") {
// Python: result = get_weather(**func_args)
} else {
// Python: result = {"error": f"未知工具: {func_name}"}
return result;
// Python: except TimeoutError as e:
if (attempt str:
// Python: messages = [
// Python: {"role": "system", "content": "你是一个天气助手。如果工具返回错误信息，请根据错误提示调整策略或如实告诉用户。"},
// Python: {"role": "user", "content": user_message},
// Python: ]
// Python: while True:
// Python: try:
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=tools_schema,
// Python: )
// Python: except Exception as e:
// LLM API 本身出错
// Python: logger.error(f"❌ LLM API 异常: {e}")
return "抱歉，AI 服务暂时不可用，请稍后再试。";
// Python: msg = response.choices[0].message
if (!msg.tool_calls) {
return msg.content;
// Python: messages.append(msg)
for (var tool_call : msg.tool_calls) {
// Python: func_name = tool_call.function.name
// Python: try:
// Python: func_args = json.loads(tool_call.function.arguments)
// Python: except json.JSONDecodeError:
// AI 返回了无效 JSON → 直接告诉 AI
// Python: func_args = {}
// Python: logger.warning(f"⚠️ AI 返回无效参数")
System.out.println(String.format("$1"));
// 使用带重试的工具执行
// Python: result = execute_tool_with_retry(func_name, func_args)
System.out.println(String.format("$1"));
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试各种异常场景
System.out.println("=== 正常查询 ===");
System.out.println(run_agent_robust("北京今天天气怎么样？"));
System.out.println("\n=== 不支持的城市 ===");
System.out.println(run_agent_robust("纽约今天天气怎么样？"));
System.out.println("\n=== 可能触发超时重试 ===");
System.out.println(run_agent_robust("上海明天天气怎么样？"));
}
运行效果：
=== 正常查询 ===
**🔧 调用工具: get_weather({'city': '北京', 'date': 'today'})**
**📦 结果: {'city': '北京', 'date': 'today', 'weather': '晴', 'temperature': '25°C'}**
北京今天晴天，气温25°C。
=== 不支持的城市 ===
**🔧 调用工具: get_weather({'city': '纽约', 'date': 'today'})**
**📦 结果: {'error': '不支持查询 纽约 的天气，目前支持：北京、上海、广州', 'supported_cities': ['北京', '上海', '广州']}**
抱歉，目前不支持查询纽约的天气。我可以查询北京、上海和广州的天气，您想查哪个城市？
=== 可能触发超时重试 ===
**🔧 调用工具: get_weather({'city': '上海', 'date': 'tomorrow'})**
**⚠️ 第 1 次重试，等待 1s: 天气服务响应超时**
**⚠️ 第 2 次重试，等待 2s: 天气服务响应超时**
**📦 结果: {'city': '上海', 'date': 'tomorrow', 'weather': '阴', 'temperature': '27°C'}**
上海明天阴天，气温27°C。
注意 AI 在面对不同错误时的自适应行为：
- 不支持的城市：AI 看到了 supported_cities 列表，主动推荐可用城市
- 超时重试成功：Agent 重试后成功获取数据，正常回答
- 超时重试耗尽：Agent 告知用户服务不可用，建议稍后重试 💡 指数退避的数学原理
指数退避（Exponential Backoff）的等待时间是 2^attempt：第1次等1秒，第2次等2秒，第3次等4秒……这比固定间隔重试更友好——给服务恢复的时间，避免雪崩。生产环境通常还会加入随机 jitter（抖动），避免所有客户端同时重试。
3.8.3 重试 vs 降级的决策表
哪些错误应该重试？哪些应该直接降级？这是工程决策，不是拍脑袋： | 异常类型 | 能否重试 | 策略 | 示例 | | --- | --- | --- | --- | | TimeoutError | ✅ 可以 | 指数退避重试 | API 超时、网络抖动 | | 429 Rate Limit | ✅ 可以 | 指数退避 + jitter | 请求过多被限流 | | 503 Service Unavailable | ✅ 可以 | 指数退避重试 | 服务临时不可用 | | TypeError（参数错误） | ❌ 不行 | 告诉 AI 调整参数 | 传了 int 给 string 参数 | | JSONDecodeError | ❌ 不行 | 告诉 AI 参数无效 | AI 生成了非法 JSON | | KeyError（数据不存在） | ❌ 不行 | 降级返回提示 | 查询不存在的城市 | | PermissionError | ❌ 不行 | 降级返回提示 | 没有权限访问某资源 | 核心判断原则：可重试 = 临时性故障（可能自己恢复），不可重试 = 逻辑性错误（再试也一样）。
2.7 节的多轮对话让 Agent 能和用户持续交互，2.8 节的异常处理让 Agent 在面对现实世界的不确定性时不崩溃。两者结合，Agent 才从"玩具"变成"产品"。接下来我们让 Agent 从单工具进化到多工具协作。
3.9 多工具协作
真正的 Agent 不只有一个工具。就像人不止会查天气——你还会查天气、看地图、规划行程。Agent 也应该能协调多个工具完成复杂任务。
这一节我们给天气助手加一个新伙伴：行程规划工具。然后看看 Agent 如何让两个工具协作。
3.9.1 定义行程规划工具
def plan_trip(city: str, weather_condition: str = "", days: int = 1) -> dict:
"""
根据城市和天气情况规划出行建议
Args:
city: 目的地城市
weather_condition: 天气情况（晴/多云/小雨/雷阵雨等）
days: 出行天数
Returns:
行程建议字典
"""
trip_data = {
"北京": {
"晴": "适合去故宫、颐和园，推荐户外游览",
"多云": "适合逛博物馆、胡同，室内外均可",
"小雨": "建议参观国家博物馆、798艺术区，以室内为主",
"雷阵雨": "强烈建议室内活动：故宫室内展区、国家大剧院",
},
"上海": {
"晴": "外滩散步、豫园游览、迪士尼乐园",
"多云": "南京路逛街、田子坊艺术区",
"小雨": "上海博物馆、环球金融中心观光厅",
"雷阵雨": "室内商场、上海大剧院",
},
"广州": {
"晴": "白云山登山、珠江夜游",
"多云": "陈家祠、沙面岛散步",
"小雨": "广州图书馆、广东省博物馆",
"雷阵雨": "天河城购物中心、室内美食探店",
},
}
if city not in trip_data:
return {
"error": f"暂不支持 {city} 的行程规划",
"supported_cities": list(trip_data.keys()),
}
if not weather_condition:
return {
"city": city,
"note": "请先查询天气，我再给出更精准的行程建议",
"general_tip": trip_data[city].get("晴", "建议户外游览"),
}
recommendation = trip_data[city].get(
weather_condition,
"建议根据天气灵活安排室内外活动"
)
return {
"city": city,
"weather_condition": weather_condition,
"recommendation": recommendation,
"days": days,
}
# 测试
print(plan_trip("北京", "雷阵雨", 2))
# {'city': '北京', 'weather_condition': '雷阵雨', 'recommendation': '强烈建议室内活动：故宫室内展区、国家大剧院', 'days': 2}
// def plan_trip(city: str, weather_condition: str = "", days: int = 1) -> dict:
/** docstring */
// 根据城市和天气情况规划出行建议
// Args:
// city: 目的地城市
// weather_condition: 天气情况（晴/多云/小雨/雷阵雨等）
// days: 出行天数
// Returns:
// 行程建议字典
/** docstring */
const trip_data = {;
// "北京": {
// "晴": "适合去故宫、颐和园，推荐户外游览",
// "多云": "适合逛博物馆、胡同，室内外均可",
// "小雨": "建议参观国家博物馆、798艺术区，以室内为主",
// "雷阵雨": "强烈建议室内活动：故宫室内展区、国家大剧院",
// },
// "上海": {
// "晴": "外滩散步、豫园游览、迪士尼乐园",
// "多云": "南京路逛街、田子坊艺术区",
// "小雨": "上海博物馆、环球金融中心观光厅",
// "雷阵雨": "室内商场、上海大剧院",
// },
// "广州": {
// "晴": "白云山登山、珠江夜游",
// "多云": "陈家祠、沙面岛散步",
// "小雨": "广州图书馆、广东省博物馆",
// "雷阵雨": "天河城购物中心、室内美食探店",
// },
// }
if (city !in trip_data) {
return {;
// "error": f"暂不支持 {city} 的行程规划",
// "supported_cities": list(trip_data.keys()),
// }
if (!weather_condition) {
return {;
// "city": city,
// "note": "请先查询天气，我再给出更精准的行程建议",
// "general_tip": trip_data[city].get("晴", "建议户外游览"),
// }
const recommendation = trip_data[city].get(;
// weather_condition,
// "建议根据天气灵活安排室内外活动"
// )
return {;
// "city": city,
// "weather_condition": weather_condition,
// "recommendation": recommendation,
// "days": days,
// }
// 测试
console.log(plan_trip("北京", "雷阵雨", 2));
// {'city': '北京', 'weather_condition': '雷阵雨', 'recommendation': '强烈建议室内活动：故宫室内展区、国家大剧院', 'days': 2}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: def plan_trip(city: str, weather_condition: str = "", days: int = 1) -> dict:
// Python: 根据城市和天气情况规划出行建议
// Python: Args:
// Python: city: 目的地城市
// Python: weather_condition: 天气情况（晴/多云/小雨/���阵雨等）
// Python: days: 出行天数
// Python: Returns:
// Python: 行程建议字典
// Python: trip_data = {
// Python: "北京": {
// Python: "晴": "适合去故宫、颐和园，推荐户外游览",
// Python: "多云": "适合逛博物馆、胡同，室内外均可",
// Python: "小雨": "建议参观国家博物馆、798艺术区，以室内为主",
// Python: "雷阵雨": "强烈建议室内活动：故宫室内展区、国家大剧院",
// Python: },
// Python: "上海": {
// Python: "晴": "外滩散步、豫园游览、迪士尼乐园",
// Python: "多云": "南京路逛街、田子坊艺术区",
// Python: "小雨": "上海博物馆、环球金融中心观光厅",
// Python: "雷阵雨": "室内商场、上海大剧院",
// Python: },
// Python: "广州": {
// Python: "晴": "白云山登山、珠江夜游",
// Python: "多云": "陈家祠、沙面岛散步",
// Python: "小雨": "广州图书馆、广东省博物馆",
// Python: "雷阵雨": "天河城购物中心、室内美食探店",
// Python: },
// Python: }
if city not in trip_data {
return {
// Python: "error": f"暂不支持 {city} 的行程规划",
// Python: "supported_cities": list(trip_data.keys()),
// Python: }
if not weather_condition {
return {
// Python: "city": city,
// Python: "note": "请先查询天气，我再给出更精准的行程建议",
// Python: "general_tip": trip_data[city].get("晴", "建议户外游览"),
// Python: }
// Python: recommendation = trip_data[city].get(
// Python: weather_condition,
// Python: "建议根据天气灵活安排室内外活动"
// Python: )
return {
// Python: "city": city,
// Python: "weather_condition": weather_condition,
// Python: "recommendation": recommendation,
// Python: "days": days,
// Python: }
// 测试
fmt.Println(plan_trip("北京", "雷阵雨", 2))
// {'city': '北京', 'weather_condition': '雷阵雨', 'recommendation': '强烈建议室内活动：故宫室内展区、国家大剧院', 'days': 2}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: def plan_trip(city: str, weather_condition: str = "", days: int = 1) -> dict:
// Python: 根据城市和天气情况规划出行建议
// Python: Args:
// Python: city: 目的地城市
// Python: weather_condition: 天气情况（晴/多云/小雨/雷阵雨等）
// Python: days: 出行天数
// Python: Returns:
// Python: 行程建议字典
// Python: trip_data = {
// Python: "北京": {
// Python: "晴": "适合去故宫、颐和园，推荐户外游览",
// Python: "多云": "适合逛博物馆、胡同，室内外均可",
// Python: "小雨": "建议参观国家博物馆、798艺术区，以室内为主",
// Python: "雷阵雨": "强烈建议室内活动：故宫室内展区、国家大剧院",
// Python: },
// Python: "上海": {
// Python: "晴": "外滩散步、豫园游览、迪士尼乐园",
// Python: "多云": "南京路逛街、田子坊艺术区",
// Python: "小雨": "上海博物馆、环球金融中心观光厅",
// Python: "雷阵雨": "室内商场、上海大剧院",
// Python: },
// Python: "广州": {
// Python: "晴": "白云山登山、珠江夜游",
// Python: "多云": "陈家祠、沙面岛散步",
// Python: "小雨": "广州图书馆、广东省博物馆",
// Python: "雷阵雨": "天河城购物中心、室内美食探店",
// Python: },
// Python: }
if (city !in trip_data) {
return {;
// Python: "error": f"暂不支持 {city} 的行程规划",
// Python: "supported_cities": list(trip_data.keys()),
// Python: }
if (!weather_condition) {
return {;
// Python: "city": city,
// Python: "note": "请先查询天气，我再给出更精准的行程建议",
// Python: "general_tip": trip_data[city].get("晴", "建议户外游览"),
// Python: }
// Python: recommendation = trip_data[city].get(
// Python: weather_condition,
// Python: "建议根据天气灵活安排室内外活动"
// Python: )
return {;
// Python: "city": city,
// Python: "weather_condition": weather_condition,
// Python: "recommendation": recommendation,
// Python: "days": days,
// Python: }
// 测试
System.out.println(plan_trip("北京", "雷阵雨", 2));
// {'city': '北京', 'weather_condition': '雷阵雨', 'recommendation': '强烈建议室内活动：故宫室内展区、国家大剧院', 'days': 2}
}
3.9.2 扩展工具 Schema
现在有两个工具，Schema 也需要扩展：
tools_schema = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市指定日期的天气信息，包括天气状况和温度",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如 北京、上海、广州"
},
"date": {
"type": "string",
"description": "日期，可以是 today、tomorrow 或 YYYY-MM-DD",
"default": "today"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "plan_trip",
"description": "根据城市和天气情况规划出行建议，需要天气信息作为输入",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "目的地城市，如 北京、上海、广州"
},
"weather_condition": {
"type": "string",
"description": "天气状况，如 晴、多云、小雨、雷阵雨"
},
"days": {
"type": "integer",
"description": "出行天数，默认1天",
"default": 1
}
},
"required": ["city"]
}
}
}
]
3.9.3 多工具 Agent 循环
Agent 循环的结构不变——AI 自己决定先查天气还是先规划行程：
def run_multi_tool_agent(user_message: str) -> str:
"""多工具协作 Agent"""
messages = [
{
"role": "system",
"content": "你是一个出行助手，可以查询天气和规划行程。\n"
"如果用户问出行建议，先查天气再规划行程。\n"
"如果工具返回错误，根据错误信息调整策略。",
},
{"role": "user", "content": user_message},
]
tool_map = {
"get_weather": get_weather,
"plan_trip": plan_trip,
}
while True:
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools_schema,
)
except Exception as e:
return f"AI 服务暂时不可用: {e}"
msg = response.choices[0].message
if not msg.tool_calls:
return msg.content
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
try:
func_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
func_args = {}
print(f"🔧 调用: {func_name}({func_args})")
# 使用带重试的执行（复用 2.8 节的代码）
result = execute_tool_with_retry(func_name, func_args)
print(f"📦 结果: {result}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 测试多工具协作
print("=== 简单天气查询 ===")
print(run_multi_tool_agent("北京明天天气怎么样？"))
print("\n=== 出行建议（触发双工具协作） ===")
print(run_multi_tool_agent("北京明天适不适合出游？"))
print("\n=== 直接问行程（Agent 自行决定先查天气） ===")
print(run_multi_tool_agent("帮我规划一下上海的行程"))
// def run_multi_tool_agent(user_message: str) -> str:
/** docstring */
const messages = [;
// {
// "role": "system",
// "content": "你是一个出行助手，可以查询天气和规划行程。\n"
// "如果用户问出行建议，先查天气再规划行程。\n"
// "如果工具返回错误，根据错误信息调整策略。",
// },
// {"role": "user", "content": user_message},
// ]
const tool_map = {;
// "get_weather": get_weather,
// "plan_trip": plan_trip,
// }
while (true) {
try {
const response = client.chat.completions.create(;
const model = "gpt-4o-mini",;
const messages = messages,;
const tools = tools_schema,;
// )
} catch (Exception) {
return `AI 服务暂时不可用: ${$1}`;
const msg = response.choices[0].message;
if (!msg.tool_calls) {
return msg.content;
// messages.append(msg)
for (const tool_call of msg.tool_calls) {
const func_name = tool_call.function.name;
try {
const func_args = json.loads(tool_call.function.arguments);
// except json.JSONDecodeError:
const func_args = {};
console.log(`🔧 调用: ${$1}(${$1})`);
// 使用带重试的执行（复用 2.8 节的代码）
const result = execute_tool_with_retry(func_name, func_args);
console.log(`📦 结果: ${$1}`);
// messages.append({
// "role": "tool",
// "tool_call_id": tool_call.id,
// "content": json.dumps(result, ensure_ascii=False),
// })
// 测试多工具协作
console.log("=== 简单天气查询 ===");
console.log(run_multi_tool_agent("北京明天天气怎么样？"));
console.log("\n=== 出行建议（触发双工具协作） ===");
console.log(run_multi_tool_agent("北京明天适不适合出游？"));
console.log("\n=== 直接问行程（Agent 自行决定先查天气） ===");
console.log(run_multi_tool_agent("帮我规划一下上海的行程"));
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: def run_multi_tool_agent(user_message: str) -> str:
// Python: messages = [
// Python: {
// Python: "role": "system",
// Python: "content": "你是一个出行助手，可以查询天气和规划行程。\n"
// Python: "如果用户问出行建议，先查天气再规划行程。\n"
// Python: "如果工具返回错误，根据错误信息调整策略。",
// Python: },
// Python: {"role": "user", "content": user_message},
// Python: ]
// Python: tool_map = {
// Python: "get_weather": get_weather,
// Python: "plan_trip": plan_trip,
// Python: }
// Python: while True:
// try block
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=tools_schema,
// Python: )
// except block
return f"AI 服务暂时不可用: {e}"
// Python: msg = response.choices[0].message
if not msg.tool_calls {
return msg.content
// Python: messages.append(msg)
for _, tool_call := range msg.tool_calls {
// Python: func_name = tool_call.function.name
// try block
// Python: func_args = json.loads(tool_call.function.arguments)
// except block
// Python: func_args = {}
fmt.Println(f"🔧 调用: {func_name}({func_args})")
// 使用带重试的执行（复用 2.8 节的代码）
// Python: result = execute_tool_with_retry(func_name, func_args)
fmt.Println(f"📦 结果: {result}")
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试多工具协作
fmt.Println("=== 简单天气查询 ===")
fmt.Println(run_multi_tool_agent("北京明天天气怎么样？"))
fmt.Println("\n=== 出行建议（触发双工具协作） ===")
fmt.Println(run_multi_tool_agent("北京明天适不适合出游？"))
fmt.Println("\n=== 直接问行程（Agent 自行决定先查天气） ===")
fmt.Println(run_multi_tool_agent("帮我规划一下上海的行程"))
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: def run_multi_tool_agent(user_message: str) -> str:
// Python: messages = [
// Python: {
// Python: "role": "system",
// Python: "content": "你是一个出行助手，可以查询天气和规划行程。\n"
// Python: "如果用户问出行建议，先查天气再规划行程。\n"
// Python: "如果工具返回错误，根据错误信息调整策略。",
// Python: },
// Python: {"role": "user", "content": user_message},
// Python: ]
// Python: tool_map = {
// Python: "get_weather": get_weather,
// Python: "plan_trip": plan_trip,
// Python: }
// Python: while True:
// Python: try:
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=tools_schema,
// Python: )
// Python: except Exception as e:
return f"AI 服务暂时不可用: {e}";
// Python: msg = response.choices[0].message
if (!msg.tool_calls) {
return msg.content;
// Python: messages.append(msg)
for (var tool_call : msg.tool_calls) {
// Python: func_name = tool_call.function.name
// Python: try:
// Python: func_args = json.loads(tool_call.function.arguments)
// Python: except json.JSONDecodeError:
// Python: func_args = {}
System.out.println(String.format("$1"));
// 使用带重试的执行（复用 2.8 节的代码）
// Python: result = execute_tool_with_retry(func_name, func_args)
System.out.println(String.format("$1"));
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试多工具协作
System.out.println("=== 简单天气查询 ===");
System.out.println(run_multi_tool_agent("北京明天天气怎么样？"));
System.out.println("\n=== 出行建议（触发双工具协作） ===");
System.out.println(run_multi_tool_agent("北京明天适不适合出游？"));
System.out.println("\n=== 直接问行程（Agent 自行决定先查天气） ===");
System.out.println(run_multi_tool_agent("帮我规划一下上海的行程"));
}
运行效果：
=== 简单天气查询 ===
**🔧 调用: get_weather({'city': '北京', 'date': 'tomorrow'})**
**📦 结果: {'city': '北京', 'date': 'tomorrow', 'weather': '多云', 'temperature': '23°C'}**
北京明天多云，气温23°C。
=== 出行建议（触发双工具协作） ===
**🔧 调用: get_weather({'city': '北京', 'date': 'tomorrow'})**
**📦 结果: {'city': '北京', 'date': 'tomorrow', 'weather': '多云', 'temperature': '23°C'}**
**🔧 调用: plan_trip({'city': '北京', 'weather_condition': '多云', 'days': 1})**
**📦 结果: {'city': '北京', 'weather_condition': '多云', 'recommendation': '适合逛博物馆、胡同，室内外均可', 'days': 1}**
北京明天多云23°C，适合逛博物馆和胡同，室内外活动都可以安排。
=== 直接问行程（Agent 自行决定先查天气） ===
**🔧 调用: get_weather({'city': '上海', 'date': 'today'})**
**📦 结果: {'city': '上海', 'date': 'today', 'weather': '小雨', 'temperature': '28°C'}**
**🔧 调用: plan_trip({'city': '上海', 'weather_condition': '小雨', 'days': 1})**
**📦 结果: {'city': '上海', 'weather_condition': '小雨', 'recommendation': '上海博物馆、环球金融中心观光厅', 'days': 1}**
上海今天小雨28°C，建议以室内为主：上海博物馆和环球金融中心观光厅。
💡 AI 自己决定了调用顺序！
我们没有写"先查天气再规划行程"的逻辑——AI 自己推理出了这个顺序。这就是 Agent 的核心优势：你不需要硬编码流程，AI 会根据目标和工具描述自行编排。但这也有代价——AI 有可能编错顺序，这正是 2.10 节要分析的问题。
3.9.4 多工具 vs 单工具的成本对比
多工具协作的代价是什么？ | 维度 | 单工具（查天气） | 双工具（天气+行程） | | --- | --- | --- | | 工具 Schema tokens | ~150 | ~350 | | 典型循环次数 | 1 次 | 2-3 次 | | 总 token 消耗 | ~500 | ~1500 | | 延迟 | 1-2s | 3-6s | | 出错概率 | 低 | 中（AI 可能调错顺序） | 工具越多，AI 的决策空间越大，出错概率越高，延迟也越长。这不是免费的——2.11 节会讨论 Agent 的边界，什么时候不该用多工具 Agent。
2.9 节展示了 Agent 从单工具到多工具的进化。但一个关键问题还没回答：**Agent 循环到底消耗了多少资源？**我们需要量化分析，才能做出理性决策。这正是下一节的主题。
3.10 工具调用链分析
Agent 不是魔法——每次循环都在消耗 token、时间和钱。一个优秀的 Agent 工程师必须能量化分析每次调用，才能优化成本和延迟。
这一节我们给 Agent 加上调用链追踪——打印每次循环的 token 消耗、调用次数、延迟，让你看清 Agent 的真实运行过程。
3.10.1 什么是调用链？
一次 Agent 请求可能触发多次循环，每次循环包含：
- LLM 调用：消耗 token，产生延迟
- 工具执行：消耗时间，可能有延迟
- 消息累积：历史越来越长，token 消耗递增
3.10.2 带追踪的 Agent
import time
import json
from openai import OpenAI
client = OpenAI()
class AgentTracer:
"""Agent 调用链追踪器"""
def __init__(self):
self.loops = [] # 每次循环的记录
self.total_tokens = 0
self.total_time = 0.0
self.tool_calls_count = 0
def record_loop(self, loop_num: int, response, loop_time: float, tool_names: list = None):
"""记录一次循环"""
usage = response.usage
loop_info = {
"loop": loop_num,
"prompt_tokens": usage.prompt_tokens if usage else 0,
"completion_tokens": usage.completion_tokens if usage else 0,
"total_tokens": usage.total_tokens if usage else 0,
"time_seconds": round(loop_time, 2),
"tool_calls": tool_names or [],
"has_tool_call": len(tool_names or []) > 0,
}
self.loops.append(loop_info)
self.total_tokens += loop_info["total_tokens"]
self.total_time += loop_time
if tool_names:
self.tool_calls_count += len(tool_names)
def print_report(self):
"""打印分析报告"""
print("\n" + "="*50)
print("📊 Agent 调用链分析报告")
print("="*50)
print(f"总循环次数: {len(self.loops)}")
print(f"总工具调用: {self.tool_calls_count} 次")
print(f"总 token 消耗: {self.total_tokens}")
print(f"总耗时: {round(self.total_time, 2)}s")
print(f"\n--- 每次循环详情 ---")
for loop in self.loops:
role = "🔧 工具调用" if loop["has_tool_call"] else "💬 最终回答"
print(f" 循环 {loop['loop']}: {role}")
print(f" tokens: {loop['total_tokens']} (prompt: {loop['prompt_tokens']}, completion: {loop['completion_tokens']})")
print(f" 延迟: {loop['time_seconds']}s")
if loop['tool_calls']:
print(f" 工具: {', '.join(loop['tool_calls'])}")
print("="*50)
# 成本估算（以 GPT-4o-mini 为例）
cost_per_1k_input = 0.15 / 1000 # $0.15/1M tokens → $0.00015/1K
cost_per_1k_output = 0.60 / 1000 # $0.60/1M tokens → $0.0006/1K
input_tokens = sum(l["prompt_tokens"] for l in self.loops)
output_tokens = sum(l["completion_tokens"] for l in self.loops)
estimated_cost = input_tokens * cost_per_1k_input + output_tokens * cost_per_1k_output
print(f"💰 预估成本: ${estimated_cost:.6f}")
def run_agent_with_tracer(user_message: str) -> str:
"""带调用链追踪的 Agent"""
tracer = AgentTracer()
messages = [
{
"role": "system",
"content": "你是一个出行助手，可以查询天气和规划行程。"
},
{"role": "user", "content": user_message},
]
tool_map = {
"get_weather": get_weather,
"plan_trip": plan_trip,
}
loop_num = 0
while True:
loop_num += 1
start_time = time.time()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools_schema,
)
loop_time = time.time() - start_time
msg = response.choices[0].message
# 记录工具调用名
tool_names = [tc.function.name for tc in (msg.tool_calls or [])]
tracer.record_loop(loop_num, response, loop_time, tool_names)
if not msg.tool_calls:
tracer.print_report()
return msg.content
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"🔧 调用: {func_name}({func_args})")
result = execute_tool_with_retry(func_name, func_args)
print(f"📦 结果: {result}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 测试
print(run_agent_with_tracer("北京明天适不适合出游？"))
// Node.js built-in or npm package for: time
// Node.js built-in or npm package for: json
import OpenAI from 'openai';
const client = OpenAI();
class AgentTracer {
/** docstring */
constructor() {
// self.loops = [] # 每次循环的记录
// self.total_tokens = 0
// self.total_time = 0.0
// self.tool_calls_count = 0
record_loop(loop_num: number, response, loop_time: number, tool_names: list = null) {
/** docstring */
// usage = response.usage
// loop_info = {
// "loop": loop_num,
// "prompt_tokens": usage.prompt_tokens if usage else 0,
// "completion_tokens": usage.completion_tokens if usage else 0,
// "total_tokens": usage.total_tokens if usage else 0,
// "time_seconds": round(loop_time, 2),
// "tool_calls": tool_names or [],
// "has_tool_call": len(tool_names or []) > 0,
// }
// self.loops.append(loop_info)
// self.total_tokens += loop_info["total_tokens"]
// self.total_time += loop_time
if (tool_names) {
// self.tool_calls_count += len(tool_names)
print_report() {
/** docstring */
console.log("\n" + "="*50);
console.log("📊 Agent 调用链分析报告");
console.log("="*50);
console.log(`总循环次数: {len(self.loops)}`);
console.log(`总工具调用: {self.tool_calls_count} 次`);
console.log(`总 token 消耗: {self.total_tokens}`);
console.log(`总耗时: {round(self.total_time, 2)}s`);
console.log(`\n--- 每次循环详情 ---`);
for (const loop of self.loops) {
// role = "🔧 工具调用" if loop["has_tool_call"] else "💬 最终回答"
console.log(` 循环 {loop['loop']}: ${$1}`);
console.log(` tokens: {loop['total_tokens']} (prompt: {loop['prompt_tokens']}, completion: {loop['completion_tokens']})`);
console.log(` 延迟: {loop['time_seconds']}s`);
if (loop['tool_calls']) {
console.log(` 工具: {', '.join(loop['tool_calls'])}`);
console.log("="*50);
// 成本估算（以 GPT-4o-mini 为例）
// cost_per_1k_input = 0.15 / 1000 # $0.15/1M tokens → $0.00015/1K
// cost_per_1k_output = 0.60 / 1000 # $0.60/1M tokens → $0.0006/1K
// input_tokens = sum(l["prompt_tokens"] for l in self.loops)
// output_tokens = sum(l["completion_tokens"] for l in self.loops)
// estimated_cost = input_tokens * cost_per_1k_input + output_tokens * cost_per_1k_output
console.log(`💰 预估成本: ${estimated_cost:.6f}`);
// def run_agent_with_tracer(user_message: str) -> str:
/** docstring */
// tracer = AgentTracer()
// messages = [
// {
// "role": "system",
// "content": "你是一个出行助手，可以查询天气和规划行程。"
// },
// {"role": "user", "content": user_message},
// ]
// tool_map = {
// "get_weather": get_weather,
// "plan_trip": plan_trip,
// }
// loop_num = 0
while (true) {
// loop_num += 1
// start_time = time.time()
// response = client.chat.completions.create(
// model = "gpt-4o-mini",
// messages = messages,
// tools = tools_schema,
// )
// loop_time = time.time() - start_time
// msg = response.choices[0].message
// 记录工具调用名
// tool_names = [tc.function.name for tc in (msg.tool_calls or [])]
// tracer.record_loop(loop_num, response, loop_time, tool_names)
if (!msg.tool_calls) {
// tracer.print_report()
return msg.content;
// messages.append(msg)
for (const tool_call of msg.tool_calls) {
// func_name = tool_call.function.name
// func_args = json.loads(tool_call.function.arguments)
console.log(`🔧 调用: ${$1}(${$1})`);
// result = execute_tool_with_retry(func_name, func_args)
console.log(`📦 结果: ${$1}`);
// messages.append({
// "role": "tool",
// "tool_call_id": tool_call.id,
// "content": json.dumps(result, ensure_ascii=False),
// })
// 测试
console.log(run_agent_with_tracer("北京明天适不适合出游？"));
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// import time
// import json
// from openai import OpenAI
// Python: client = OpenAI()
// AgentTracer - CLI Agent class
type AgentTracer struct {
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.loops = [] # 每次循环的记录
// Python: self.total_tokens = 0
// Python: self.total_time = 0.0
// Python: self.tool_calls_count = 0
func record_loop() {
// Python: usage = response.usage
// Python: loop_info = {
// Python: "loop": loop_num,
// Python: "prompt_tokens": usage.prompt_tokens if usage else 0,
// Python: "completion_tokens": usage.completion_tokens if usage else 0,
// Python: "total_tokens": usage.total_tokens if usage else 0,
// Python: "time_seconds": round(loop_time, 2),
// Python: "tool_calls": tool_names or [],
// Python: "has_tool_call": len(tool_names or []) > 0,
// Python: }
// Python: self.loops.append(loop_info)
// Python: self.total_tokens += loop_info["total_tokens"]
// Python: self.total_time += loop_time
if tool_names {
// Python: self.tool_calls_count += len(tool_names)
func print_report() {
fmt.Println("\n" + "="*50)
fmt.Println("📊 Agent 调用链分析报告")
fmt.Println("="*50)
fmt.Println(f"总循环次数: {len(self.loops)}")
fmt.Println(f"总工具调用: {self.tool_calls_count} 次")
fmt.Println(f"总 token 消耗: {self.total_tokens}")
fmt.Println(f"总耗时: {round(self.total_time, 2)}s")
fmt.Println(f"\n--- 每次循环详情 ---")
for _, loop := range self.loops {
// Python: role = "🔧 工具调用" if loop["has_tool_call"] else "💬 最终回答"
fmt.Println(f" 循环 {loop['loop']}: {role}")
fmt.Println(f" tokens: {loop['total_tokens']} (prompt: {loop['prompt_tokens']}, completion: {loop['completion_tokens']})")
fmt.Println(f" 延迟: {loop['time_seconds']}s")
if loop['tool_calls'] {
fmt.Println(f" 工具: {', '.join(loop['tool_calls'])}")
fmt.Println("="*50)
// 成本估算（以 GPT-4o-mini 为例）
// Python: cost_per_1k_input = 0.15 / 1000 # $0.15/1M tokens → $0.00015/1K
// Python: cost_per_1k_output = 0.60 / 1000 # $0.60/1M tokens → $0.0006/1K
// Python: input_tokens = sum(l["prompt_tokens"] for l in self.loops)
// Python: output_tokens = sum(l["completion_tokens"] for l in self.loops)
// Python: estimated_cost = input_tokens * cost_per_1k_input + output_tokens * cost_per_1k_output
fmt.Println(f"💰 预估成本: ${estimated_cost:.6f}")
// Python: def run_agent_with_tracer(user_message: str) -> str:
// Python: tracer = AgentTracer()
// Python: messages = [
// Python: {
// Python: "role": "system",
// Python: "content": "你是一个出行助手，可以查询天气和规划行程。"
// Python: },
// Python: {"role": "user", "content": user_message},
// Python: ]
// Python: tool_map = {
// Python: "get_weather": get_weather,
// Python: "plan_trip": plan_trip,
// Python: }
// Python: loop_num = 0
// Python: while True:
// Python: loop_num += 1
// Python: start_time = time.time()
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=tools_schema,
// Python: )
// Python: loop_time = time.time() - start_time
// Python: msg = response.choices[0].message
// 记录工具调用名
// Python: tool_names = [tc.function.name for tc in (msg.tool_calls or [])]
// Python: tracer.record_loop(loop_num, response, loop_time, tool_names)
if not msg.tool_calls {
// Python: tracer.print_report()
return msg.content
// Python: messages.append(msg)
for _, tool_call := range msg.tool_calls {
// Python: func_name = tool_call.function.name
// Python: func_args = json.loads(tool_call.function.arguments)
fmt.Println(f"🔧 调用: {func_name}({func_args})")
// Python: result = execute_tool_with_retry(func_name, func_args)
fmt.Println(f"📦 结果: {result}")
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试
fmt.Println(run_agent_with_tracer("北京明天适不适合出游？"))
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// import time
// import json
// from openai import OpenAI
// Python: client = OpenAI()
public class AgentTracer {
public AgentTracer() {
// Python: self.loops = [] # 每次循环的记录
// Python: self.total_tokens = 0
// Python: self.total_time = 0.0
// Python: self.tool_calls_count = 0
public static void record_loop(loop_numint, response, loop_timedouble, tool_names: list ) {
// Python: usage = response.usage
// Python: loop_info = {
// Python: "loop": loop_num,
// Python: "prompt_tokens": usage.prompt_tokens if usage else 0,
// Python: "completion_tokens": usage.completion_tokens if usage else 0,
// Python: "total_tokens": usage.total_tokens if usage else 0,
// Python: "time_seconds": round(loop_time, 2),
// Python: "tool_calls": tool_names or [],
// Python: "has_tool_call": len(tool_names or []) > 0,
// Python: }
// Python: self.loops.append(loop_info)
// Python: self.total_tokens += loop_info["total_tokens"]
// Python: self.total_time += loop_time
if (tool_names) {
// Python: self.tool_calls_count += len(tool_names)
public static void print_report() {
System.out.println("\n" + "="*50);
System.out.println("📊 Agent 调用链分析报告");
System.out.println("="*50);
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
for (var loop : self.loops) {
// Python: role = "🔧 工具调用" if loop["has_tool_call"] else "💬 最终回答"
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
if (loop['tool_calls']) {
System.out.println(String.format("$1"));
System.out.println("="*50);
// 成本估算（以 GPT-4o-mini 为例）
// Python: cost_per_1k_input = 0.15 / 1000 # $0.15/1M tokens → $0.00015/1K
// Python: cost_per_1k_output = 0.60 / 1000 # $0.60/1M tokens → $0.0006/1K
// Python: input_tokens = sum(l["prompt_tokens"] for l in self.loops)
// Python: output_tokens = sum(l["completion_tokens"] for l in self.loops)
// Python: estimated_cost = input_tokens * cost_per_1k_input + output_tokens * cost_per_1k_output
System.out.println(String.format("$1"));
// Python: def run_agent_with_tracer(user_message: str) -> str:
// Python: tracer = AgentTracer()
// Python: messages = [
// Python: {
// Python: "role": "system",
// Python: "content": "你是一个出行助手，可以查询天气和规划行程。"
// Python: },
// Python: {"role": "user", "content": user_message},
// Python: ]
// Python: tool_map = {
// Python: "get_weather": get_weather,
// Python: "plan_trip": plan_trip,
// Python: }
// Python: loop_num = 0
// Python: while True:
// Python: loop_num += 1
// Python: start_time = time.time()
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=tools_schema,
// Python: )
// Python: loop_time = time.time() - start_time
// Python: msg = response.choices[0].message
// 记录工具调用名
// Python: tool_names = [tc.function.name for tc in (msg.tool_calls or [])]
// Python: tracer.record_loop(loop_num, response, loop_time, tool_names)
if (!msg.tool_calls) {
// Python: tracer.print_report()
return msg.content;
// Python: messages.append(msg)
for (var tool_call : msg.tool_calls) {
// Python: func_name = tool_call.function.name
// Python: func_args = json.loads(tool_call.function.arguments)
System.out.println(String.format("$1"));
// Python: result = execute_tool_with_retry(func_name, func_args)
System.out.println(String.format("$1"));
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试
System.out.println(run_agent_with_tracer("北京明天适不适合出游？"));
}
}
运行效果：
**🔧 调用: get_weather({'city': '北京', 'date': 'tomorrow'})**
**📦 结果: {'city': '北京', 'date': 'tomorrow', 'weather': '多云', 'temperature': '23°C'}**
**🔧 调用: plan_trip({'city': '北京', 'weather_condition': '多云', 'days': 1})**
**📦 结果: {'city': '北京', 'weather_condition': '多云', 'recommendation': '适合逛博物馆、胡同，室内外均可', 'days': 1}**
==================================================
**📊 Agent 调用链分析报告**
==================================================
总循环次数: 3
总工具调用: 2 次
总 token 消耗: 1478
总耗时: 3.21s
--- 每次循环详情 ---
循环 1: 🔧 工具调用
tokens: 523 (prompt: 385, completion: 138)
延迟: 1.12s
工具: get_weather
循环 2: 🔧 工具调用
tokens: 675 (prompt: 532, completion: 143)
廳迟: 1.35s
工具: plan_trip
循环 3: 💬 最终回答
tokens: 280 (prompt: 210, completion: 70)
延迟: 0.74s
==================================================
💰 预估成本: $0.000354
北京明天多云23°C，适合逛博物馆和胡同，室内外活动均可。
3.10.3 关键发现：token 递增现象
注意循环 2 的 prompt_tokens (532) > 循环 1 的 prompt_tokens (385)。为什么？因为每次循环都把上一次的消息加入历史：
# 循环 1 的 messages:
# [system, user] → prompt_tokens: 385
# 循环 2 的 messages:
# [system, user, AI_msg(含工具调用), tool_result] → prompt_tokens: 532
# 循环 3 的 messages:
# [system, user, AI_msg, tool_result, AI_msg, tool_result] → prompt_tokens: 210
# （注意：最终回答的 prompt 包含了所有历史）
这就是 token 递增问题：Agent 循环次数越多，历史越长，token 消耗越大。对于简单任务，这不是问题；对于需要 10+ 次循环的复杂任务，成本可能飙升。 ⚠️ 生产环境的 token 管理
- 设置最大循环次数：防止无限循环（如 max_loops=10）
- 设置最大 token 预算：超过预算就停止（如 max_tokens=5000）
- 压缩历史消息：循环超过 N 次后，摘要之前的对话
- 选择合适的模型：简单任务用 gpt-4o-mini，复杂任务用 gpt-4o
3.10.4 不同任务的调用链对比 | 用户问题 | 循环次数 | 工具调用 | 总 tokens | 耗时 | 成本 | | --- | --- | --- | --- | --- | --- | | 北京明天天气？ | 2 | 1 (get_weather) | ~500 | 1.5s | $0.0001 | | 北京明天适合出游？ | 3 | 2 (天气+行程) | ~1500 | 3.2s | $0.0004 | | 帮我规划上海2天行程 | 3 | 2 (天气+行程) | ~1800 | 4.0s | $0.0005 | | 北京上海广州三城天气对比 | 4 | 3 (三次天气) | ~2500 | 5.5s | $0.0008 | 从数据可以看出：工具越多、循环越多，成本和延迟越高。这不是线性的——因为 token 递增效应，第 N 次循环的成本比第 1 次更高。
2.10 节给了我们量化的视角。数据告诉我们，Agent 不是万能的——有时候它太贵、太慢。那么什么时候不该用 Agent？什么时候必须用？这正是下一节的核心问题。
3.11 Agent 的边界
Agent 很酷，但不是万能药。理解 Agent 的能力边界，比理解它的能力更重要——这样你才不会在应该写普通程序的场景里强行用 Agent。
3.11.1 什么时候 Agent 不如传统程序？
以下场景，传统程序完胜 Agent： | 场景 | 传统程序 | Agent | 胜者 | | --- | --- | --- | --- | | CRUD 操作（增删改查） | 确定性逻辑，100% 准确 | 可能理解错意图 | ❌ 传统 | | 数学计算 | 精确无误 | LLM 算术不可靠 | ❌ 传统 | | 格式转换（CSV → JSON） | 规则明确，速度快 | 没必要让 AI 决策 | ❌ 传统 | | 定时批处理任务 | 稳定、低成本 | 每次调用都花钱 | ❌ 传统 | | 高并发低延迟 API | 毫秒级响应 | 秒级响应，成本高 | ❌ 传统 | 核心原因：确定性任务不需要决策，而决策是 Agent 的核心价值。给一个不需要决策的任务加 Agent，就像给计算器装个大脑——费钱还更慢。
3.11.2 什么时候必须用 Agent？
反过来，以下场景 Agent 完胜传统程序： | 场景 | 传统程序 | Agent | 胜者 | | --- | --- | --- | --- | | 自然语言交互 | 需要写大量解析规则 | 天然理解意图 | ✅ Agent | | 多步骤编排（先查再规划） | 需要硬编码所有流程 | 自主决定步骤顺序 | ✅ Agent | | 灵活错误处理 | 每种错误写 if-else | 根据错误自行调整 | ✅ Agent | | 跨领域问答 | 无法覆盖所有领域 | 通用推理能力 | ✅ Agent | | 长尾场景（罕见情况） | 无法穷举 | 临场推理 | ✅ Agent | 核心原因：不确定性任务需要决策，而 AI 的推理能力远超硬编码规则。
3.11.3 混合架构：最佳实践
最聪明的做法不是"全用 Agent"或"全不用 Agent"，而是混合架构：
# === 混合架构示例：天气出行系统 ===
# 确定性部分：用传统程序
def calculate_route(start: str, end: str) -> dict:
"""路线计算 → 传统程序（确定性逻辑）"""
routes = {
"北京-上海": {"distance": "1200km", "time": "高铁4小时"},
"北京-广州": {"distance": "2200km", "time": "高铁8小时"},
"上海-广州": {"distance": "1400km", "time": "高铁6小时"},
}
key = f"{start}-{end}"
if key in routes:
return routes[key]
# 反向查找
key_rev = f"{end}-{start}"
if key_rev in routes:
r = routes[key_rev]
return {"distance": r["distance"], "time": r["time"] + "（反向）"}
return {"error": f"暂不支持 {start} 到 {end} 的路线"}
def format_weather_report(weather: dict) -> str:
"""格式化天气报告 → 传统程序（确定性逻辑）"""
if "error" in weather:
return f"⚠️ {weather['error']}"
return f"{weather['city']} {weather['date']}: {weather['weather']}，气温{weather['temperature']}"
# 不确定性部分：用 Agent
# （复用 2.8 的多工具 Agent）
def run_hybrid_agent(user_message: str) -> str:
"""混合架构 Agent：确定性用传统程序，不确定性用 AI"""
# 先让 Agent 处理意图理解和工具调用
messages = [
{
"role": "system",
"content": "你是一个出行助手。查询天气、规划行程、计算路线。\n"
"天气和行程用工具查，路线计算用 calculate_route 工具。",
},
{"role": "user", "content": user_message},
]
# 扩展 Schema，加入路线计算工具
hybrid_tools = tools_schema + [
{
"type": "function",
"function": {
"name": "calculate_route",
"description": "计算两个城市之间的路线信息（距离和交通时间）",
"parameters": {
"type": "object",
"properties": {
"start": {
"type": "string",
"description": "出发城市"
},
"end": {
"type": "string",
"description": "目的地城市"
}
},
"required": ["start", "end"]
}
}
}
]
tool_map = {
"get_weather": get_weather,
"plan_trip": plan_trip,
"calculate_route": calculate_route, # 传统程序！
}
while True:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=hybrid_tools,
)
msg = response.choices[0].message
if not msg.tool_calls:
return msg.content
messages.append(msg)
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"🔧 调用: {func_name}({func_args})")
result = execute_tool_with_retry(func_name, func_args)
print(f"📦 结果: {result}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False),
})
# 测试混合架构
print(run_hybrid_agent("我想从北京去上海玩两天，帮我规划一下"))
// === 混合架构示例：天气出行系统 ===
// 确定性部分：用传统程序
// def calculate_route(start: str, end: str) -> dict:
/** docstring */
const routes = {;
// "北京-上海": {"distance": "1200km", "time": "高铁4小时"},
// "北京-广州": {"distance": "2200km", "time": "高铁8小时"},
// "上海-广州": {"distance": "1400km", "time": "高铁6小时"},
// }
const key = `${$1}-${$1}`;
if (key in routes) {
return routes[key];
// 反向查找
const key_rev = `${$1}-${$1}`;
if (key_rev in routes) {
const r = routes[key_rev];
return {"distance": r["distance"], "time": r["time"] + "（反向）"};
return {"error": `暂不支持 ${$1} 到 ${$1} 的路线`};
// def format_weather_report(weather: dict) -> str:
/** docstring */
if ("error" in weather) {
return `⚠️ {weather['error']}`;
return `{weather['city']} {weather['date']}: {weather['weather']}，气温{weather['temperature']}`;
// 不确定性部分：用 Agent
// （复用 2.8 的多工具 Agent）
// def run_hybrid_agent(user_message: str) -> str:
/** docstring */
// 先让 Agent 处理意图理解和工具调用
const messages = [;
// {
// "role": "system",
// "content": "你是一个出行助手。查询天气、规划行程、计算路线。\n"
// "天气和行程用工具查，路线计算用 calculate_route 工具。",
// },
// {"role": "user", "content": user_message},
// ]
// 扩展 Schema，加入路线计算工具
const hybrid_tools = tools_schema + [;
// {
// "type": "function",
// "function": {
// "name": "calculate_route",
// "description": "计算两个城市之间的路线信息（距离和交通时间）",
// "parameters": {
// "type": "object",
// "properties": {
// "start": {
// "type": "string",
// "description": "出发城市"
// },
// "end": {
// "type": "string",
// "description": "目的地城市"
// }
// },
// "required": ["start", "end"]
// }
// }
// }
// ]
const tool_map = {;
// "get_weather": get_weather,
// "plan_trip": plan_trip,
// "calculate_route": calculate_route, # 传统程序！
// }
while (true) {
const response = client.chat.completions.create(;
const model = "gpt-4o-mini",;
const messages = messages,;
const tools = hybrid_tools,;
// )
const msg = response.choices[0].message;
if (!msg.tool_calls) {
return msg.content;
// messages.append(msg)
for (const tool_call of msg.tool_calls) {
const func_name = tool_call.function.name;
const func_args = json.loads(tool_call.function.arguments);
console.log(`🔧 调用: ${$1}(${$1})`);
const result = execute_tool_with_retry(func_name, func_args);
console.log(`📦 结果: ${$1}`);
// messages.append({
// "role": "tool",
// "tool_call_id": tool_call.id,
// "content": json.dumps(result, ensure_ascii=False),
// })
// 测试混合架构
console.log(run_hybrid_agent("我想从北京去上海玩两天，帮我规划一下"));
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// === 混合架构示例：天气出行系统 ===
// 确定性部分：用传统程序
// Python: def calculate_route(start: str, end: str) -> dict:
// Python: routes = {
// Python: "北京-上海": {"distance": "1200km", "time": "高铁4小时"},
// Python: "北京-广州": {"distance": "2200km", "time": "高铁8小时"},
// Python: "上海-广州": {"distance": "1400km", "time": "高铁6小时"},
// Python: }
// Python: key = f"{start}-{end}"
if key in routes {
return routes[key]
// 反向查找
// Python: key_rev = f"{end}-{start}"
if key_rev in routes {
// Python: r = routes[key_rev]
return {"distance": r["distance"], "time": r["time"] + "（反向）"}
return {"error": f"暂不支持 {start} 到 {end} 的路线"}
// Python: def format_weather_report(weather: dict) -> str:
if "error" in weather {
return f"⚠️ {weather['error']}"
return f"{weather['city']} {weather['date']}: {weather['weather']}，气温{weather['temperature']}"
// 不确定性部分：用 Agent
// （复用 2.8 的多工具 Agent）
// Python: def run_hybrid_agent(user_message: str) -> str:
// 先让 Agent 处理意图理解和工具调用
// Python: messages = [
// Python: {
// Python: "role": "system",
// Python: "content": "你是一个出行助手。查询天气、规划行程、计算路线。\n"
// Python: "天气和行程用工具查，路线计算用 calculate_route 工具。",
// Python: },
// Python: {"role": "user", "content": user_message},
// Python: ]
// 扩展 Schema，加入路线计算工具
// Python: hybrid_tools = tools_schema + [
// Python: {
// Python: "type": "function",
// Python: "function": {
// Python: "name": "calculate_route",
// Python: "description": "计算两个城市之间的路线信息（距离和交通时间）",
// Python: "parameters": {
// Python: "type": "object",
// Python: "properties": {
// Python: "start": {
// Python: "type": "string",
// Python: "description": "出发城市"
// Python: },
// Python: "end": {
// Python: "type": "string",
// Python: "description": "目的地城市"
// Python: }
// Python: },
// Python: "required": ["start", "end"]
// Python: }
// Python: }
// Python: }
// Python: ]
// Python: tool_map = {
// Python: "get_weather": get_weather,
// Python: "plan_trip": plan_trip,
// Python: "calculate_route": calculate_route, # 传统程序！
// Python: }
// Python: while True:
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=hybrid_tools,
// Python: )
// Python: msg = response.choices[0].message
if not msg.tool_calls {
return msg.content
// Python: messages.append(msg)
for _, tool_call := range msg.tool_calls {
// Python: func_name = tool_call.function.name
// Python: func_args = json.loads(tool_call.function.arguments)
fmt.Println(f"🔧 调用: {func_name}({func_args})")
// Python: result = execute_tool_with_retry(func_name, func_args)
fmt.Println(f"📦 结果: {result}")
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试混合架构
fmt.Println(run_hybrid_agent("我想从北京去上海玩两天，帮我规划一下"))
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// === 混合架构示例：天气出行系统 ===
// 确定性部分：用传统程序
// Python: def calculate_route(start: str, end: str) -> dict:
// Python: routes = {
// Python: "北京-上海": {"distance": "1200km", "time": "高铁4小时"},
// Python: "北京-广州": {"distance": "2200km", "time": "高铁8小时"},
// Python: "上海-广州": {"distance": "1400km", "time": "高铁6小时"},
// Python: }
// Python: key = f"{start}-{end}"
if (key in routes) {
return routes[key];
// 反向查找
// Python: key_rev = f"{end}-{start}"
if (key_rev in routes) {
// Python: r = routes[key_rev]
return {"distance": r["distance"], "time": r["time"] + "（反向）"};
return {"error": f"暂不支持 {start} 到 {end} 的路线"};
// Python: def format_weather_report(weather: dict) -> str:
if ("error" in weather) {
return f"⚠️ {weather['error']}";
return f"{weather['city']} {weather['date']}: {weather['weather']}，气温{weather['temperature']}";
// 不确定性部分：用 Agent
// （复用 2.8 的多工具 Agent）
// Python: def run_hybrid_agent(user_message: str) -> str:
// 先让 Agent 处理意图理解和工具调用
// Python: messages = [
// Python: {
// Python: "role": "system",
// Python: "content": "你是一个出行助手。查询天气、规划行程、计算路线。\n"
// Python: "天气和行程用工具查，路线计算用 calculate_route 工具。",
// Python: },
// Python: {"role": "user", "content": user_message},
// Python: ]
// 扩展 Schema，加入路线计算工具
// Python: hybrid_tools = tools_schema + [
// Python: {
// Python: "type": "function",
// Python: "function": {
// Python: "name": "calculate_route",
// Python: "description": "计算两个城市之间的路线信息（距离和交通时间）",
// Python: "parameters": {
// Python: "type": "object",
// Python: "properties": {
// Python: "start": {
// Python: "type": "string",
// Python: "description": "出发城市"
// Python: },
// Python: "end": {
// Python: "type": "string",
// Python: "description": "目的地城市"
// Python: }
// Python: },
// Python: "required": ["start", "end"]
// Python: }
// Python: }
// Python: }
// Python: ]
// Python: tool_map = {
// Python: "get_weather": get_weather,
// Python: "plan_trip": plan_trip,
// Python: "calculate_route": calculate_route, # 传统程序！
// Python: }
// Python: while True:
// Python: response = client.chat.completions.create(
// Python: model="gpt-4o-mini",
// Python: messages=messages,
// Python: tools=hybrid_tools,
// Python: )
// Python: msg = response.choices[0].message
if (!msg.tool_calls) {
return msg.content;
// Python: messages.append(msg)
for (var tool_call : msg.tool_calls) {
// Python: func_name = tool_call.function.name
// Python: func_args = json.loads(tool_call.function.arguments)
System.out.println(String.format("$1"));
// Python: result = execute_tool_with_retry(func_name, func_args)
System.out.println(String.format("$1"));
// Python: messages.append({
// Python: "role": "tool",
// Python: "tool_call_id": tool_call.id,
// Python: "content": json.dumps(result, ensure_ascii=False),
// Python: })
// 测试混合架构
System.out.println(run_hybrid_agent("我想从北京去上海玩两天，帮我规划一下"));
}
运行效果：
**🔧 调用: get_weather({'city': '上海', 'date': 'tomorrow'})**
**📦 结果: {'city': '上海', 'date': 'tomorrow', 'weather': '阴', 'temperature': '27°C'}**
**🔧 调用: plan_trip({'city': '上海', 'weather_condition': '阴', 'days': 2})**
**📦 结果: {'city': '上海', 'weather_condition': '阴', 'recommendation': '南京路逛街、田子坊艺术区', 'days': 2}**
**🔧 调用: calculate_route({'start': '北京', 'end': '上海'})**
**📦 结果: {'distance': '1200km', 'time': '高铁4小时'}**
好的！从北京到上海两天行程建议：
- 交通：高铁4小时，距离1200km
- 天气：上海明天阴天27°C
- 行程：南京路逛街、田子坊艺术区
- 建议带伞，阴天可能有小雨
注意 calculate_route 是纯确定性逻辑——AI 不会"推理"路线，它只是判断"用户需要路线 → 调用 calculate_route"。确定性计算交给传统程序，不确定性决策交给 AI，这就是混合架构的精髓。
💡 选择架构的决策树
问自己三个问题：(1) 步骤是否固定？→ 固定用传统程序。(2) 输入是否结构化？→ 结构化用传统程序。(3) 是否需要理解自然语言？→ 需要就用 Agent。如果部分固定部分灵活 → 混合架构。
3.11.4 Agent 的成本陷阱
即使该用 Agent 的场景，也要警惕成本陷阱：
# 成本对比：查天气的三种方式
# 方式1：直接调用天气API（传统程序）
# 成本：0（免费API） | 延迟：0.1s | 准确率：100%
# 方式2：硬编码规则匹配（传统程序 + NLU）
# 成本：0 | 延迟：0.05s | 准确率：80%（无法处理"明天"等模糊词）
# 方式3：Agent（AI + 工具）
# 成本：$0.0001/次 | 延迟：1.5s | 准确率：95%（能理解"明天"、"后天"等）
# 结论：如果每天查询10次 → Agent成本$0.001/天，可以接受
# 如果每天查询10000次 → Agent成本$1/天，需要考虑
# 如果每天查询1000000次 → Agent成本$100/天，必须用传统程序
关键指标：每秒查询数（QPS）和单次成本。当 QPS × 单次成本 > 你的预算，就该考虑传统程序或混合架构。
2.11 节讨论了 Agent 的边界。理解边界不是为了否定 Agent，而是为了在正确的场景使用正确的工具。现在，让我们回顾整章内容，做个全面总结。
3.12 本章小结
🔗 本章核心
Agent = AI + 工具 + 循环：三者缺一不可。没有工具的 AI 只能聊天，没有循环的 AI 只能做一次工具调用，没有异常处理的 Agent 在现实世界会崩溃。
工具 Schema 是桥梁：description 决定 AI 是否调用，parameters 决定 AI 怎么调用。description 写得好，Agent 就聪明；写得差，Agent 就困惑。
Agent 循环：感知→决策→行动→观察→循环，直到 AI 不需要再调工具。多轮对话让循环跨多轮交互。
异常处理三策略：可重试的用指数退避，不可重试的用优雅降级，所有错误都要告诉 AI。
多工具协作：AI 自行编排工具调用顺序——先查天气再规划行程，不需要硬编码流程。
调用链分析：每次循环都消耗 token，历史递增导致成本递增。量化分析是优化 Agent 的前提。
Agent 的边界：确定性任务用传统程序，不确定性任务用 Agent，混合架构是最佳实践。
本章局限性：我们的天气 Agent 用的是简单的"用户问→调用工具→回答"模式。它没有在调用工具前思考为什么要调用，也没有在观察结果后反思是否需要进一步行动。比如用户问"北京明天适合户外运动吗？"，简单 Agent 只查天气就回答了，但更聪明的 Agent 应该先推理（"适合户外需要看天气+空气质量+温度"），再分别调用工具，最后综合判断——这就是ReAct 模式的价值。
下一步：第5章 ReAct 模式会在这个循环基础上加入"推理"环节，让 Agent 在行动前先思考，减少不必要的工具调用。 📊 本章知识图谱
从简单到复杂的递进路径：
- 3.1-3.4：理解问题 → 准备环境 → 定义工具 → 设计 Schema（入门）
- 3.5-3.6：组装 Agent → 理解循环（核心）
- 3.7：多轮对话 → Agent 有���忆（增强）
- 3.8：异常处理 → Agent 不崩溃（健壮性）
- 3.9：多工具协作 → Agent 自主编排（进化）
- 3.10：调用链分析 → 量化成本（工程化）
- 3.11：Agent 边界 → 理性选择架构（工程智慧） 📋 八股总结 — 面试高频考点
Q1: Agent 和普通 LLM 调用的区别是什么？
普通 LLM 调用是一问一答，没有工具；Agent 是循环执行，AI 可以调用工具、观察结果、再推理，直到任务完成。核心区别在于行动能力和循环决策能力。
Q2: 工具 Schema 中最重要的字段是什么？
description。AI 靠它判断是否调用工具，写得越具体越准确。name 只是标识，parameters 决定怎么调用，但 AI 首先靠 description 决定"要不要调"。
Q3: Agent 循环什么时候停止？
当 AI 不再请求调用工具时（finish_reason 为 stop），循环结束，返回最终回答。需设置 max_loops 防止无限循环。
Q4: 为什么需要多轮对话？
AI 可能需要多次工具调用才能完成任务；用户可能追问相关问题，需要从历史中获取上下文。多轮对话让 Agent 具备"记忆"能力。
Q5: 指数退避重试的原理和适用场景？
等待时间按 2^n 递增（1s→2s→4s），适用于临时性故障（超时、限流、503），不适用于逻辑性错误（参数错误、权限不足）。生产环境需加入随机 jitter 防止雪崩。
Q6: 多工具协作中 AI 如何决定调用顺序？
AI 根据工具 description 和用户意图自行推理调用顺序，无需硬编码流程。代价是出错概率增加、延迟和成本递增。
Q7: Agent 调用链的 token 递增现象？
每次循环把上一次消息加入历史，导致 prompt_tokens 逐次递增。循环次数越多，单次循环成本越高。需设置 max_loops 和 token 预算上限。 📝 思考题
思考1: 场景判断
"用户输入城市名，返回固定格式天气报告"——这个需求该用 Agent 还是传统程序？如果是"用户用自然语言问天气，可能追问、可能对比多个城市"——又该用什么？
思考2: 重试设计
一个 Agent 同时调用天气 API 和路线 API，天气超时了但路线成功。如果直接重试整个循环，路线结果就浪费了。如何设计重试机制，只重试失败的工具，而不是重试整个循环？
思考3: 成本优化
你的天气 Agent 每天处理 50000 次请求，每次平均消耗 800 tokens。按 GPT-4o-mini 价格（$0.15/1M input, $0.60/1M output），月成本是多少？如果 80% 的请求只是"北京今天天气"这种简单查询，如何用混合架构降低成本？
第3章 你的第一个Agent-天气查询助手
http://www.clxhxhhr.top/posts/706/ 评论
0 条
还没有评论，先写一条吧。