第2章-LLM工程基础-模型选型API调用与成本优化-《Agentic AI 智能体应用开发》
·
第2章：LLM 工程基础 —— 模型选型、API 调用与成本优化
本章导引
要构建 Agent，首先需要理解 Agent 的"大脑"——大语言模型（LLM）。本章从工程实践的角度，系统讲解 LLM 的选型、调用、优化和环境搭建。
很多 AI 工程师的第一个 Agent 都是从几行 API 调用开始的：构造一个 messages 数组，调用 chat.completions.create，然后获取返回的文本。但当你要构建一个生产级 Agent 时，问题就没这么简单了：
- 面对 GPT-5、Claude 4、Gemini 2.5、DeepSeek、Qwen 等十几个模型，应该选哪个？
- 流式输出的 SSE 协议如何处理？如何处理网络中断和重试？
- 如何管理并发请求？如何处理速率限制（Rate Limit）？
- Token 成本如何估算和控制？如何在不同模型间自动切换以优化成本？
- 如何安全地管理 API Key？如何使用本地模型降低成本和延迟？
本章将逐一回答这些问题，为构建生产级 Agent 打下坚实的 LLM 工程基础。
2.1 主流 LLM 模型体系
2.1.1 模型谱系全景
2026年的 LLM 市场已经形成了多层次、多梯队的竞争格局。理解这个格局，是做出正确选型决策的前提。
模型能力（推理 + 代码 + 多模态）
▲
│
┌──────────┼──────────┐
│ Tier 1: 顶级闭源 │
│ GPT-5, Claude 4 │
│ Gemini 2.5 Pro │
│ 综合能力最强，价格最高 │
│ │ │
│ ┌───────┼───────┐ │
│ │ Tier 2: 高性价比 │ │
│ │ Claude Sonnet │ │
│ │ GPT-4o, Qwen3 │ │
│ │ 能力接近T1，价格低 │ │
│ │ │ │ │
│ │ ┌────┼─────┐ │ │
│ │ │Tier 3: 开源 │ │ │
│ │ │DeepSeek-V3 │ │ │
│ │ │Llama 4 │ │ │
│ │ │Mistral │ │ │
│ │ │可自部署，成本最低│ │ │
│ │ └───────────┘ │ │
│ └─────────────────┘ │
└────────────────────────┘
Tier 1：顶级闭源模型
| 模型 | 厂商 | 核心优势 | 主要不足 | 适用场景 |
|---|---|---|---|---|
| Claude Opus 4.7 | Anthropic | 最深度的推理、最安全、200K上下文 | 价格最高 | 复杂Agent、安全敏感场景、长文本分析 |
| Claude Sonnet 4.6 | Anthropic | 推理/速度/价格的最佳平衡点 | 极端复杂推理弱于Opus | 日常Agent的主力模型 |
| GPT-5 | OpenAI | 代码生成能力顶尖 | 安全护栏相对弱 | 代码生成、技术问答 |
| Gemini 2.5 Pro | 多模态+1M上下文+搜索整合 | 推理一致性不如Claude | 多模态Agent、长文档处理 |
Tier 2：高性价比闭源
| 模型 | 厂商 | 核心优势 | 主要不足 | 适用场景 |
|---|---|---|---|---|
| GPT-4o | OpenAI | 多模态能力强、速度快 | 推理深度不如GPT-5 | 多模态、快速响应场景 |
| Qwen3-Max | 阿里 | 中文最强、性价比高 | 英文推理不如Claude/GPT | 中文场景、国内部署 |
| DeepSeek-V3.1 | DeepSeek | 极致性价比、代码能力强 | 推理深度有差距 | 大批量任务、代码生成 |
Tier 3：开源模型
| 模型 | 厂商 | 核心优势 | 主要不足 | 适用场景 |
|---|---|---|---|---|
| DeepSeek-V3 | DeepSeek | 开源最佳综合能力 | 需要GPU资源 | 私有化部署、数据安全 |
| Llama 4 | Meta | 开源生态最丰富 | 综合能力不如DeepSeek | 微调、定制化场景 |
| Qwen3 | 阿里 | 中文开源最佳 | 英文弱于Llama 4 | 中文私有化部署 |
| Mistral Large 3 | Mistral | 欧洲最佳、合规友好 | 中文能力弱 | 欧洲合规、多语言 |
2.1.2 模型能力维度详解
选型不能只看综合分数，需要从具体能力维度进行精细评估：
维度一：推理能力
推理能力是 Agent 的"智商"，直接影响 Agent 的决策质量。
评估推理能力的关键指标：
| 指标 | 含义 | 为什么重要 |
|---|---|---|
| 多步推理 | 能否正确推导包含多个逻辑步骤的问题 | Agent 需要规划多步任务 |
| 数学推理 | 能否正确解决数学和逻辑问题 | 数据分析、计算类工具需要 |
| 因果推理 | 能否理解事件间的因果关系 | 故障诊断、根因分析 |
| 反事实推理 | 能否考虑"如果…会怎样"的假设情况 | 风险评估、决策优化 |
推理能力的实际测试：
测试题：一个房间里有3盏灯和3个开关。开关在房间外。
你只能进房间一次。如何确定哪个开关控制哪盏灯？
正确推理过程：
1. 打开开关1，等待5分钟，然后关闭开关1
2. 打开开关2，立即进入房间
3. 亮的灯 → 开关2控制
4. 关着但是热的灯 → 开关1控制（曾被打开过一段时间）
5. 关着而且凉的灯 → 开关3控制（从未被打开）
Claude Opus 4.7: ✓ 正确推理，并解释热量残留原理
GPT-5: ✓ 正确推理
Claude Sonnet 4.6: ✓ 正确推理
Gemini 2.5 Pro: ✓ 正确推理，但解释稍简略
DeepSeek-V3: ✓ 正确推理（有时会在第一次尝试中遗漏热量细节）
Qwen3-Max: ✗ 有时会给出打开两个开关的方案
维度二：代码能力
代码能力对 DevAgent 类应用至关重要。评估维度包括：
- 代码生成准确性：生成的代码能否正确运行
- 代码风格一致性：是否符合项目的编码规范
- 调试能力：能否定位和修复 bug
- 架构理解：能否理解大型项目的结构和设计模式
代码能力的实际测试：
测试任务：实现一个支持并发控制、重试和超时的 API 请求客户端
Claude Opus 4.7:
- 完整的 TypeScript 实现，类型安全
- 包含并发队列、指数退避重试、请求超时
- 100+ 行，可直接用于生产
- 配有详细的注释和使用示例
Claude Sonnet 4.6:
- 实现质量接近 Opus
- 类型定义稍有简化但完整可用
- 约80行，适合快速集成
GPT-5:
- 代码质量与 Opus 相当
- 偏好使用更复杂的类型体操
- Python 版本更出色
DeepSeek-V3.1:
- 代码质量接近 Sonnet
- Python 实现出色
- TypeScript 泛型使用有时不够精准
维度三：多语言能力
对于中文场景的 Agent，中文理解和生成能力至关重要：
| 模型 | 中文理解 | 中文生成 | 中英混合 | 古文/诗词 |
|---|---|---|---|---|
| Qwen3-Max | 优秀 | 优秀 | 优秀 | 优秀 |
| Claude Opus 4.7 | 优秀 | 优秀 | 优秀 | 良好 |
| GPT-5 | 良好 | 良好 | 良好 | 一般 |
| DeepSeek-V3 | 优秀 | 良好 | 优秀 | 良好 |
| Gemini 2.5 Pro | 良好 | 良好 | 良好 | 一般 |
注意：中文"理解"和"生成"是不同的。有些模型能很好地理解中文，但生成的中文可能有"翻译腔"。Claude 4 和 Qwen3 在这方面表现最好。
维度四：多模态能力
如果 Agent 需要处理图片、文档、音视频，多模态能力就成为刚需：
| 模型 | 图片理解 | PDF解析 | 图表分析 | 视频理解 |
|---|---|---|---|---|
| Gemini 2.5 Pro | 优秀 | 优秀 | 优秀 | 支持 |
| GPT-5 / GPT-4o | 优秀 | 优秀 | 优秀 | 支持 |
| Claude Opus 4.7 | 优秀 | 良好 | 良好 | 不支持 |
| Qwen3-VL | 优秀 | 良好 | 优秀 | 支持 |
2.1.3 模型选型决策框架
选型不是做一道选择题，而是设计一个决策框架。本书推荐 任务-成本-延迟三维权衡（TCL 框架）：
任务复杂度 (Task Complexity)
▲
│
┌───────┼────────┐
│ 复杂任务 │
│ Opus 4.7 │
│ GPT-5 │
│ │
├─────────────────┤
│ 中等任务 │
│ Sonnet 4.6 │
│ DeepSeek-V3.1 │
│ Qwen3-Max │
│ │
├─────────────────┤
│ 简单任务 │
│ Haiku 4.5 │
│ GPT-4o-mini │
│ 本地小模型 │
└─────────────────┘
成本 (Cost) ◄─────────► 延迟要求 (Latency)
高成本可接受 低延迟要求
← →
选强模型 选快模型
选型决策树：
开始
│
├─ 是否需要私有化部署？
│ ├─ 是 → 评估 DeepSeek-V3 / Llama 4 / Qwen3
│ │ 进一步考虑：GPU资源？→ 是/否 → 调整模型大小
│ └─ 否 → 继续
│
├─ 是否需要多模态能力？
│ ├─ 是 → Gemini 2.5 Pro / GPT-4o / GPT-5
│ └─ 否 → 继续
│
├─ 任务复杂度？
│ ├─ 高（多步推理、代码生成、架构设计）
│ │ ├─ 成本敏感 → Claude Sonnet 4.6 / DeepSeek-V3.1
│ │ └─ 成本不敏感 → Claude Opus 4.7 / GPT-5
│ ├─ 中（摘要、翻译、简单分析）
│ │ ├─ 中文场景 → Qwen3-Max / DeepSeek
│ │ └─ 英文场景 → Claude Sonnet 4.6 / GPT-4o
│ └─ 低（分类、提取、格式化）
│ └─ Claude Haiku 4.5 / GPT-4o-mini / 本地小模型
│
├─ 安全合规要求？
│ ├─ 高 → Claude 4 系列（安全护栏最完善）
│ └─ 中 → GPT-5 / Gemini
│
└─ 推荐默认选择：Claude Sonnet 4.6
（综合能力、成本、安全的最佳平衡，适合95%的Agent场景）
实战建议：模型组合策略
不要只用一个大模型，而是根据任务特征动态路由：
// 模型路由器
class ModelRouter {
route(task: Task): Model {
// 简单分类/提取 → 快速便宜模型
if (task.type === 'classification' || task.type === 'extraction') {
return this.models.haiku;
}
// 代码生成 → GPT-5（代码能力最强）
if (task.type === 'code_generation') {
return this.models.gpt5;
}
// 中文任务 → Qwen3 或 DeepSeek
if (task.language === 'zh' && task.complexity !== 'high') {
return this.models.qwen3;
}
// 复杂推理 → Opus
if (task.complexity === 'high' && task.budget !== 'low') {
return this.models.opus;
}
// 默认 → Sonnet（最佳性价比）
return this.models.sonnet;
}
}
2.2 API 调用实践
2.2.1 OpenAI API / Anthropic API 深度使用指南
两种主流 API 虽然底层都是 HTTP + SSE，但在接口设计、参数体系、错误处理上有显著差异。理解这些差异，才能在构建 Agent 时写出健壮的模型调用代码。
OpenAI API 核心调用
# Python: OpenAI API 完整调用示例
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key="sk-...",
max_retries=3,
timeout=120.0,
)
async def call_openai(prompt: str) -> str:
response = await client.chat.completions.create(
model="gpt-5",
messages=[
{
"role": "system",
"content": "你是一个专业的 Python 工程师。"
},
{
"role": "user",
"content": prompt
}
],
temperature=0.7, # 创造性控制（0=确定性，1=创造性）
max_tokens=4096, # 最大输出 token 数
top_p=0.95, # nucleus sampling
frequency_penalty=0.0, # 降低词语重复
presence_penalty=0.0, # 鼓励话题多样性
stop=["\n\n\n"], # 停止序列
)
return response.choices[0].message.content
// TypeScript: OpenAI API 完整调用示例
import OpenAI from 'openai';
const client = new OpenAI({
apiKey: process.env.OPENAI_API_KEY,
maxRetries: 3,
timeout: 120 * 1000,
});
async function callOpenAI(prompt: string): Promise<string> {
const response = await client.chat.completions.create({
model: 'gpt-5',
messages: [
{ role: 'system', content: '你是一个专业的 TypeScript 工程师。' },
{ role: 'user', content: prompt },
],
temperature: 0.7,
max_tokens: 4096,
top_p: 0.95,
frequency_penalty: 0,
presence_penalty: 0,
});
return response.choices[0]?.message?.content ?? '';
}
Anthropic API 核心调用
# Python: Anthropic API 完整调用示例
from anthropic import AsyncAnthropic
client = AsyncAnthropic(
api_key="sk-ant-...",
max_retries=3,
timeout=120.0,
)
async def call_claude(prompt: str) -> str:
response = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
system="你是一个专业的 Python 工程师。",
messages=[
{"role": "user", "content": prompt}
],
temperature=0.7,
# Anthropic 特有参数
# top_p 默认 1.0, 不像 OpenAI 那样常用
)
# Anthropic 的 content 是 ContentBlock 数组
for block in response.content:
if block.type == "text":
return block.text
return ""
// TypeScript: Anthropic API 完整调用示例
import { Anthropic } from '@anthropic-ai/sdk';
const client = new Anthropic({
apiKey: process.env.ANTHROPIC_API_KEY,
maxRetries: 3,
timeout: 120 * 1000,
});
async function callClaude(prompt: string): Promise<string> {
const response = await client.messages.create({
model: 'claude-sonnet-4-6',
max_tokens: 4096,
system: '你是一个专业的 TypeScript 工程师。',
messages: [
{ role: 'user', content: prompt },
],
temperature: 0.7,
});
const textBlock = response.content.find(b => b.type === 'text');
return textBlock?.text ?? '';
}
两种 API 的关键差异
| 差异点 | OpenAI | Anthropic |
|---|---|---|
| 端点 | /chat/completions |
/messages |
| System Prompt | 在 messages 数组中（role=‘system’） | 独立的 system 参数 |
| 返回结构 | response.choices[0].message.content（字符串或 None） |
response.content（ContentBlock 数组） |
| 图片格式 | image_url 类型 content part |
source 类型 content block（base64） |
| Temperature | 默认 1.0 | 默认 1.0 |
| max_tokens | 默认无限 | 必须显式设置 |
| Stop sequences | 支持数组 | 支持字符串（需转为 Anthropic 格式） |
重要提醒：Anthropic API 的 max_tokens 是必填参数，如果不设置会报错。而 OpenAI 的 max_tokens 是可选的。
2.2.2 流式输出处理：SSE 与 Stream 协议实战
流式输出（Streaming）是构建好的 Agent 交互体验的关键。用户不想等 20 秒看一个完整的回复——他们希望像打字一样看到逐字输出。
Server-Sent Events (SSE) 协议基础
LLM API 的流式输出基于 SSE（Server-Sent Events）协议：
HTTP Response:
Content-Type: text/event-stream
Transfer-Encoding: chunked
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"你"}}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"好"}}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"！"}}]}
data: [DONE]
OpenAI 流式处理
# Python: OpenAI 流式输出
async def stream_openai(prompt: str):
stream = await client.chat.completions.create(
model="gpt-5",
messages=[{"role": "user", "content": prompt}],
stream=True, # 开启流式
stream_options={"include_usage": True}, # 在最后输出 token 用量
)
full_text = []
async for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_text.append(content)
print(content, end="", flush=True) # 实时输出
# 最后一个 chunk 包含 usage 信息
if chunk.usage:
print(f"\n[Token用量: {chunk.usage.total_tokens}]")
return "".join(full_text)
// TypeScript: OpenAI 流式输出
async function streamOpenAI(prompt: string): Promise<string> {
const stream = await client.chat.completions.create({
model: 'gpt-5',
messages: [{ role: 'user', content: prompt }],
stream: true,
stream_options: { include_usage: true },
});
const chunks: string[] = [];
for await (const chunk of stream) {
const content = chunk.choices[0]?.delta?.content;
if (content) {
chunks.push(content);
process.stdout.write(content); // 实时输出
}
if (chunk.usage) {
console.log(`\n[Token: ${chunk.usage.total_tokens}]`);
}
}
return chunks.join('');
}
Anthropic 流式处理
Anthropic 的流式事件类型更丰富，包括 message_start、content_block_start、content_block_delta、message_delta、message_stop 等：
# Python: Anthropic 流式输出
async def stream_claude(prompt: str):
full_text = []
async with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=4096,
system="你是一个有帮助的助手。",
messages=[{"role": "user", "content": prompt}],
) as stream:
async for event in stream:
if event.type == "text":
full_text.append(event.text)
print(event.text, end="", flush=True)
elif event.type == "content_block_start":
# 新内容块开始（可能是 text 或 tool_use）
pass
elif event.type == "content_block_stop":
# 内容块结束
pass
elif event.type == "message_start":
# 消息开始，包含 input_tokens
pass
elif event.type == "message_delta":
# 消息结束，包含 output_tokens
print(f"\n[Token: 输入{event.usage.input_tokens} + 输出{event.usage.output_tokens}]")
return "".join(full_text)
// TypeScript: Anthropic 流式输出
async function streamClaude(prompt: string): Promise<string> {
const stream = client.messages.stream({
model: 'claude-sonnet-4-6',
max_tokens: 4096,
system: '你是一个有帮助的助手。',
messages: [{ role: 'user', content: prompt }],
});
const chunks: string[] = [];
stream.on('text', (text) => {
chunks.push(text);
process.stdout.write(text);
});
stream.on('contentBlockStart', (block) => {
// 处理 content block 开始事件
});
const message = await stream.finalMessage();
console.log(`\n[Token: 输入${message.usage.input_tokens} + 输出${message.usage.output_tokens}]`);
return chunks.join('');
}
Agent 场景的流式输出最佳实践
在 Agent 场景中，流式处理需要特别注意以下几点：
// 流式输出的生产级封装
class StreamingModelClient {
private buffer = '';
private bufferThreshold = 200; // 200字符阈值
async *streamWithBuffer(model: string, messages: Message[]): AsyncGenerator<StreamEvent> {
const stream = await this.client.messages.stream({
model,
max_tokens: 4096,
messages,
});
for await (const event of stream) {
if (event.type === 'text') {
this.buffer += event.text;
// 达到阈值时发送批量事件，减少 IPC 开销
if (this.buffer.length >= this.bufferThreshold) {
yield { type: 'text_delta', text: this.buffer };
this.buffer = '';
}
} else if (event.type === 'tool_use_start') {
// 工具调用开始时先 flush buffer
if (this.buffer.length > 0) {
yield { type: 'text_delta', text: this.buffer };
this.buffer = '';
}
yield { type: 'tool_use_start', tool: event.name };
} else if (event.type === 'tool_use_end') {
yield { type: 'tool_use_end', tool: event.name, result: event.result };
}
}
// 最后 flush 剩余 buffer
if (this.buffer.length > 0) {
yield { type: 'text_delta', text: this.buffer };
this.buffer = '';
}
yield { type: 'done' };
}
}
2.2.3 错误处理与重试策略
LLM API 调用可能因各种原因失败，健壮的错误处理是生产级 Agent 的基本要求。
错误类型全景
| 错误类型 | HTTP 状态码 | 原因 | 处理策略 |
|---|---|---|---|
| 速率限制 | 429 | 请求太频繁 | 指数退避重试 |
| 服务过载 | 529 / 503 | 服务端压力大 | 指数退避重试 |
| 认证失败 | 401 | API Key 无效或过期 | 不重试，立即告警 |
| 权限不足 | 403 | 无权限访问该资源 | 不重试，检查权限 |
| 上下文超限 | 400 | 输入 token 超过模型限制 | 压缩上下文后重试 |
| 内容过滤 | 400 | 触发了内容安全策略 | 不重试，审查输入内容 |
| 网络超时 | — | 请求超时 | 重试，缩短超时 |
| 服务端错误 | 500 | API 内部错误 | 指数退避重试 |
| 流中断 | — | 流式输出中途断开 | 检查已接收内容，尝试重新请求 |
指数退避重试实现
// TypeScript: 指数退避重试器
class RetryManager {
private config: RetryConfig;
constructor(config: Partial<RetryConfig> = {}) {
this.config = {
maxRetries: config.maxRetries ?? 5,
baseDelayMs: config.baseDelayMs ?? 1000,
maxDelayMs: config.maxDelayMs ?? 60000,
jitter: config.jitter ?? true,
retryableStatuses: config.retryableStatuses ?? [429, 503, 529],
};
}
async execute<T>(
fn: () => Promise<T>,
context: string = 'API Call'
): Promise<T> {
let lastError: Error | null = null;
for (let attempt = 0; attempt <= this.config.maxRetries; attempt++) {
try {
return await fn();
} catch (error) {
lastError = error as Error;
// 判断是否可重试
if (!this.isRetryable(error) || attempt === this.config.maxRetries) {
throw this.enrichError(error, context, attempt);
}
// 计算延迟
const delay = this.calculateDelay(attempt);
console.warn(
`[${context}] 第 ${attempt + 1} 次尝试失败，${delay}ms 后重试: ${(error as any).message}`
);
await this.sleep(delay);
}
}
throw lastError;
}
private isRetryable(error: any): boolean {
// 网络错误总是可重试的
if (error.code === 'ECONNRESET' || error.code === 'ETIMEDOUT' ||
error.code === 'ECONNREFUSED') {
return true;
}
// 检查 HTTP 状态码
const status = error.status || error.statusCode;
if (status && this.config.retryableStatuses.includes(status)) {
return true;
}
// Anthropic 特有的 overloaded 错误
if (error.type === 'overloaded_error') {
return true;
}
return false;
}
private calculateDelay(attempt: number): number {
// 指数退避: baseDelay * 2^attempt
let delay = this.config.baseDelayMs * Math.pow(2, attempt);
// 上限
delay = Math.min(delay, this.config.maxDelayMs);
// 随机抖动（避免惊群效应）
if (this.config.jitter) {
delay = delay * (0.5 + Math.random() * 0.5);
}
return Math.round(delay);
}
private sleep(ms: number): Promise<void> {
return new Promise(resolve => setTimeout(resolve, ms));
}
private enrichError(error: any, context: string, attempts: number): Error {
const msg = `[${context}] 重试 ${attempts} 次后仍然失败: ${error.message || error}`;
const enriched = new Error(msg);
enriched.cause = error;
return enriched;
}
}
// 使用示例
const retryManager = new RetryManager({
maxRetries: 3,
baseDelayMs: 1000,
});
const result = await retryManager.execute(
() => client.messages.create({ model: 'claude-sonnet-4-6', max_tokens: 1024, messages: [...] }),
'Claude API'
);
# Python: 指数退避重试器
import asyncio
import random
from typing import Callable, Awaitable, TypeVar
T = TypeVar('T')
class RetryManager:
def __init__(
self,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
jitter: bool = True,
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.jitter = jitter
self.retryable_statuses = {429, 503, 529}
async def execute(
self, fn: Callable[[], Awaitable[T]], context: str = "API Call"
) -> T:
last_error = None
for attempt in range(self.max_retries + 1):
try:
return await fn()
except Exception as e:
last_error = e
if not self._is_retryable(e) or attempt == self.max_retries:
raise self._enrich_error(e, context, attempt)
delay = self._calculate_delay(attempt)
print(f"[{context}] 第 {attempt + 1} 次失败，{delay:.0f}ms 后重试: {e}")
await asyncio.sleep(delay / 1000)
raise last_error
def _is_retryable(self, error: Exception) -> bool:
status = getattr(error, 'status_code', None) or getattr(error, 'status', None)
if status in self.retryable_statuses:
return True
error_type = getattr(error, 'type', None)
if error_type == 'overloaded_error':
return True
return False
def _calculate_delay(self, attempt: int) -> float:
delay = self.base_delay * (2 ** attempt) * 1000 # 转换为毫秒
delay = min(delay, self.max_delay * 1000)
if self.jitter:
delay *= (0.5 + random.random() * 0.5)
return delay
熔断器模式
当后端持续失败时，持续的请求只会加剧问题。熔断器（Circuit Breaker）模式可以在检测到高失败率时暂时停止发送请求：
// TypeScript: 熔断器
class CircuitBreaker {
private state: 'closed' | 'open' | 'half-open' = 'closed';
private failureCount = 0;
private lastFailureTime = 0;
private successCount = 0;
constructor(
private config: {
failureThreshold: number; // 连续失败 N 次后熔断
recoveryTimeoutMs: number; // 熔断后多久尝试恢复
halfOpenMaxSuccess: number; // 半开状态成功 N 次后关闭熔断
}
) {}
async execute<T>(fn: () => Promise<T>): Promise<T> {
if (this.state === 'open') {
if (Date.now() - this.lastFailureTime > this.config.recoveryTimeoutMs) {
this.state = 'half-open';
this.successCount = 0;
console.log('[熔断器] 进入半开状态，尝试恢复');
} else {
throw new Error('熔断器开启，拒绝请求');
}
}
try {
const result = await fn();
this.onSuccess();
return result;
} catch (error) {
this.onFailure();
throw error;
}
}
private onSuccess(): void {
this.failureCount = 0;
if (this.state === 'half-open') {
this.successCount++;
if (this.successCount >= this.config.halfOpenMaxSuccess) {
this.state = 'closed';
console.log('[熔断器] 恢复关闭');
}
}
}
private onFailure(): void {
this.failureCount++;
this.lastFailureTime = Date.now();
if (this.failureCount >= this.config.failureThreshold) {
this.state = 'open';
console.warn('[熔断器] 熔断开启！连续失败:', this.failureCount);
}
}
}
2.2.4 并发控制与速率限制管理
生产环境中，Agent 系统通常需要并发处理多个用户请求。不当的并发管理会导致 429 错误（速率限制）或 API 额度快速耗尽。
并发控制策略
// TypeScript: 并发请求管理器
class ConcurrencyManager {
private queue: Array<() => Promise<void>> = [];
private running = 0;
constructor(
private maxConcurrent: number = 10,
private maxRequestsPerMinute: number = 1000,
) {
// 每分钟重置计数器
setInterval(() => this.requestCount = 0, 60000);
}
private requestCount = 0;
async execute<T>(fn: () => Promise<T>): Promise<T> {
// 速率限制检查
if (this.requestCount >= this.maxRequestsPerMinute) {
await this.waitForRateLimit();
}
// 并发控制
while (this.running >= this.maxConcurrent) {
await new Promise<void>(resolve => this.queue.push(resolve as any));
}
this.running++;
this.requestCount++;
try {
return await fn();
} finally {
this.running--;
this.dequeue();
}
}
private dequeue(): void {
const next = this.queue.shift();
if (next) next();
}
private async waitForRateLimit(): Promise<void> {
const backoffMs = Math.min(
(this.requestCount - this.maxRequestsPerMinute + 1) * 100,
30000
);
await new Promise(resolve => setTimeout(resolve, backoffMs));
}
}
# Python: 并发请求管理器
import asyncio
import time
class ConcurrencyManager:
def __init__(
self,
max_concurrent: int = 10,
max_per_minute: int = 1000,
):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.max_per_minute = max_per_minute
self.request_times: list[float] = []
async def execute(self, fn):
# 速率限制
await self._wait_for_rate_limit()
# 并发控制
async with self.semaphore:
self.request_times.append(time.time())
return await fn()
async def _wait_for_rate_limit(self):
# 清理旧的请求记录
now = time.time()
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) >= self.max_per_minute:
# 等到最早的一个请求过期
wait = 60 - (now - self.request_times[0]) + 0.1
if wait > 0:
await asyncio.sleep(wait)
2.3 Token 与成本管理
2.3.1 Token 计数与估算
Token 是 LLM API 计费的基本单位。理解 Token 的计数方式，是成本管理的基础。
Token 是什么？
Token 是模型处理文本的最小单位。英文中大约 1 token ≈ 0.75 个单词，中文中大约 1 token ≈ 0.5 个汉字。
| 文本 | 估计 Token 数 |
|---|---|
| “Hello World” | 2 tokens |
| “你好世界” | 3 tokens |
| “Artificial Intelligence” | 3 tokens |
| “人工智能” | 2 tokens |
| 一页英文 (500 words) | ~667 tokens |
| 一页中文 (800字) | ~400 tokens |
使用 tiktoken（OpenAI 的 tokenizer）可以精确计算：
# Python: Token 精确计数
import tiktoken
# OpenAI 模型
encoder = tiktoken.encoding_for_model("gpt-4")
tokens = encoder.encode("你好世界！Hello World!")
print(f"Token 数: {len(tokens)}") # 输出: Token 数: 9
print(f"Token IDs: {tokens}")
# Claude 模型（使用同等编码器估算）
# Anthropic 没有公开 tokenizer，但可以用 cl100k_base 近似估算
encoder = tiktoken.get_encoding("cl100k_base")
tokens = encoder.encode("这是一个测试段落。")
print(f"估计 Token 数: {len(tokens)}")
// TypeScript: Token 估算
// tiktoken 在 Node.js 中的等价实现
import { encoding_for_model, get_encoding } from 'tiktoken';
// OpenAI 模型精确计数
const encoder = encoding_for_model('gpt-4');
const tokens = encoder.encode('你好世界！Hello World!');
console.log(`Token 数: ${tokens.length}`);
// 释放 encoder（重要！避免内存泄漏）
encoder.free();
// 通用估算（不需要 tiktoken）
function estimateTokens(text: string): number {
// 中文字符约 0.5 token/字，英文约 0.25 token/字符
const chineseChars = (text.match(/[一-鿿]/g) || []).length;
const otherChars = text.length - chineseChars;
return Math.ceil(chineseChars * 0.5 + otherChars * 0.25);
}
// 更精确的估算（考虑了中英文混合）
function estimateTokensV2(text: string): number {
// 中文：约 1.5-2 字符/token，取 1.8
// 英文：约 4 字符/token
// 使用启发式方法
const chinesePattern = /[一-鿿 -〿-]/g;
const chineseMatch = text.match(chinesePattern);
const chineseChars = chineseMatch ? chineseMatch.length : 0;
// 非中文字符（包括英文、数字、标点等）
const nonChinese = text.length - chineseChars;
return Math.ceil(chineseChars / 1.8 + nonChinese / 4);
}
2.3.2 成本优化策略
LLM API 的成本可以非常可观。以一个处理日均 10 万次请求的 Agent 系统为例：
| 策略 | 日均成本 | 节省 |
|---|---|---|
| 全用 Opus 4.7 | $2000 | — |
| 全用 Sonnet 4.6 | $500 | 75% |
| 全用 Haiku 4.5 | $50 | 97.5% |
| 智能路由 | $300 | 85% |
策略一：请求缓存（Semantic Cache）
对于相似的请求，缓存之前的回复可以大幅降低成本：
# Python: 语义缓存
import hashlib
import json
from functools import lru_cache
class SemanticCache:
def __init__(self, redis_client=None, ttl: int = 3600):
self.cache = {} if redis_client is None else None
self.redis = redis_client
self.ttl = ttl
def _get_key(self, messages: list) -> str:
"""对 messages 做规范化哈希作为缓存键"""
# 移除无关差异（多余空格、大小写等）
normalized = json.dumps(messages, sort_keys=True, ensure_ascii=False)
normalized = ' '.join(normalized.lower().split())
return hashlib.sha256(normalized.encode()).hexdigest()[:16]
async def get(self, messages: list) -> str | None:
key = self._get_key(messages)
if self.redis:
return await self.redis.get(key)
return self.cache.get(key)
async def set(self, messages: list, response: str):
key = self._get_key(messages)
if self.redis:
await self.redis.setex(key, self.ttl, response)
else:
self.cache[key] = response
# 使用
cache = SemanticCache()
async def cached_call(messages):
cached = await cache.get(messages)
if cached:
return cached
response = await client.messages.create(...)
await cache.set(messages, response)
return response
策略二：Prompt 压缩
减少 Prompt 中的重复和冗余信息：
# Python: 对话历史压缩
def compress_conversation(messages: list, max_tokens: int = 8000) -> list:
"""压缩对话历史到指定 token 数以内"""
# 1. System Prompt 不可压缩
system = [m for m in messages if m["role"] == "system"]
others = [m for m in messages if m["role"] != "system"]
# 2. 保留最近 N 条完整消息
recent_count = 4
recent = others[-recent_count:]
rest = others[:-recent_count]
# 3. 对较旧的消息做摘要压缩
if rest:
summary = "【历史对话摘要】" + "；".join(
f"{'用户' if m['role']=='user' else '助手'}: {str(m['content'])[:100]}"
for m in rest
)
rest_compressed = [{"role": "user", "content": summary}]
else:
rest_compressed = []
return system + rest_compressed + recent
策略三：多模型路由
根据任务难度动态选择模型，是性价比最高的优化策略：
// TypeScript: 智能模型路由器（增强版）
interface ModelRoute {
model: string;
costPer1kInput: number; // 输入 $/1K tokens
costPer1kOutput: number; // 输出 $/1K tokens
maxTokens: number;
latency: 'fast' | 'medium' | 'slow';
}
class SmartModelRouter {
private routes: Record<string, ModelRoute> = {
'opus-4.7': { model: 'claude-opus-4-7', costPer1kInput: 0.015, costPer1kOutput: 0.075, maxTokens: 200000, latency: 'slow' },
'sonnet-4.6': { model: 'claude-sonnet-4-6', costPer1kInput: 0.003, costPer1kOutput: 0.015, maxTokens: 200000, latency: 'medium' },
'haiku-4.5': { model: 'claude-haiku-4-5', costPer1kInput: 0.0008, costPer1kOutput: 0.004, maxTokens: 200000, latency: 'fast' },
};
async route(task: Task): Promise<ModelRoute> {
// 简单分类/格式化 → 最便宜的
if (task.complexity === 'low') {
return this.routes['haiku-4.5'];
}
// 代码生成 → Sonnet 代码能力足够
if (task.type === 'code') {
return this.routes['sonnet-4.6'];
}
// 需要深度推理 → Opus
if (task.requiresDeepReasoning) {
return this.routes['opus-4.7'];
}
// 默认 → Sonnet
return this.routes['sonnet-4.6'];
}
// 预估某次调用的成本
estimateCost(route: ModelRoute, inputTokens: number, estimatedOutputTokens: number): number {
const inputCost = (inputTokens / 1000) * route.costPer1kInput;
const outputCost = (estimatedOutputTokens / 1000) * route.costPer1kOutput;
return inputCost + outputCost;
}
}
2.3.3 多模型 Fallback 链设计
当主模型不可用时，自动降级到备用模型：
// TypeScript: 多模型 Fallback 链
class ModelFallbackChain {
private chain: ModelConfig[];
constructor(models: ModelConfig[]) {
this.chain = models; // 按优先级排列
}
async call(
messages: Message[],
options: CallOptions
): Promise<ModelResponse> {
const errors: Error[] = [];
for (const modelConfig of this.chain) {
try {
console.log(`尝试模型: ${modelConfig.name}`);
const response = await this.callModel(modelConfig, messages, options);
// 记录使用了哪个模型（用于成本追踪）
response.actualModel = modelConfig.name;
response.fallbackCount = errors.length;
return response;
} catch (error) {
errors.push(error as Error);
console.warn(`模型 ${modelConfig.name} 失败:`, (error as Error).message);
// 不可重试的错误（如认证失败）直接失败，不尝试下一个
if (this.isNonRetryable(error as Error)) {
break;
}
}
}
throw new Error(
`所有模型调用失败 (共${this.chain.length}个):\n` +
errors.map((e, i) => ` ${i+1}. ${this.chain[i].name}: ${e.message}`).join('\n')
);
}
private async callModel(config: ModelConfig, messages: Message[], options: CallOptions): Promise<ModelResponse> {
const client = this.getClient(config.provider);
// 根据 provider 选择正确的 API 调用方式
switch (config.provider) {
case 'anthropic':
return this.callAnthropic(client, config.model, messages, options);
case 'openai':
return this.callOpenAI(client, config.model, messages, options);
case 'deepseek':
// DeepSeek 兼容 OpenAI API 格式
return this.callOpenAI(client, config.model, messages, options);
default:
throw new Error(`不支持的 provider: ${config.provider}`);
}
}
private isNonRetryable(error: Error): boolean {
const status = (error as any).status;
return status === 401 || status === 403;
}
}
// 配置 Fallback 链：Opus → Sonnet → GPT → DeepSeek
const fallbackChain = new ModelFallbackChain([
{ name: 'claude-opus', provider: 'anthropic', model: 'claude-opus-4-7' },
{ name: 'claude-sonnet', provider: 'anthropic', model: 'claude-sonnet-4-6' },
{ name: 'gpt-5', provider: 'openai', model: 'gpt-5' },
{ name: 'deepseek', provider: 'deepseek', model: 'deepseek-chat' },
]);
2.3.4 实践案例：构建成本感知的模型调度器
将上述所有策略整合为一个完整的、带成本感知的模型调度器：
# Python: 成本感知模型调度器
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
@dataclass
class UsageTracker:
"""追踪每次 API 调用的成本"""
daily_cost: float = 0.0
monthly_cost: float = 0.0
calls: list = field(default_factory=list)
def record(self, model: str, input_tokens: int, output_tokens: int, cost: float):
self.daily_cost += cost
self.monthly_cost += cost
self.calls.append({
'time': datetime.now().isoformat(),
'model': model,
'input_tokens': input_tokens,
'output_tokens': output_tokens,
'cost': cost,
})
def should_downgrade(self, daily_budget: float) -> bool:
return self.daily_cost > daily_budget * 0.8 # 达到80%预算时降级
class CostAwareScheduler:
def __init__(
self,
daily_budget: float = 100.0, # 每日预算 $100
monthly_budget: float = 2000.0,
):
self.daily_budget = daily_budget
self.monthly_budget = monthly_budget
self.tracker = UsageTracker()
self.model_costs = {
'opus-4.7': {'input': 0.015, 'output': 0.075},
'sonnet-4.6': {'input': 0.003, 'output': 0.015},
'haiku-4.5': {'input': 0.0008, 'output': 0.004},
}
async def call(self, messages: list, task_complexity: str = 'medium') -> str:
# 选择模型
model = self._select_model(task_complexity)
# 预估成本
est_input = sum(len(str(m.get('content', ''))) for m in messages) // 4
est_cost = self._estimate_cost(model, est_input, 1000)
print(f"[调度器] 选择模型: {model}, 预估成本: ${est_cost:.4f}")
# 调用模型
response = await client.messages.create(
model=model,
max_tokens=4096,
messages=messages,
)
# 记录成本
actual_cost = self._calculate_actual_cost(
model,
response.usage.input_tokens,
response.usage.output_tokens,
)
self.tracker.record(model, response.usage.input_tokens,
response.usage.output_tokens, actual_cost)
print(f"[调度器] 实际成本: ${actual_cost:.4f}, 今日累计: ${self.tracker.daily_cost:.2f}")
return response.content[0].text
def _select_model(self, complexity: str) -> str:
# 预算紧张时降级
if self.tracker.should_downgrade(self.daily_budget):
print("[调度器] ⚠️ 达到日预算80%，降级到 Haiku")
return 'haiku-4.5'
# 按复杂度选择
match complexity:
case 'high':
return 'opus-4.7'
case 'medium':
return 'sonnet-4.6'
case 'low':
return 'haiku-4.5'
return 'sonnet-4.6'
def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
costs = self.model_costs[model]
return (input_tokens / 1000) * costs['input'] + (output_tokens / 1000) * costs['output']
def _calculate_actual_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
costs = self.model_costs[model]
return (input_tokens / 1000) * costs['input'] + (output_tokens / 1000) * costs['output']
2.4 开发环境搭建
2.4.1 项目初始化
本节搭建本书后续所有代码的基础开发环境。
TypeScript 项目初始化
# 创建项目目录
mkdir agent-book && cd agent-book
# 初始化 monorepo
pnpm init
# 创建包结构
mkdir -p packages/core/src
mkdir -p packages/core/tests
mkdir -p packages/tools/src
mkdir -p packages/memory/src
mkdir -p apps/devassistant
# 安装核心依赖
cd packages/core
pnpm add @anthropic-ai/sdk openai tiktoken
pnpm add -D typescript @types/node vitest
# tsconfig.json
cat > tsconfig.json << 'EOF'
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": ["ES2022"],
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist", "tests"]
}
EOF
Python 项目初始化
# 创建 Python 实验环境
mkdir -p python-lab
cd python-lab
# 使用 uv 管理依赖（推荐，比 pip 快 10-100 倍）
uv init
uv add anthropic openai tiktoken python-dotenv chromadb
# 创建环境变量文件
cat > .env.example << 'EOF'
ANTHROPIC_API_KEY=sk-ant-xxx
OPENAI_API_KEY=sk-xxx
DEEPSEEK_API_KEY=sk-xxx
EOF
# 创建 main.py
cat > main.py << 'EOF'
"""Agentic AI 应用开发 - Python 实验入口"""
import asyncio
import os
from dotenv import load_dotenv
load_dotenv()
async def main():
print("Agentic AI 开发环境已就绪")
print(f"Python 版本: {__import__('sys').version}")
print(f"Anthropic SDK: {__import__('anthropic').__version__}")
print(f"OpenAI SDK: {__import__('openai').__version__}")
if __name__ == "__main__":
asyncio.run(main())
EOF
2.4.2 API Key 管理与安全最佳实践
API Key 泄露是最常见的安全事故之一。以下是最佳实践：
// TypeScript: API Key 安全管理
class APIKeyManager {
private keys: Map<string, string> = new Map();
constructor() {
// 优先级1: 环境变量
this.loadFromEnv('ANTHROPIC_API_KEY');
this.loadFromEnv('OPENAI_API_KEY');
this.loadFromEnv('DEEPSEEK_API_KEY');
// 优先级2: 加密配置文件
this.loadFromEncryptedConfig();
// 安全检查：启动时验证 Key 格式
this.validateKeys();
}
private loadFromEnv(key: string): void {
const value = process.env[key];
if (value) {
// 不记录 Key 内容到日志！
console.log(`[KeyManager] 从环境变量加载: ${key} (长度=${value.length})`);
this.keys.set(key, value);
}
}
getKey(provider: 'anthropic' | 'openai' | 'deepseek'): string {
const keyMap: Record<string, string> = {
anthropic: 'ANTHROPIC_API_KEY',
openai: 'OPENAI_API_KEY',
deepseek: 'DEEPSEEK_API_KEY',
};
const key = this.keys.get(keyMap[provider]);
if (!key) {
throw new Error(
`未找到 ${provider} 的 API Key。请设置环境变量 ${keyMap[provider]}`
);
}
return key;
}
private validateKeys(): void {
// Anthropic Key 以 sk-ant- 开头
const anthropicKey = this.keys.get('ANTHROPIC_API_KEY');
if (anthropicKey && !anthropicKey.startsWith('sk-ant-')) {
throw new Error('ANTHROPIC_API_KEY 格式异常，应以 sk-ant- 开头');
}
// OpenAI Key 以 sk- 开头
const openaiKey = this.keys.get('OPENAI_API_KEY');
if (openaiKey && !openaiKey.startsWith('sk-')) {
throw new Error('OPENAI_API_KEY 格式异常，应以 sk- 开头');
}
}
private loadFromEncryptedConfig(): void {
// 生产环境中，应使用 AES 加密存储敏感的 API Key
// 详见第7章安全性实战
}
// 脱敏显示（用于日志/调试）
maskKey(key: string): string {
if (key.length <= 8) return '****';
return key.slice(0, 4) + '****' + key.slice(-4);
}
}
API Key 安全清单：
- 永远不要将 API Key 硬编码在源代码中
- 永远不要将 API Key 提交到 Git（使用
.gitignore排除.env）
# .gitignore
.env
.env.local
.env.*.local
*.key
credentials.json
- 使用环境变量或加密的配置文件存储 Key
- 在 CI/CD 中使用密钥管理服务（如 GitHub Secrets、Vault）
- 定期轮换 API Key
- 为不同环境（开发/测试/生产）使用不同的 API Key
- 设置 API 用量限制和告警
2.4.3 本地模型部署：Ollama + vLLM 快速上手
对于数据安全敏感的场景，本地模型部署是必要选项。
Ollama（最简单）
# 安装 Ollama
curl -fsSL https://ollama.com/install.sh | sh
# 拉取模型
ollama pull qwen3:14b # 阿里 Qwen3 14B，中文优秀
ollama pull deepseek-r1:14b # DeepSeek R1 14B，推理能力强
ollama pull llama3.2:3b # Meta Llama 3B，适合资源受限
# 启动 API 服务
ollama serve
# 测试调用
curl http://localhost:11434/api/generate -d '{
"model": "qwen3:14b",
"prompt": "解释什么是 Agentic AI",
"stream": false
}'
# Python: 使用 Ollama API（兼容 OpenAI SDK）
from openai import AsyncOpenAI
ollama_client = AsyncOpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama", # Ollama 不需要真实 key
)
async def call_ollama(prompt: str) -> str:
response = await ollama_client.chat.completions.create(
model="qwen3:14b",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
)
return response.choices[0].message.content
vLLM（高性能生产级）
# 安装 vLLM
pip install vllm
# 启动兼容 OpenAI API 的服务
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen3-14B \
--tensor-parallel-size 2 \
--max-model-len 32768 \
--port 8000
# Python: 调用 vLLM 服务
vllm_client = AsyncOpenAI(
base_url="http://localhost:8000/v1",
api_key="not-needed",
)
async def call_vllm(prompt: str) -> str:
response = await vllm_client.chat.completions.create(
model="Qwen/Qwen3-14B",
messages=[{"role": "user", "content": prompt}],
max_tokens=2048,
)
return response.choices[0].message.content
本地模型 vs 云端 API 对比
| 维度 | 本地模型 (Ollama/vLLM) | 云端 API |
|---|---|---|
| 数据安全 | 数据不离开服务器 | 数据发送到第三方 |
| 延迟 | 取决于硬件，可优化 | 网络延迟 + 推理时间 |
| 成本 | GPU 硬件成本高，边际成本低 | 按 Token 付费，大规模时成本高 |
| 模型能力 | 受限（通常 < 100B 参数） | 最强（可访问前沿模型） |
| 运维 | 需要自行管理 | 零运维 |
| 推荐场景 | 数据安全敏感、大批量处理 | 需要最强推理能力、快速原型 |
推荐策略：大部分场景使用 Claude Sonnet 4.6（云端 API），同时部署本地 Qwen3/DeepSeek 作为降级方案和敏感数据处理方案。
2.5 本章小结
本章系统讲解了 LLM 工程基础的四大核心领域：
-
模型选型：2026年 LLM 市场呈现 Tier 1（顶级闭源）、Tier 2（高性价比）、Tier 3（开源）三层格局。选型应从任务复杂度、成本预算、延迟要求三维度综合决策。推荐默认选择 Claude Sonnet 4.6，复杂推理时升级 Opus，简单任务降级 Haiku。
-
API 调用：OpenAI 和 Anthropic API 在接口设计上有显著差异，需要在工程层做适配抽象。流式输出是构建好体验的基础，生产级代码需要正确处理各类流式事件。错误处理需要结合指数退避、熔断器和并发控制构建完整的可靠性体系。
-
Token 与成本管理：成本管理是 Agent 系统可持续运营的基石。核心策略包括语义缓存、Prompt 压缩、多模型智能路由。一个百万元用户量级的 Agent 系统，通过智能路由可将成本降低 85%。
-
开发环境：TypeScript + Python 双语言开发环境，通过 Ollama 和 vLLM 可以快速搭建本地模型服务，用于开发测试和敏感数据处理。
从下一章开始，我们将进入 Agent 架构核心，深入解析 Agent = Model + Harness 的工程实现。
关键术语
| 术语 | 英文 | 定义 |
|---|---|---|
| Token | Token | LLM 处理文本的最小单位 |
| 流式输出 | Streaming | 逐字返回生成内容的输出方式 |
| 指数退避 | Exponential Backoff | 重试间隔按 2^n 增长的策略 |
| 熔断器 | Circuit Breaker | 检测高失败率时暂停请求的保护机制 |
| 语义缓存 | Semantic Cache | 基于语义相似度匹配的请求缓存 |
| SSE | Server-Sent Events | 服务端向客户端推送事件的 HTTP 协议 |
| tiktoken | tiktoken | OpenAI 的 Token 计数工具库 |
| 速率限制 | Rate Limiting | API 对请求频率的限制 |
思考与练习
-
模型选型实践：为以下三个场景选择合适的模型并解释理由：
- 场景A：面向中国用户的 AI 客服 Agent（日请求量 50 万次）
- 场景B：面向全球开发者的代码审查 Agent（对推理质量要求极高）
- 场景C：处理医疗数据的诊断辅助 Agent（数据不能离开内网）
-
流式输出实现：使用你喜欢的一种语言，实现一个带缓冲（200字符阈值）的流式输出封装。
-
错误处理测试：实现指数退避重试器，然后模拟 API 错误（429、503、网络超时），观察重试行为是否符合预期。
-
成本优化：如果你有一个日请求量 10 万次、平均 2000 input tokens + 1000 output tokens 的 Agent 系统，全用 Opus 4.7 和全用 Sonnet 4.6 的日成本差多少？设计一个智能路由方案，预估能节省多少。
-
本地模型部署：使用 Ollama 部署一个 Qwen3 或 DeepSeek 模型，编写 Python 脚本调用它完成一个简单任务（如翻译一段中文为英文），体验本地模型和云端 API 的延迟差异。
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
0
0- 0
已为社区贡献47条内容
所有评论(0)