阶段 1：Agent 基础 ⭐ 必学

一句话总结：Agent = LLM + 工具 + 循环——让 AI 不只是回答问题，而是自主完成任务。

📊 学习进度
状态：⬜ 未开始
预计时长：3-4 小时
已完成：0/3 个模块
在整体流程中的位置：AI Agent 开发·第 1 阶段

📍 本章定位
服务方案：方案 1（重要 60%）/ 方案 3（核心 80%）
学习方式：⭐ 必学
在流程中的作用：理解 Agent 的核心概念和基本架构
核心知识点：ReAct、Tool Use、Function Calling
预计时长：3-4 小时
完成后能做什么：能搭建一个单 Agent 系统

1. 传统模式：痛点与瓶颈

1.1 传统 AI 应用的角色定位

传统 AI 应用在企业中通常扮演"问答助手"角色——用户输入问题，AI 返回答案，交互结束。这种模式在客服、翻译、摘要等场景下效率很高，但面对需要多步骤、多工具协作的复杂任务时，暴露了根本性局限。

传统 AI 应用的典型工作日：

09:00 用户提问 → AI 回答
09:05 用户根据答案手动执行下一步
09:10 用户再次提问（携带上一步结果）
09:15 AI 回答 → 用户再次执行
循环往复...

1.2 沟通效率与协作成本

传统模式下，用户需要在多个工具间手动切换，每次切换都产生"上下文损耗"：

损耗类型	说明	量化数据
上下文重复输入	每次调用需重新描述背景	平均重复 4.2 次/任务
工具切换成本	在不同应用间复制粘贴	平均切换 3.8 次/任务
结果格式转换	不同工具输出格式不一致	转换耗时占总时间 25%
错误累积	手动传递数据易出错	错误率 15%

1.3 量化痛点数据

据 LangChain 2025 年基准测试数据：

指标	传统 AI 应用	Agent 模式	改善幅度
多步骤任务完成率	23%	89%	+287%
平均交互轮次	8.2 轮	1.3 轮	-84%
任务完成时间	12 分钟	2.5 分钟	-79%
Token 浪费率	45%	12%	-73%

2. OPC 模式：重新定义

2.1 核心理念

Agent 的本质是一个"思考-行动-观察"的循环。与传统 AI 应用的根本区别在于：Agent 能自主决定下一步做什么，而不是被动等待用户指令。

Agent 三要素：

要素	角色	说明	OPC 价值
LLM	大脑	负责理解意图、推理决策	替代人类思考
工具	手脚	负责执行具体操作	替代人类执行
循环	意志	持续思考直到任务完成	替代人类跟进

2.2 人机分工矩阵

任务	人类角色	AI 角色	协作方式
需求定义	明确目标和边界	理解并拆解	人说→AI 理解
架构设计	设计 Agent 能力边界	建议最佳实践	人决策→AI 辅助
代码实现	审核代码质量	生成代码	AI 写→人审
测试验证	判断结果正确性	执行测试	AI 跑→人判
Prompt 工程	定义行为准则	优化 Prompt	人定→AI 优化

2.3 效率对比

指标	传统模式	OPC+Agent 模式	提效倍数
多步骤任务完成时间	12 分钟	2.5 分钟	4.8x
工具调用错误率	15%	6%	2.5x
上下文重复输入次数	4.2 次	0.3 次	14x
单日可处理任务数	3-5 个	15-20 个	4x

3. 实操案例

3.1 场景描述

场景：搭建一个"加密货币价格查询 Agent"，用户输入自然语言问题，Agent 自动调用价格 API 获取数据并回答。

技术栈：Claude API + Python + CoinGecko API

3.2 执行过程

3.2.1 Agent 核心循环：ReAct 模式

ReAct（Reasoning + Acting） 是 Agent 最核心的执行模式：

ReAct 执行流程示例：

用户：帮我查一下今天 ETH 的价格，以及和昨天相比涨了多少

Agent 思考（Reasoning）：用户需要两个信息——今天价格和涨跌幅。
我需要先获取今天的 ETH 价格，再获取昨天的价格。

Agent 行动（Acting）：调用 get_price(symbol="ETH", date="today")
Agent 观察（Observation）：ETH 今天价格 = $3,500

Agent 思考（Reasoning）：还需要昨天的价格来计算涨跌幅。
Agent 行动（Acting）：调用 get_price(symbol="ETH", date="yesterday")
Agent 观察（Observation）：ETH 昨天价格 = $3,420

Agent 思考（Reasoning）：现在有了两个数据，可以计算涨跌幅。
涨幅 = (3500 - 3420) / 3420 = 2.34%

Agent 回答：今天 ETH 价格为 $3,500，相比昨天的 $3,420 上涨了 2.34%。

3.2.2 Tool Use 接口定义（Anthropic Claude）

接口文档：

字段	类型	说明
`name`	string	工具名称，必须唯一
`description`	string	工具功能描述，LLM 据此决定是否调用
`input_schema`	object	JSON Schema 格式的参数定义

Python 实现：

python

import anthropic
import json

client = anthropic.Anthropic()

# 定义工具列表
tools = [
    {
        "name": "get_crypto_price",
        "description": "获取加密货币的当前价格。当用户询问某种加密货币的价格时使用此工具。",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "加密货币符号，如 BTC、ETH、SOL"
                },
                "currency": {
                    "type": "string",
                    "description": "计价货币，默认 USD",
                    "default": "USD"
                }
            },
            "required": ["symbol"]
        }
    }
]

# 模拟工具执行函数
def execute_tool(name: str, input_data: dict) -> str:
    """工具执行器 - 根据工具名称分发到对应实现"""
    if name == "get_crypto_price":
        symbol = input_data["symbol"]
        currency = input_data.get("currency", "USD")
        # 实际项目中调用 CoinGecko API
        prices = {"BTC": 67500, "ETH": 3500, "SOL": 145}
        price = prices.get(symbol.upper(), 0)
        return json.dumps({"symbol": symbol, "price": price, "currency": currency})
    return json.dumps({"error": f"Unknown tool: {name}"})

# Agent 主循环
def run_agent(user_message: str, max_iterations: int = 5):
    """Agent 主循环 - ReAct 模式"""
    messages = [{"role": "user", "content": user_message}]
    
    for i in range(max_iterations):
        # 调用 Claude
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        
        # 检查是否需要调用工具
        if response.stop_reason == "tool_use":
            # 提取工具调用
            tool_calls = [b for b in response.content if b.type == "tool_use"]
            tool_results = []
            
            for call in tool_calls:
                result = execute_tool(call.name, call.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": call.id,
                    "content": result
                })
            
            # 将工具结果加入对话
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            # 不需要工具，返回最终回答
            return response.content[0].text
    
    return "达到最大迭代次数，任务未完成"

# 使用示例
result = run_agent("今天 ETH 价格是多少？")
print(result)

TypeScript 实现：

typescript

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

// 工具定义
const tools: Anthropic.Tool[] = [
  {
    name: "get_crypto_price",
    description: "获取加密货币的当前价格",
    input_schema: {
      type: "object" as const,
      properties: {
        symbol: {
          type: "string",
          description: "加密货币符号，如 BTC、ETH",
        },
      },
      required: ["symbol"],
    },
  },
];

// 工具执行器
function executeTool(name: string, input: Record<string, unknown>): string {
  const prices: Record<string, number> = {
    BTC: 67500,
    ETH: 3500,
    SOL: 145,
  };
  if (name === "get_crypto_price") {
    const symbol = (input.symbol as string).toUpperCase();
    return JSON.stringify({ symbol, price: prices[symbol] || 0 });
  }
  return JSON.stringify({ error: `Unknown tool: ${name}` });
}

// Agent 主循环
async function runAgent(
  userMessage: string,
  maxIterations = 5
): Promise<string> {
  const messages: Anthropic.MessageParam[] = [
    { role: "user", content: userMessage },
  ];

  for (let i = 0; i < maxIterations; i++) {
    const response = await client.messages.create({
      model: "claude-sonnet-4-20250514",
      max_tokens: 1024,
      tools,
      messages,
    });

    if (response.stop_reason === "tool_use") {
      const toolCalls = response.content.filter(
        (b) => b.type === "tool_use"
      );
      const toolResults = toolCalls.map((call) => ({
        type: "tool_result" as const,
        tool_use_id: call.id,
        content: executeTool(call.name, call.input as Record<string, unknown>),
      }));

      messages.push({ role: "assistant", content: response.content });
      messages.push({ role: "user", content: toolResults });
    } else {
      const textBlock = response.content.find((b) => b.type === "text");
      return textBlock?.text || "No response";
    }
  }
  return "达到最大迭代次数";
}

// 使用
runAgent("今天 ETH 价格是多少？").then(console.log);

3.2.3 Function Calling（OpenAI）

OpenAI 的 Function Calling 机制与 Claude Tool Use 类似，但语法略有不同：

python

from openai import OpenAI

client = OpenAI()

# 工具定义（OpenAI 称为 functions）
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_crypto_price",
            "description": "获取加密货币的当前价格",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "加密货币符号"
                    }
                },
                "required": ["symbol"]
            }
        }
    }
]

# 调用方式
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "ETH 价格多少？"}],
    tools=tools,
    tool_choice="auto"  # auto | none | required
)

# 检查是否有工具调用
if response.choices[0].message.tool_calls:
    for call in response.choices[0].message.tool_calls:
        print(f"调用工具: {call.function.name}")
        print(f"参数: {call.function.arguments}")

3.3 前后对比

维度	传统模式	Agent 模式	改善
开发时间	2 天（手动串联）	4 小时（Agent 架构）	80%
代码行数	200+ 行（胶水代码）	80 行（核心逻辑）	60%
扩展新工具	改动 50% 代码	添加 1 个工具定义	90%
错误处理	手动 try-catch	框架自动处理	85%

4. 趋势预判（未来 1-3 年）

4.1 技术演进方向

技术方向	当前状态	1 年后	3 年后
Tool Use 准确率	94%	97%	99%
MCP 协议生态	初期	成熟	标准化
Agent 安全框架	缺失	初步建立	完善
多模态工具	文本为主	图文音视频	全模态

4.2 角色变化趋势

角色	当前	1 年后	3 年后
Agent 开发者	稀缺	主流	基础能力
Prompt 工程师	独立岗位	融入开发	内化素养
工具开发者	小众	增长	生态核心

4.3 OPC 需要提前准备的能力

Prompt 工程：写出清晰的工具描述和系统 Prompt
架构思维：设计 Agent 的能力边界和安全策略
成本意识：理解 Token 计费，优化调用效率
调试能力：使用 LangSmith/Langfuse 追踪 Agent 执行过程

5. 核心洞察

🔑 关键洞察

Agent 的核心不是"更聪明的 LLM"，而是"LLM + 工具 + 循环"的组合。一个配备合适工具的 GPT-4o，在特定任务上可以超越没有工具的 Claude Opus。工具定义的质量，比模型选择更重要。

⚠️ 常见陷阱

Agent 死循环是最常见的成本杀手。务必设置 max_iterations（建议 5-10 次）和 Token 预算上限。一个没有退出条件的 Agent，可能在一次任务中消耗 $10+ 的 Token 费用。

6. 参考与延伸

[1] Anthropic. "Tool Use" — Claude 工具使用官方文档（2025）

[2] OpenAI. "Function Calling Guide" — OpenAI 函数调用文档（2025）

[3] Yao et al. "ReAct: Synergizing Reasoning and Acting in Language Models" — ReAct 论文（2022）

[4] LangChain. "Agent Concepts" — Agent 核心概念（2025）

[5] LangSmith. "Tracing" — Agent 执行追踪（2025）

[6] Shinn et al. "Reflexion: Language Agents with Verbal Reinforcement Learning" — 自我反思 Agent 论文（2023）

[7] Wang et al. "Plan-and-Solve Prompting" — 规划执行模式论文（2023）

[8] Anthropic. "Claude Agent SDK" — Agent 开发框架（2025）

[9] Yao et al. "Tree of Thoughts: Deliberate Problem Solving with Large Language Models" — ToT 论文（2023）

[10] Press et al. "Measuring and Narrowing the Compositionality Gap in Language Models" — Self-Ask 论文（2023）

[11] Anthropic. "Claude Prompt Caching" — Prompt 缓存文档（2025）

[12] LangChain. "Streaming" — 流式输出文档（2025）

常见问题

问题	原因	解决方案
Agent 死循环	缺少退出条件	设置 max_iterations（建议 5-10）
工具选择错误	工具描述不清晰	优化 description，添加示例
成本失控	循环次数过多	设置 Token 预算上限
工具参数错误	Schema 不准确	完善 input_schema，添加 enum 约束
响应太慢	模型选择不当	简单任务用 Haiku，复杂任务用 Sonnet

Agent 设计模式

模式 1：单步 Agent

最简单的 Agent 模式，适合简单任务：

适用场景：文本分类、情感分析、简单问答

模式 2：ReAct Agent

最常用的 Agent 模式，支持工具调用：

适用场景：多步骤任务、需要外部数据、需要工具调用

模式 3：规划 Agent

先规划再执行，适合复杂任务：

适用场景：项目管理、复杂分析、多阶段任务

模式 4：Plan-and-Execute Agent

先生成完整计划，再逐步执行，执行过程中可动态调整计划。与"规划 Agent"不同，Plan-and-Execute 在执行阶段会根据实际结果重新规划后续步骤。

适用场景：复杂调研、多阶段项目、需要动态调整的任务

与规划 Agent 的区别：

维度	规划 Agent	Plan-and-Execute Agent
计划时机	执行前一次性规划	执行前规划 + 执行中动态调整
容错能力	计划错误则整个任务失败	单步失败可重新规划后续步骤
适用场景	步骤确定、变数少的任务	步骤不确定、需要探索的任务
Token 消耗	较低	较高（需重新规划）

模式 5：Reflexion Agent

Reflexion 是一种"自我反思"模式——Agent 在执行后会回顾自己的表现，总结经验教训，并在下一次尝试中改进。据 Shinn et al. 2023 年论文，Reflexion 在编程任务上的成功率比标准 ReAct 高出 22%。

Reflexion 执行示例：

用户：帮我写一个 Python 函数，计算两个日期之间的工作日天数。

Agent 第1次尝试：
def count_workdays(start, end):
    return (end - start).days * 5 / 7

自我反思：这个实现忽略了节假日，且没有处理边界情况（开始日期晚于结束日期）。

Agent 第2次尝试：
def count_workdays(start, end):
    if start > end:
        start, end = end, start
    days = (end - start).days
    workdays = 0
    for i in range(days):
        if (start + timedelta(days=i)).weekday() < 5:
            workdays += 1
    return workdays

自我反思：正确处理了边界情况，但没有考虑节假日。在没有节假日数据的情况下，这是合理的实现。

Reflexion vs 标准 ReAct 对比：

指标	标准 ReAct	Reflexion	改善
编程任务成功率	67%	89%	+33%
复杂推理准确率	52%	71%	+37%
平均尝试次数	1.2	2.1	-
Token 消耗	基准	1.8x	-

数据来源：Shinn et al. "Reflexion: Language Agents with Verbal Reinforcement Learning"（2023）

模式 6：Tree-of-Thought Agent

Tree-of-Thought（ToT）是一种"分支探索"模式——Agent 在面对复杂问题时，会同时探索多个可能的解决路径，评估每条路径的可行性，然后选择最优路径继续深入。据 Google DeepMind 2023 年论文，ToT 在需要回溯的推理任务上比标准 ReAct 成功率高 35%。

ToT 与标准 ReAct 对比：

指标	标准 ReAct	Tree-of-Thought	改善
需要回溯的推理任务成功率	48%	83%	+73%
数学问题求解准确率	62%	88%	+42%
平均 Token 消耗	基准	2.5x	-
平均延迟	基准	1.8x	-

适用场景：数学推理、逻辑谜题、需要探索多种可能性的决策问题

ToT 实现示例：

python

from typing import List, Dict
from dataclasses import dataclass

@dataclass
class ThoughtNode:
    """思考节点"""
    thought: str
    score: float
    children: List['ThoughtNode']
    is_terminal: bool = False

class TreeOfThoughtAgent:
    """Tree-of-Thought Agent"""

    def __init__(self, client, model: str = "claude-sonnet-4-20250514"):
        self.client = client
        self.model = model

    async def solve(self, problem: str, max_depth: int = 3, branch_factor: int = 3) -> str:
        """使用 ToT 模式解决问题"""
        # 1. 生成初始思考方向
        thoughts = await self._generate_thoughts(problem, branch_factor)

        # 2. 评估每个方向
        scored_thoughts = []
        for thought in thoughts:
            score = await self._evaluate_thought(problem, thought)
            scored_thoughts.append((score, thought))

        # 3. 选择最有前景的方向继续探索
        scored_thoughts.sort(reverse=True)
        best_score, best_thought = scored_thoughts[0]

        if best_score > 0.8 or max_depth <= 1:
            return best_thought

        # 4. 递归探索
        sub_problem = f"{problem}\n\n当前进展：{best_thought}"
        return await self.solve(sub_problem, max_depth - 1, branch_factor)

    async def _generate_thoughts(self, problem: str, n: int) -> List[str]:
        """生成多个思考方向"""
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"对于以下问题，请生成 {n} 个不同的解决思路（每个一行）：\n\n{problem}"
            }]
        )
        return response.content[0].text.strip().split('\n')[:n]

    async def _evaluate_thought(self, problem: str, thought: str) -> float:
        """评估思考方向的可行性"""
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=100,
            messages=[{
                "role": "user",
                "content": f"问题：{problem}\n\n解决思路：{thought}\n\n请评估这个思路的可行性（0-1 的数字）："
            }]
        )
        try:
            return float(response.content[0].text.strip())
        except ValueError:
            return 0.5

模式 7：Self-Ask Agent

Self-Ask 是一种"自我提问"模式——Agent 在回答复杂问题时，会先将问题分解为多个子问题，逐一回答子问题，最后综合得出最终答案。据 Press et al. 2023 年论文，Self-Ask 在多跳推理任务上的准确率比直接回答高 25%。

Self-Ask 执行示例：

用户：Vitalik Buterin 创立的以太坊，其创始人出生在哪个国家？

Agent 自问：这个问题需要两个步骤：
1. Vitalik Buterin 出生在哪个国家？
2. 确认以太坊是他创立的

Agent 自答子问题1：Vitalik Buterin 1994 年出生于俄罗斯科洛姆纳。
Agent 自答子问题2：是的，Vitalik Buterin 是以太坊的联合创始人。

Agent 综合回答：Vitalik Buterin 出生在俄罗斯。

Self-Ask 实现示例：

python

class SelfAskAgent:
    """Self-Ask Agent - 自我提问分解复杂问题"""

    def __init__(self, client, model: str = "claude-sonnet-4-20250514"):
        self.client = client
        self.model = model

    async def answer(self, question: str, max_sub_questions: int = 5) -> str:
        """使用 Self-Ask 模式回答问题"""
        # 1. 分解问题
        sub_questions = await self._decompose(question)

        # 2. 逐一回答子问题
        sub_answers = []
        for sq in sub_questions[:max_sub_questions]:
            answer = await self._answer_sub_question(sq)
            sub_answers.append(f"问题：{sq}\n答案：{answer}")

        # 3. 综合答案
        context = "\n\n".join(sub_answers)
        final = await self._synthesize(question, context)
        return final

    async def _decompose(self, question: str) -> List[str]:
        """将复杂问题分解为子问题"""
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"请将以下问题分解为需要先回答的子问题（每行一个）：\n\n{question}"
            }]
        )
        return [q.strip() for q in response.content[0].text.strip().split('\n') if q.strip()]

    async def _answer_sub_question(self, question: str) -> str:
        """回答单个子问题"""
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=200,
            messages=[{"role": "user", "content": question}]
        )
        return response.content[0].text.strip()

    async def _synthesize(self, original_question: str, sub_answers: str) -> str:
        """综合子问题答案，得出最终答案"""
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"原始问题：{original_question}\n\n子问题和答案：\n{sub_answers}\n\n请综合以上信息，给出最终答案："
            }]
        )
        return response.content[0].text.strip()

模式对比总结

模式	复杂度	适用场景	成本	推荐度
单步 Agent	低	简单分类、格式转换	低	入门
ReAct Agent	中	多步骤任务、工具调用	中	最常用
规划 Agent	中高	复杂分析、多阶段任务	中高	进阶
Plan-and-Execute	高	探索性任务、变数多	高	高级
Reflexion	高	编程、推理、需要高精度	高	特定场景
Tree-of-Thought	高	数学推理、逻辑谜题	高	特定场景
Self-Ask	中	多跳推理、复合问题	中	推荐

Agent 调试技巧

使用 LangSmith 追踪

LangSmith 是 LangChain 提供的 Agent 追踪平台，可以可视化 Agent 的每一步执行：

python

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# Agent 执行会自动记录到 LangSmith
result = run_agent("今天 ETH 价格是多少？")
# 在 LangSmith 控制台查看执行轨迹

调试检查清单

检查项	说明	工具
Prompt 质量	检查系统 Prompt 是否清晰	LangSmith
工具调用	检查工具选择是否正确	MCP Inspector
Token 消耗	检查 Token 使用是否合理	LangSmith
执行时间	检查每步耗时	Prometheus
错误日志	检查是否有异常	结构化日志

错误处理与重试机制

Agent 在生产环境中会遇到各种异常：API 超时、工具调用失败、Token 限制等。健壮的错误处理是 Agent 可靠运行的关键。

带重试的 Agent 执行器：

python

import asyncio
from typing import Optional
from dataclasses import dataclass

@dataclass
class RetryConfig:
    max_retries: int = 3
    base_delay: float = 1.0    # 基础延迟（秒）
    max_delay: float = 30.0    # 最大延迟（秒）
    exponential_base: float = 2.0  # 指数退避基数

class AgentError(Exception):
    """Agent 执行错误"""
    def __init__(self, message: str, retryable: bool = True, token_used: int = 0):
        super().__init__(message)
        self.retryable = retryable
        self.token_used = token_used

async def run_agent_with_retry(
    user_message: str,
    retry_config: Optional[RetryConfig] = None
) -> str:
    """带重试机制的 Agent 执行器"""
    config = retry_config or RetryConfig()
    total_tokens = 0

    for attempt in range(config.max_retries + 1):
        try:
            result = await run_agent(user_message)
            return result
        except AgentError as e:
            total_tokens += e.token_used

            if not e.retryable:
                raise  # 不可重试的错误直接抛出

            if attempt == config.max_retries:
                raise AgentError(
                    f"达到最大重试次数 ({config.max_retries})，总 Token 消耗: {total_tokens}",
                    retryable=False,
                    token_used=total_tokens
                )

            # 指数退避
            delay = min(
                config.base_delay * (config.exponential_base ** attempt),
                config.max_delay
            )
            print(f"Agent 执行失败（第 {attempt + 1} 次），{delay:.1f} 秒后重试: {e}")
            await asyncio.sleep(delay)

    raise AgentError("未知错误", retryable=False, token_used=total_tokens)

TypeScript 实现：

typescript

interface RetryConfig {
  maxRetries: number;
  baseDelay: number;
  maxDelay: number;
}

async function runAgentWithRetry(
  userMessage: string,
  config: RetryConfig = { maxRetries: 3, baseDelay: 1000, maxDelay: 30000 }
): Promise<string> {
  for (let attempt = 0; attempt <= config.maxRetries; attempt++) {
    try {
      return await runAgent(userMessage);
    } catch (error: any) {
      if (!error.retryable) throw error;
      if (attempt === config.maxRetries) throw error;

      const delay = Math.min(
        config.baseDelay * Math.pow(2, attempt),
        config.maxDelay
      );
      console.log(`重试第 ${attempt + 1} 次，等待 ${delay}ms`);
      await new Promise((r) => setTimeout(r, delay));
    }
  }
  throw new Error("达到最大重试次数");
}

常见错误类型与处理策略：

错误类型	是否可重试	处理策略
API 超时	是	指数退避重试
Rate Limit	是	等待后重试
Token 限制	否	缩短输入或切换模型
工具调用失败	视情况	重试或降级
Prompt 注入检测	否	终止并记录

Agent 性能优化技巧

Token 优化策略

Agent 的循环机制会消耗大量 Token，优化 Token 使用是降低成本的关键。

Token 优化清单：

策略	说明	节省比例	实现方式
系统 Prompt 精简	去除冗余描述，保留核心指令	30-40%	Prompt 压缩、去除示例
上下文裁剪	只传递必要历史，避免全量传递	40-60%	滑动窗口、摘要压缩
工具结果精简	工具返回值只保留必要字段	20-30%	结果过滤、格式化
模型路由	简单任务用小模型，复杂任务用大模型	50-80%	任务复杂度评估
缓存复用	相同查询复用缓存结果	60-90%	语义缓存、精确缓存

模型路由实现示例：

python

from enum import Enum

class ModelTier(Enum):
    HAIKU = "claude-haiku-4-20250414"    # $0.25/1M tokens - 简单任务
    SONNET = "claude-sonnet-4-20250514"  # $3/1M tokens - 中等任务
    OPUS = "claude-opus-4-20250514"      # $15/1M tokens - 复杂任务

def select_model(task_type: str, complexity: str = "medium") -> str:
    """根据任务类型和复杂度选择模型"""
    # 简单任务：分类、格式转换、简单查询
    if task_type in ["classify", "format", "simple_query"]:
        return ModelTier.HAIKU.value
    # 中等任务：分析、总结、翻译
    elif task_type in ["analyze", "summarize", "translate"]:
        return ModelTier.SONNET.value
    # 复杂任务：推理、创作、复杂决策
    elif task_type in ["reason", "create", "complex_decision"]:
        return ModelTier.OPUS.value
    # 默认使用中等模型
    return ModelTier.SONNET.value

语义缓存实现示例：

python

import hashlib
from typing import Optional, Dict
import numpy as np

class SemanticCache:
    """语义缓存 - 相似查询复用结果"""

    def __init__(self, similarity_threshold: float = 0.92):
        self.cache: Dict[str, dict] = {}
        self.threshold = similarity_threshold

    def _compute_hash(self, text: str) -> str:
        """计算文本哈希"""
        return hashlib.md5(text.encode()).hexdigest()

    def get(self, query: str) -> Optional[str]:
        """查询缓存"""
        # 1. 精确匹配
        query_hash = self._compute_hash(query)
        if query_hash in self.cache:
            return self.cache[query_hash]["result"]

        # 2. 语义匹配（简化版：基于关键词重叠度）
        query_words = set(query.lower().split())
        for key, entry in self.cache.items():
            cached_words = set(entry["query"].lower().split())
            overlap = len(query_words & cached_words) / max(len(query_words), 1)
            if overlap >= self.threshold:
                return entry["result"]

        return None

    def set(self, query: str, result: str):
        """设置缓存"""
        query_hash = self._compute_hash(query)
        self.cache[query_hash] = {
            "query": query,
            "result": result
        }

响应延迟优化

优化策略	说明	效果
流式输出	使用 Streaming API，用户边看边等	感知延迟降低 70%
并行工具调用	多个工具同时执行	总延迟降低 50-80%
预测性预加载	预测下一步可能需要的数据	减少等待 30%
异步执行	非阻塞调用，提高并发能力	吞吐量提升 3-5x

并行工具调用实现示例：

python

import asyncio
from typing import List, Dict

async def execute_tools_parallel(tool_calls: List[Dict]) -> List[Dict]:
    """并行执行多个工具调用"""
    async def execute_single(call):
        # 模拟工具执行
        await asyncio.sleep(0.5)  # 模拟 API 调用延迟
        return {"tool": call["name"], "result": "success"}

    # 并行执行所有工具调用
    tasks = [execute_single(call) for call in tool_calls]
    results = await asyncio.gather(*tasks)
    return results

# 使用示例
tool_calls = [
    {"name": "get_price", "args": {"symbol": "BTC"}},
    {"name": "get_price", "args": {"symbol": "ETH"}},
    {"name": "get_news", "args": {"query": "crypto"}}
]

# 串行执行：1.5秒
# 并行执行：0.5秒（节省 67%）
results = asyncio.run(execute_tools_parallel(tool_calls))

Agent 安全最佳实践

权限边界

安全层级	说明	示例
只读	Agent 只能读取数据	查询价格、搜索信息
受限写入	Agent 可以写入，但需审核	生成报告草稿
完全控制	Agent 可以执行任何操作	自动交易（高风险）

安全检查清单

[ ] 设置最大迭代次数（建议 5-10）
[ ] 设置 Token 预算上限
[ ] 工具权限最小化
[ ] 敏感操作需要人工确认
[ ] 记录所有 Agent 操作日志
[ ] 定期审查 Agent 行为

OPC 实战建议

从简单开始

第一个 Agent：单步 Agent，不使用工具
第二个 Agent：ReAct Agent，使用 1-2 个工具
第三个 Agent：ReAct Agent，使用 3-5 个工具
进阶：多 Agent 协作系统

成本控制建议

策略	说明	节省比例
模型选择	简单任务用 Haiku	80%
Prompt 优化	减少不必要的上下文	30%
缓存	相同查询缓存结果	50%
批处理	合并多个小任务	40%

下一步

理解 Agent 基础后，进入阶段 2：工具集成 — 学习如何为 Agent 注册更多工具，以及 MCP 协议的使用。

阶段 1：Agent 基础 ⭐ 必学 ​

1. 传统模式：痛点与瓶颈 ​

1.1 传统 AI 应用的角色定位 ​

1.2 沟通效率与协作成本 ​

1.3 量化痛点数据 ​

2. OPC 模式：重新定义 ​

2.1 核心理念 ​

2.2 人机分工矩阵 ​

2.3 效率对比 ​

3. 实操案例 ​

3.1 场景描述 ​

3.2 执行过程 ​

3.2.1 Agent 核心循环：ReAct 模式 ​

3.2.2 Tool Use 接口定义（Anthropic Claude） ​

3.2.3 Function Calling（OpenAI） ​

3.3 前后对比 ​

4. 趋势预判（未来 1-3 年） ​

4.1 技术演进方向 ​

4.2 角色变化趋势 ​

4.3 OPC 需要提前准备的能力 ​

5. 核心洞察 ​

6. 参考与延伸 ​

常见问题 ​

Agent 设计模式 ​

模式 1：单步 Agent ​

模式 2：ReAct Agent ​

模式 3：规划 Agent ​

模式 4：Plan-and-Execute Agent ​

模式 5：Reflexion Agent ​

模式 6：Tree-of-Thought Agent ​

模式 7：Self-Ask Agent ​

模式对比总结 ​

Agent 调试技巧 ​

使用 LangSmith 追踪 ​

调试检查清单 ​

错误处理与重试机制 ​

Agent 性能优化技巧 ​

Token 优化策略 ​

响应延迟优化 ​

Agent 安全最佳实践 ​

权限边界 ​

安全检查清单 ​

OPC 实战建议 ​

从简单开始 ​

成本控制建议 ​

下一步 ​

阶段 1：Agent 基础 ⭐ 必学

1. 传统模式：痛点与瓶颈

1.1 传统 AI 应用的角色定位

1.2 沟通效率与协作成本

1.3 量化痛点数据

2. OPC 模式：重新定义

2.1 核心理念

2.2 人机分工矩阵

2.3 效率对比

3. 实操案例

3.1 场景描述

3.2 执行过程

3.2.1 Agent 核心循环：ReAct 模式

3.2.2 Tool Use 接口定义（Anthropic Claude）

3.2.3 Function Calling（OpenAI）

3.3 前后对比

4. 趋势预判（未来 1-3 年）

4.1 技术演进方向

4.2 角色变化趋势

4.3 OPC 需要提前准备的能力

5. 核心洞察

6. 参考与延伸

常见问题

Agent 设计模式

模式 1：单步 Agent

模式 2：ReAct Agent

模式 3：规划 Agent

模式 4：Plan-and-Execute Agent

模式 5：Reflexion Agent

模式 6：Tree-of-Thought Agent

模式 7：Self-Ask Agent

模式对比总结

Agent 调试技巧

使用 LangSmith 追踪

调试检查清单

错误处理与重试机制

Agent 性能优化技巧

Token 优化策略

响应延迟优化

Agent 安全最佳实践

权限边界

安全检查清单

OPC 实战建议

从简单开始

成本控制建议

下一步