第 7 章 AI AgentsLangChainLangGraph
第 7 章 在真实产品中部署 Agent
加固骨架
关于本 notebook
从诊断实验室走向生产 API，就是从**"agent 玄学"转向"系统物理学"**。本 notebook 实现了让 agent 在生产环境中存活的三个运维支柱：
- Checkpointing（检查点） —— 周期性状态快照，让 agent 在重启后能够存活。
- Context Pruning（上下文剪枝） —— 感知 token 的历史截断，防止"记忆膨胀"。
- Circuit Breaker（熔断器） —— 基于轮次和基于成本的保险丝，阻止失控循环。
| 模式 | 关注点 | 机制 | "为什么" |
|---|---|---|---|
| Checkpointing | 可靠性 | 周期性地将状态快照保存到磁盘/Redis | 如果进程在第 4 步（共 10 步）崩溃，agent 从第 4 步恢复而不是重新开始。 |
| Context Pruning | 性能 | 感知 token 的历史截断 | 防止"上下文膨胀"，即陈旧、无关的消息降低推理质量并推高成本。 |
| Circuit Breaker | 安全性 | 基于轮次和基于成本的保险丝 | 如果 agent 陷入工具调用失败的无限循环或超出预算，则停止它。 |
我们将这三种模式整合到一个加固版 agent 循环中，然后运行一个实时演示来证明每一种模式都能工作。
1. AgentSession 让恢复成为可能
通过 Pydantic 使 agent 的整个状态可序列化，你就可以在每一轮之后对它进行快照。当进程在第 4 步（共 10 步）崩溃时，你从第 4 步恢复，而不是从头开始。在生产环境中，这会落到 Redis 或 Postgres，而不是一个 JSON 文件。
2. 上下文剪枝不是可选项
随着上下文增长，模型会变慢并失去焦点。上下文管家（Context Janitor）保持系统 prompt 完整，保留最近的轮次，并总结被丢弃的内容。这不是清理，而是一个性能与成本控制机制。
3. 为每个会话硬编码预算上限
熔断器是防御无限循环的首要手段。一个卡在工具调用循环中的 agent 会在几分钟内烧光你的 API 预算。每个会话 0.50 美元的上限是廉价的保险。当熔断器跳闸时，你的 FastAPI 层捕获异常并返回一个用户友好的"我需要人工介入"消息。
4. 这些模式可以组合
Checkpointing、剪枝和熔断器不是独立的功能，它们是同一个循环的不同层。每一轮：剪枝 → 调用 → 更新指标 → 检查熔断器 → 保存检查点。去掉任何一个，系统就会出现盲区。
5. 流式传输 vs. 轮询
在 FastAPI 生产环境中，对于长时间运行的 agent 任务，避免使用标准的 REST POST 请求。Agent 需要时间来"思考"，30 秒超时很常见。使用 WebSockets 或 Server-Sent Events (SSE) 将 agent 的思考过程流式传输到 UI。这并不会让模型更快，但看到 agent "正在工作"可以降低用户感知到的延迟。
%pip install -q openai pydantic python-dotenv tiktokenimport json, os, time, tempfile, shutil
from pathlib import Path
from typing import Any, Dict, List, Literal, Optional
from dataclasses import dataclass, field
from pydantic import BaseModel, Field
from dotenv import load_dotenv
from openai import OpenAI
import tiktoken
load_dotenv()
OPENROUTER_API_KEY = os.getenv('OPENROUTER_API_KEY')
BASE = "https://openrouter.ai/api/v1"
client = OpenAI(
base_url=BASE,
api_key=OPENROUTER_API_KEY,
)
MODEL = "openai/gpt-4o-mini"
# tiktoken encoding matched to the model — this gives us *exact* token
# counts for pre-call context management, not the chars÷4 guesswork
# you see in tutorials.
ENCODING = tiktoken.encoding_for_model("gpt-4o-mini")
print(f"Model: {MODEL}")
print(f"API base: {BASE}")
print(f"Encoder: {ENCODING.name} ({len(ENCODING._mergeable_ranks):,} merge rules)")Model: openai/gpt-4o-mini
API base: https://openrouter.ai/api/v1
Encoder: o200k_base (199,998 merge rules)Agent 工具
在生产环境中，这些函数会调用你的订单数据库、策略服务和工单分类器。这里我们使用带有真实数据的确定性替身，使加固模式可测试、可复现——但工具 schema、调度器和函数调用契约正是你将要上线发布的东西。
# Tool definitions (OpenAI function-calling schema)
TOOL_SCHEMAS: list[dict] = [
{
"type": "function",
"function": {
"name": "classify_ticket",
"description": (
"Classify a customer support ticket into a category. "
"Returns the category and a confidence score."
),
"parameters": {
"type": "object",
"properties": {
"summary": {
"type": "string",
"description": "A one-sentence summary of the customer's issue.",
},
"category": {
"type": "string",
"enum": ["billing", "technical", "account", "general"],
"description": "The most fitting category for this ticket.",
},
},
"required": ["summary", "category"],
},
},
},
{
"type": "function",
"function": {
"name": "lookup_order",
"description": (
"Look up an order by its ID and return current status, "
"shipping estimate, and item details."
),
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order identifier, e.g. ORD-1024.",
},
},
"required": ["order_id"],
},
},
},
{
"type": "function",
"function": {
"name": "check_policy",
"description": (
"Look up a company policy by keyword. Useful for return, "
"refund, warranty, or shipping policy questions."
),
"parameters": {
"type": "object",
"properties": {
"keyword": {
"type": "string",
"description": "Policy topic to search for, e.g. 'return', 'refund', 'warranty'.",
},
},
"required": ["keyword"],
},
},
},
]
# Mock data
_MOCK_ORDERS: dict[str, dict] = {
"ORD-1024": {
"status": "processing",
"placed": "2025-02-04",
"estimated_ship": "2025-02-14",
"items": ["Wireless Keyboard x1", "USB-C Hub x1"],
},
"ORD-5567": {
"status": "shipped",
"placed": "2025-02-01",
"tracking": "1Z999AA10123456784",
"estimated_delivery": "2025-02-12",
"items": ["Running Shoes (Size 10) x1"],
},
"ORD-8842": {
"status": "delayed",
"placed": "2025-01-20",
"reason": "Supplier back-order; new estimated ship date 2025-02-18.",
"items": ["Standing Desk Frame x1", "Monitor Arm x2"],
},
}
_MOCK_POLICIES: dict[str, str] = {
"return": (
"Items may be returned within 30 days of delivery in original "
"packaging. Electronics must be unopened or defective. Refunds "
"are processed within 5-7 business days."
),
"refund": (
"Refunds are issued to the original payment method. Partial "
"refunds may apply if the item shows signs of use. Shipping "
"costs are non-refundable unless the return is due to our error."
),
"warranty": (
"All electronics carry a 1-year limited warranty covering "
"manufacturing defects. Accessories are covered for 90 days. "
"File a warranty claim through your account dashboard."
),
"shipping": (
"Standard shipping: 5-7 business days. Express: 2-3 business "
"days. Free standard shipping on orders over $50. International "
"shipping available to select countries."
),
}
# Mock tool implementations
def classify_ticket(summary: str, category: str) -> str:
confidence = 0.92 if category in ("billing", "technical", "account") else 0.78
return json.dumps({"category": category, "summary": summary, "confidence": confidence})
def lookup_order(order_id: str) -> str:
order_id = order_id.strip().upper()
if order_id in _MOCK_ORDERS:
return json.dumps({"order_id": order_id, **_MOCK_ORDERS[order_id]})
return json.dumps({"order_id": order_id, "error": "Order not found. Please verify the order ID."})
def check_policy(keyword: str) -> str:
keyword = keyword.strip().lower()
for key, text in _MOCK_POLICIES.items():
if key in keyword or keyword in key:
return json.dumps({"policy": key, "text": text})
return json.dumps({"keyword": keyword, "text": "No specific policy found. Contact support@example.com."})
# Dispatcher
TOOL_MAP: dict[str, callable] = {
"classify_ticket": classify_ticket,
"lookup_order": lookup_order,
"check_policy": check_policy,
}
def execute_tool(name: str, arguments: dict) -> str:
"""Dispatch a tool call by name. Returns a JSON string result."""
fn = TOOL_MAP[name]
return fn(**arguments)
print(f"✅ {len(TOOL_SCHEMAS)} tools registered: {list(TOOL_MAP.keys())}")
# Quick smoke test
print(f" lookup_order('ORD-1024') → {lookup_order('ORD-1024')[:60]}...")
print(f" check_policy('return') → {check_policy('return')[:60]}...")✅ 3 tools registered: ['classify_ticket', 'lookup_order', 'check_policy']
lookup_order('ORD-1024') → {"order_id": "ORD-1024", "status": "processing", "placed": "...
check_policy('return') → {"policy": "return", "text": "Items may be returned within 3...AgentSession：可序列化的状态
AgentSession 使 agent 的整个状态可序列化。这就是你保存到 Redis、Postgres——或者在我们的演示中——磁盘上 JSON 文件的"大脑"。
每一个对恢复至关重要的字段都在这里：对话历史、轮次计数器、累计成本和会话状态。如果进程死亡，你重新加载这个对象并精确地从你离开的地方继续。
class AgentSession(BaseModel):
"""Serializable agent state — the brain you checkpoint."""
session_id: str
model: str = MODEL
system_prompt: str = "You are a customer support triage agent."
history: List[Dict[str, Any]] = Field(default_factory=list)
turn_count: int = 0
total_tokens: int = 0
total_usd: float = 0.0
status: Literal["running", "halted", "completed"] = "running"
halt_reason: Optional[str] = None
checkpoints_saved: int = 0
def add_message(self, role: str, content: str, **extra):
"""Append a message to the conversation history."""
msg = {"role": role, "content": content, **extra}
self.history.append(msg)
def summary(self) -> str:
return (
f"Session {self.session_id} | status={self.status} | "
f"turns={self.turn_count} | tokens={self.total_tokens} | "
f"cost=${self.total_usd:.4f} | messages={len(self.history)} | "
f"checkpoints={self.checkpoints_saved}"
)
# Quick sanity check
s = AgentSession(session_id="test-001")
s.add_message("user", "Hello")
print(s.summary())
print(f"Serializes to {len(s.model_dump_json())} bytes of JSON")Session test-001 | status=running | turns=0 | tokens=0 | cost=$0.0000 | messages=1 | checkpoints=0
Serializes to 267 bytes of JSONToken 计数与成本追踪
两个不同的工作，两个不同的真相来源：
- 调用前（上下文管理）： 在发送请求之前，我们需要知道历史记录将消耗多少 token，以便上下文管家进行剪枝。我们使用
tiktoken——与模型使用的相同的 BPE tokenizer——来精确计数。 - 调用后（成本追踪）： 每次 API 响应之后，OpenRouter 返回实际的
usage.prompt_tokens和usage.completion_tokens。我们使用这些进行计费，而不是本地估算。
# Pre-call: tiktoken-based token counting
def count_message_tokens(message: Dict[str, Any]) -> int:
"""Count tokens in a single chat message using tiktoken.
Follows the OpenAI token-counting recipe: every message has a fixed
overhead for role/name framing, plus the content tokens.
See: https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb
"""
# Per-message overhead (role, delimiters) — 4 tokens for gpt-4o/gpt-4o-mini
tokens = 4
for key, value in message.items():
if isinstance(value, str):
tokens += len(ENCODING.encode(value))
elif isinstance(value, list):
# tool_calls, repro_steps, etc.
tokens += len(ENCODING.encode(json.dumps(value)))
return tokens
def count_tokens(messages: List[Dict[str, Any]]) -> int:
"""Count total tokens for a full conversation history.
Includes the 2-token reply primer that the API adds after the last message.
"""
total = sum(count_message_tokens(m) for m in messages)
total += 2 # reply primer
return total
# Post-call: cost from actual API-reported usage
# gpt-4o-mini pricing (per 1K tokens) — update these when pricing changes
COST_PER_1K_INPUT = 0.00015 # $0.15 / 1M input tokens
COST_PER_1K_OUTPUT = 0.0006 # $0.60 / 1M output tokens
def calculate_cost(prompt_tokens: int, completion_tokens: int) -> float:
"""Calculate USD cost from the API-reported token counts."""
return (
(prompt_tokens / 1000) * COST_PER_1K_INPUT +
(completion_tokens / 1000) * COST_PER_1K_OUTPUT
)
# Demo: show the difference between real counting and guessing
demo_msgs = [
{"role": "system", "content": "You are a customer support triage agent."},
{"role": "user", "content": "I placed order #ORD-8842 three weeks ago and nobody has told me what's going on."},
]
real_count = count_tokens(demo_msgs)
naive_guess = sum(len(str(m.get("content", ""))) for m in demo_msgs) // 4
print(f"tiktoken count: {real_count} tokens")
print(f"chars÷4 guess: {naive_guess} tokens")
print(f"Error: {abs(real_count - naive_guess)} tokens ({abs(real_count - naive_guess)/real_count*100:.0f}%)")
print(f"\nCost for 1000 in + 500 out: ${calculate_cost(1000, 500):.6f}")tiktoken count: 41 tokens
chars÷4 guess: 30 tokens
Error: 11 tokens (27%)
Cost for 1000 in + 500 out: $0.000450上下文管家：历史剪枝
高端模型拥有巨大的上下文窗口，但将其完全填满是一个错误。这会增加延迟并"模糊"agent 的注意力。管家通过以下方式保持历史精简：
- 从不剪枝系统 prompt —— 它包含 agent 的核心身份。
- 只保留最近的轮次 —— 保留即时上下文。
- 注入被剪枝内容的摘要 以维持长期记忆。
def prune_history(
history: List[Dict],
max_tokens: int = 4000,
keep_recent: int = 6,
) -> List[Dict]:
"""
The Context Janitor: keeps the system prompt and the last N turns,
discarding the middle. Injects a summary of pruned content.
Token counts are computed with tiktoken — the same BPE tokenizer
the model uses — so the pruning threshold is exact, not a guess.
"""
current_tokens = count_tokens(history)
if current_tokens <= max_tokens:
return history # Nothing to prune
# Never prune the system prompt (first message)
system_msg = history[0] if history and history[0].get("role") == "system" else None
if system_msg:
middle = history[1:-keep_recent] if len(history) > keep_recent + 1 else []
recent = history[-keep_recent:]
else:
middle = history[:-keep_recent] if len(history) > keep_recent else []
recent = history[-keep_recent:]
# Build a summary of what we're pruning
n_pruned = len(middle)
pruned_roles = [m.get("role", "?") for m in middle]
role_counts = {r: pruned_roles.count(r) for r in set(pruned_roles)}
role_summary = ", ".join(f"{count} {role}" for role, count in role_counts.items())
summary_msg = {
"role": "system",
"content": (
f"[Context Janitor] Pruned {n_pruned} older messages "
f"({role_summary}) to stay within token budget. "
f"Recent context preserved below."
),
}
pruned = []
if system_msg:
pruned.append(system_msg)
pruned.append(summary_msg)
pruned.extend(recent)
after_tokens = count_tokens(pruned)
print(
f" 🧹 Context Janitor: {len(history)} msgs ({current_tokens:,} tokens) → "
f"{len(pruned)} msgs ({after_tokens:,} tokens) | pruned {n_pruned} middle messages"
)
return pruned
# Demo: build a fat history and prune it
demo_history = [{"role": "system", "content": "You are a helpful agent."}]
for i in range(20):
demo_history.append({"role": "user", "content": f"Turn {i}: " + "x" * 200})
demo_history.append({"role": "assistant", "content": f"Reply {i}: " + "y" * 200})
print(f"Before: {len(demo_history)} messages, {count_tokens(demo_history):,} tokens (tiktoken)")
pruned = prune_history(demo_history, max_tokens=1500, keep_recent=6)
print(f"After: {len(pruned)} messages, {count_tokens(pruned):,} tokens (tiktoken)")
print(f"First message role: {pruned[0]['role']}")
print(f"Second message: {pruned[1]['content'][:100]}...")Before: 41 messages, 1,933 tokens (tiktoken)
🧹 Context Janitor: 41 msgs (1,933 tokens) → 8 msgs (336 tokens) | pruned 34 middle messages
After: 8 messages, 336 tokens (tiktoken)
First message role: system
Second message: [Context Janitor] Pruned 34 older messages (17 assistant, 17 user) to stay within token budget. Rece...熔断器
为每个会话硬编码预算上限是防御无限循环的首要手段。熔断器有两个保险丝：
- 预算保险丝：如果会话的累计成本超过
BUDGET_CAP，立即停止。 - 轮次保险丝：如果 agent 执行的推理步骤超过
MAX_TURNS，停止。
当熔断器跳闸时，它会抛出 CircuitBreakerTripped 异常，你的 FastAPI 层可以捕获它并将其转换为用户友好的"我卡住了"消息。
class CircuitBreakerTripped(RuntimeError):
"""Raised when the agent exceeds its operational budget or turn limit."""
def __init__(self, reason: str, session: AgentSession):
self.reason = reason
self.session = session
super().__init__(reason)
def check_circuit_breaker(
session: AgentSession,
budget_cap: float = 0.50,
max_turns: int = 12,
):
"""
Trip the fuse if the agent is burning money or stuck in a loop.
Call this AFTER every LLM call.
"""
if session.total_usd >= budget_cap:
session.status = "halted"
session.halt_reason = f"Budget exceeded: ${session.total_usd:.4f} >= ${budget_cap:.2f}"
raise CircuitBreakerTripped(session.halt_reason, session)
if session.turn_count >= max_turns:
session.status = "halted"
session.halt_reason = f"Max turns exceeded: {session.turn_count} >= {max_turns}"
raise CircuitBreakerTripped(session.halt_reason, session)
# Demo: force a trip
demo_session = AgentSession(session_id="breaker-test", total_usd=0.48)
try:
demo_session.total_usd += 0.05 # push over the $0.50 cap
check_circuit_breaker(demo_session, budget_cap=0.50)
except CircuitBreakerTripped as e:
print(f"⚡ Circuit breaker tripped: {e.reason}")
print(f" Session status: {e.session.status}")⚡ Circuit breaker tripped: Budget exceeded: $0.5300 >= $0.50
Session status: halted检查点：保存与恢复
在每一轮之后保存检查点可以确保 502 错误或容器重启不会迫使用户从头开始。在生产环境中这落在 Redis 或 Postgres。在演示中我们使用一个 JSON 文件，这使得检查点可被检查。
CHECKPOINT_DIR = Path(tempfile.mkdtemp(prefix="agent_checkpoints_"))
print(f"Checkpoint directory: {CHECKPOINT_DIR}")
def save_checkpoint(session: AgentSession) -> Path:
"""Persist the full session state to disk, counting this checkpoint."""
path = CHECKPOINT_DIR / f"{session.session_id}.json"
session.checkpoints_saved += 1
try:
path.write_text(session.model_dump_json(indent=2))
except Exception:
session.checkpoints_saved -= 1
raise
return path
def load_checkpoint(session_id: str) -> Optional[AgentSession]:
"""Restore a session from its last checkpoint. Returns None if not found."""
path = CHECKPOINT_DIR / f"{session_id}.json"
if not path.exists():
return None
data = json.loads(path.read_text())
return AgentSession.model_validate(data)
def list_checkpoints() -> List[str]:
"""List all saved session IDs."""
return [p.stem for p in CHECKPOINT_DIR.glob("*.json")]
# Demo
demo = AgentSession(session_id="ckpt-demo", turn_count=3, total_usd=0.012)
demo.add_message("system", "You are a helpful agent.")
demo.add_message("user", "What is your return policy?")
demo.add_message("assistant", "Our return policy allows returns within 30 days.")
path = save_checkpoint(demo)
print(f"Saved checkpoint to: {path}")
print(f"Checkpoint size: {path.stat().st_size} bytes")
restored = load_checkpoint("ckpt-demo")
print(f"Restored: {restored.summary()}")
print(f"History intact: {len(restored.history)} messages")Checkpoint directory: /tmp/agent_checkpoints_qu63q9s0
Saved checkpoint to: /tmp/agent_checkpoints_qu63q9s0/ckpt-demo.json
Checkpoint size: 551 bytes
Restored: Session ckpt-demo | status=running | turns=3 | tokens=0 | cost=$0.0120 | messages=3 | checkpoints=0
History intact: 3 messages加固版 Agent 循环
这就是三种模式汇聚的地方。循环的每一次迭代：
- 在调用模型之前剪枝上下文（上下文管家）
- 调用 LLM 并执行任何工具调用
- 更新指标（token、成本、轮次数）
- 检查熔断器（预算 + 轮次）
- 保存检查点（状态持久化）
如果任何一步失败，最后一个检查点让你能够恢复。
MAX_TOOL_ROUNDS = 5 # Safety cap on tool-call loops within a single turn
def hardened_agent_step(
session: AgentSession,
user_input: str,
budget_cap: float = 0.50,
max_turns: int = 12,
max_context_tokens: int = 4000,
) -> str:
"""
Run one hardened agent turn: prune → call LLM → tools → metrics → breaker → checkpoint.
Returns the assistant's final reply.
"""
if session.status != "running":
raise RuntimeError(f"Session {session.session_id} is {session.status}: {session.halt_reason}")
# Ensure system prompt is in history
if not session.history or session.history[0].get("role") != "system":
session.history.insert(0, {"role": "system", "content": session.system_prompt})
# Add the user message
session.add_message("user", user_input)
# Context Janitor: prune before calling the model
session.history = prune_history(
session.history,
max_tokens=max_context_tokens,
keep_recent=6,
)
# LLM call (with tool-call loop)
final_reply = None
for _round in range(MAX_TOOL_ROUNDS):
response = client.chat.completions.create(
model=session.model,
messages=session.history,
tools=TOOL_SCHEMAS,
tool_choice="auto",
temperature=0.2,
)
choice = response.choices[0]
message = choice.message
# Update metrics
usage = response.usage
prompt_tok = (usage.prompt_tokens or 0) if usage else 0
compl_tok = (usage.completion_tokens or 0) if usage else 0
session.total_tokens += prompt_tok + compl_tok
session.total_usd += calculate_cost(prompt_tok, compl_tok)
# No tool calls → final answer
tool_calls = getattr(message, "tool_calls", None)
if not tool_calls:
final_reply = message.content or "(no response)"
session.add_message("assistant", final_reply)
break
# Execute tool calls
session.history.append(message.model_dump())
for tc in tool_calls:
fn_name = tc.function.name
try:
fn_args = json.loads(tc.function.arguments)
except json.JSONDecodeError:
fn_args = {}
try:
result_str = execute_tool(fn_name, fn_args)
except Exception as exc:
result_str = json.dumps({"error": str(exc)})
session.history.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result_str,
})
print(f" 🔧 Tool: {fn_name}({fn_args}) → {result_str[:80]}...")
else:
final_reply = "(Agent reached maximum tool-call rounds.)"
session.add_message("assistant", final_reply)
# Increment turn counter
session.turn_count += 1
# Circuit Breaker
check_circuit_breaker(session, budget_cap=budget_cap, max_turns=max_turns)
# Checkpoint
ckpt_path = save_checkpoint(session)
print(f" 💾 Checkpoint saved (turn {session.turn_count}, ${session.total_usd:.4f})")
return final_reply
print("Hardened agent loop ready.")Hardened agent loop ready.演示：三种模式齐上阵
现在我们用四个场景运行加固版 agent，证明每种模式都能工作：
- 正常的多轮操作 —— 每一轮之后保存检查点
- 上下文剪枝触发 —— 我们膨胀历史记录，观察管家修剪它
- 熔断器跳闸 —— 我们设置一个较低的轮次上限，观察保险丝熔断
- 崩溃恢复 —— 我们"杀掉"会话并从最后一个检查点恢复
演示 1：正常的多轮操作
一段真实的客服对话：客户有一个延迟的订单，询问退货政策，然后跟进。agent 使用工具，每轮保存检查点，熔断器保持绿色。
session1 = AgentSession(
session_id="showcase-normal-001",
model=MODEL,
system_prompt=(
"You are a customer support triage agent. Your job is to:\n"
"1. Classify the ticket into a category (billing, technical, account, general).\n"
"2. Assess priority (low, medium, high, urgent).\n"
"3. Use available tools to gather context.\n"
"4. Write a brief, empathetic response.\n"
"Be concise. Never fabricate order details — use the lookup tool."
),
)
conversation = [
"I placed order #ORD-8842 three weeks ago and nobody has told me what's going on. This is unacceptable.",
"OK, so it's delayed. What are my options? Can I return it when it arrives if it's too late?",
"Fine. I'll wait, but if it's not here by the 18th I want a full refund. Please escalate this.",
]
print("=" * 70)
print(" SHOWCASE 1: Normal Multi-Turn Operation")
print("=" * 70)
for i, msg in enumerate(conversation, 1):
print(f"\n{'─' * 70}")
print(f"Turn {i} | Customer: {msg[:80]}...")
print(f"{'─' * 70}")
reply = hardened_agent_step(session1, msg)
print(f"\n 🤖 Agent: {reply[:200]}..." if len(reply) > 200 else f"\n 🤖 Agent: {reply}")
session1.status = "completed"
save_checkpoint(session1)
print(f"\n{'=' * 70}")
print(f"Final: {session1.summary()}")
print(f"{'=' * 70}")======================================================================
SHOWCASE 1: Normal Multi-Turn Operation
======================================================================
──────────────────────────────────────────────────────────────────────
Turn 1 | Customer: I placed order #ORD-8842 three weeks ago and nobody has told me what's going on....
──────────────────────────────────────────────────────────────────────
🔧 Tool: classify_ticket({'summary': "I placed order #ORD-8842 three weeks ago and nobody has told me what's going on.", 'category': 'general'}) → {"category": "general", "summary": "I placed order #ORD-8842 three weeks ago and...
🔧 Tool: lookup_order({'order_id': 'ORD-8842'}) → {"order_id": "ORD-8842", "status": "delayed", "placed": "2025-01-20", "reason": ...
💾 Checkpoint saved (turn 1, $0.0002)
🤖 Agent: **Category:** General
**Priority:** Urgent
---
Dear Customer,
I sincerely apologize for the delay with your order #ORD-8842. It is currently delayed due to a supplier back-order, and the new es...
──────────────────────────────────────────────────────────────────────
Turn 2 | Customer: OK, so it's delayed. What are my options? Can I return it when it arrives if it'...
──────────────────────────────────────────────────────────────────────
🔧 Tool: check_policy({'keyword': 'return'}) → {"policy": "return", "text": "Items may be returned within 30 days of delivery i...
💾 Checkpoint saved (turn 2, $0.0005)
🤖 Agent: You can return your order when it arrives if it's too late for your needs. According to our return policy, items may be returned within 30 days of delivery in their original packaging. Please note tha...
──────────────────────────────────────────────────────────────────────
Turn 3 | Customer: Fine. I'll wait, but if it's not here by the 18th I want a full refund. Please e...
──────────────────────────────────────────────────────────────────────
🔧 Tool: classify_ticket({'summary': 'Request for a full refund if order #ORD-8842 is not delivered by February 18.', 'category': 'billing'}) → {"category": "billing", "summary": "Request for a full refund if order #ORD-8842...
💾 Checkpoint saved (turn 3, $0.0008)
🤖 Agent: **Category:** Billing
**Priority:** High
---
Dear Customer,
I understand your concern and the urgency of your request for a full refund if your order #ORD-8842 is not delivered by February 18. ...
======================================================================
Final: Session showcase-normal-001 | status=completed | turns=3 | tokens=4055 | cost=$0.0008 | messages=14 | checkpoints=4
======================================================================演示 2：上下文剪枝实战
我们故意用 20 条填充消息膨胀历史记录，然后发送一个真实的问题。上下文管家应该在 LLM 调用之前修剪多余内容，保留系统 prompt 和最近的轮次。
session2 = AgentSession(
session_id="showcase-pruning-002",
model=MODEL,
system_prompt="You are a customer support triage agent. Be concise.",
)
# Inject the system prompt
session2.history.append({"role": "system", "content": session2.system_prompt})
# Bloat the history with 20 turns of filler
for i in range(20):
session2.history.append({"role": "user", "content": f"Filler question {i}: " + "blah " * 60})
session2.history.append({"role": "assistant", "content": f"Filler answer {i}: " + "response " * 60})
print("=" * 70)
print(" SHOWCASE 2: Context Pruning in Action")
print("=" * 70)
print(f"\nHistory BEFORE: {len(session2.history)} messages, {count_tokens(session2.history):,} tokens")
print(f"\nSending a real question into the bloated context...\n")
reply = hardened_agent_step(
session2,
"Can you look up order #ORD-5567 for me?",
max_context_tokens=1500, # Force aggressive pruning
)
print(f"\nHistory AFTER: {len(session2.history)} messages, {count_tokens(session2.history):,} tokens")
print(f"\n 🤖 Agent: {reply[:300]}")
print(f"\n{session2.summary()}")======================================================================
SHOWCASE 2: Context Pruning in Action
======================================================================
History BEFORE: 41 messages, 2,899 tokens
Sending a real question into the bloated context...
🧹 Context Janitor: 42 msgs (2,917 tokens) → 8 msgs (432 tokens) | pruned 35 middle messages
🔧 Tool: lookup_order({'order_id': 'ORD-5567'}) → {"order_id": "ORD-5567", "status": "shipped", "placed": "2025-02-01", "tracking"...
💾 Checkpoint saved (turn 1, $0.0003)
History AFTER: 11 messages, 670 tokens
🤖 Agent: Order #ORD-5567 has been shipped. Here are the details:
- **Status:** Shipped
- **Placed on:** February 1, 2025
- **Tracking Number:** 1Z999AA10123456784
- **Estimated Delivery:** February 12, 2025
- **Items:** Running Shoes (Size 10) x1
Session showcase-pruning-002 | status=running | turns=1 | tokens=1403 | cost=$0.0003 | messages=11 | checkpoints=1演示 3：熔断器跳闸
我们设置 max_turns=3 并发送 5 条消息。熔断器应该在第 4 轮跳闸，停止会话并抛出 CircuitBreakerTripped。在 FastAPI 应用中，你会捕获它并返回用户友好的"我需要人工介入"消息。
session3 = AgentSession(
session_id="showcase-breaker-003",
model=MODEL,
system_prompt="You are a customer support triage agent. Be concise.",
)
messages_to_send = [
"What is your return policy?",
"What about warranties?",
"How long does shipping take?",
"Can you look up order #ORD-1024?", # This should trigger the breaker
"One more question...", # Should never reach this
]
print("=" * 70)
print(" SHOWCASE 3: Circuit Breaker Trips (max_turns=3)")
print("=" * 70)
for i, msg in enumerate(messages_to_send, 1):
try:
print(f"\n Turn {i}: {msg}")
reply = hardened_agent_step(session3, msg, max_turns=3)
print(f" 🤖 {reply[:120]}..." if len(reply) > 120 else f" 🤖 {reply}")
except CircuitBreakerTripped as e:
print(f"\n ⚡ CIRCUIT BREAKER TRIPPED on turn {i}!")
print(f" Reason: {e.reason}")
print(f" Session status: {e.session.status}")
print(f" → In production: return 'I need to hand you to a human agent.'")
break
except RuntimeError as e:
print(f"\n 🚫 Session refused (already halted): {e}")
break
print(f"\n{session3.summary()}")======================================================================
SHOWCASE 3: Circuit Breaker Trips (max_turns=3)
======================================================================
Turn 1: What is your return policy?
🔧 Tool: check_policy({'keyword': 'return'}) → {"policy": "return", "text": "Items may be returned within 30 days of delivery i...
💾 Checkpoint saved (turn 1, $0.0001)
🤖 Our return policy allows items to be returned within 30 days of delivery in their original packaging. Electronics must b...
Turn 2: What about warranties?
🔧 Tool: check_policy({'keyword': 'warranty'}) → {"policy": "warranty", "text": "All electronics carry a 1-year limited warranty ...
💾 Checkpoint saved (turn 2, $0.0002)
🤖 All electronics come with a 1-year limited warranty covering manufacturing defects, while accessories are covered for 90...
Turn 3: How long does shipping take?
🔧 Tool: check_policy({'keyword': 'shipping'}) → {"policy": "shipping", "text": "Standard shipping: 5-7 business days. Express: 2...
⚡ CIRCUIT BREAKER TRIPPED on turn 3!
Reason: Max turns exceeded: 3 >= 3
Session status: halted
→ In production: return 'I need to hand you to a human agent.'
Session showcase-breaker-003 | status=halted | turns=3 | tokens=2315 | cost=$0.0004 | messages=13 | checkpoints=2演示 4：从检查点崩溃恢复
我们通过"遗忘"会话对象来模拟崩溃，然后从最后一个检查点恢复它。agent 精确地从它离开的地方继续。相同的历史、相同的指标、相同的轮次数。
print("=" * 70)
print(" SHOWCASE 4: Crash Recovery from Checkpoint")
print("=" * 70)
# Step 1: Start a session and run 2 turns
session4 = AgentSession(
session_id="showcase-recovery-004",
model=MODEL,
system_prompt="You are a customer support agent. Be concise and helpful.",
)
print("\n── Phase 1: Running 2 turns before the 'crash' ──")
reply1 = hardened_agent_step(session4, "I need to check on order #ORD-1024.")
print(f" 🤖 Turn 1: {reply1[:150]}")
reply2 = hardened_agent_step(session4, "When will it ship? I need it by Friday.")
print(f" 🤖 Turn 2: {reply2[:150]}")
print(f"\n State before crash: {session4.summary()}")
# Step 2: Simulate the crash
print("\n── Phase 2: 💥 SIMULATED CRASH (deleting session object) ──")
crashed_session_id = session4.session_id
del session4 # Gone. Process died. Container restarted.
print(f" Session object destroyed. Only the checkpoint remains.")
# Step 3: Recover from checkpoint
print("\n── Phase 3: Recovering from checkpoint ──")
print(f" Available checkpoints: {list_checkpoints()}")
recovered = load_checkpoint(crashed_session_id)
if recovered:
print(f" ✅ Recovered: {recovered.summary()}")
print(f" History length: {len(recovered.history)} messages")
print(f" Last message: {recovered.history[-1]['content'][:100]}...")
# Step 4: Continue the conversation
print("\n── Phase 4: Continuing the conversation after recovery ──")
reply3 = hardened_agent_step(recovered, "Actually, can you also check the shipping policy?")
print(f" 🤖 Turn 3 (post-recovery): {reply3[:200]}")
print(f"\n Final state: {recovered.summary()}")
else:
print(f" ❌ No checkpoint found for {crashed_session_id}")======================================================================
SHOWCASE 4: Crash Recovery from Checkpoint
======================================================================
── Phase 1: Running 2 turns before the 'crash' ──
🔧 Tool: lookup_order({'order_id': 'ORD-1024'}) → {"order_id": "ORD-1024", "status": "processing", "placed": "2025-02-04", "estima...
💾 Checkpoint saved (turn 1, $0.0001)
🤖 Turn 1: Your order #ORD-1024 is currently processing. It was placed on February 4, 2025, and is estimated to ship by February 14, 2025. The items in your orde
💾 Checkpoint saved (turn 2, $0.0002)
🤖 Turn 2: Your order is estimated to ship by February 14, 2025. If you need it by Friday, please note that it may not arrive in time, as February 14 is a Wednes
State before crash: Session showcase-recovery-004 | status=running | turns=2 | tokens=1019 | cost=$0.0002 | messages=7 | checkpoints=2
── Phase 2: 💥 SIMULATED CRASH (deleting session object) ──
Session object destroyed. Only the checkpoint remains.
── Phase 3: Recovering from checkpoint ──
Available checkpoints: ['showcase-recovery-004', 'showcase-breaker-003', 'showcase-pruning-002', 'ckpt-demo', 'showcase-normal-001']
✅ Recovered: Session showcase-recovery-004 | status=running | turns=2 | tokens=1019 | cost=$0.0002 | messages=7 | checkpoints=1
History length: 7 messages
Last message: Your order is estimated to ship by February 14, 2025. If you need it by Friday, please note that it ...
── Phase 4: Continuing the conversation after recovery ──
🔧 Tool: check_policy({'keyword': 'shipping'}) → {"policy": "shipping", "text": "Standard shipping: 5-7 business days. Express: 2...
💾 Checkpoint saved (turn 3, $0.0004)
🤖 Turn 3 (post-recovery): Here is the shipping policy:
- **Standard Shipping**: 5-7 business days.
- **Express Shipping**: 2-3 business days.
- **Free Standard Shipping**: Available on orders over $50.
- **International Shipp
Final state: Session showcase-recovery-004 | status=running | turns=3 | tokens=2062 | cost=$0.0004 | messages=11 | checkpoints=2汇总仪表盘
print("=" * 80)
print(" HARDENING THE BACKBONE — SHOWCASE SUMMARY")
print("=" * 80)
# Gather all sessions we can find in checkpoints
all_sessions = []
for sid in list_checkpoints():
s = load_checkpoint(sid)
if s:
all_sessions.append(s)
print(f"\n{'Session ID':<30} {'Status':<12} {'Turns':>6} {'Tokens':>8} {'Cost':>10} {'Checkpts':>10} {'Halt Reason'}")
print("─" * 110)
for s in sorted(all_sessions, key=lambda x: x.session_id):
halt = s.halt_reason or "—"
print(
f"{s.session_id:<30} {s.status:<12} {s.turn_count:>6} "
f"{s.total_tokens:>8} ${s.total_usd:>8.4f} {s.checkpoints_saved:>10} {halt}"
)
total_cost = sum(s.total_usd for s in all_sessions)
total_tokens = sum(s.total_tokens for s in all_sessions)
total_turns = sum(s.turn_count for s in all_sessions)
print(f"\n{'─' * 110}")
print(f"{'TOTAL':<30} {'':12} {total_turns:>6} {total_tokens:>8} ${total_cost:>8.4f}")
print(f"\n\n📋 Patterns Demonstrated:")
print(f" ✅ Checkpointing — {sum(s.checkpoints_saved for s in all_sessions)} checkpoints saved across all sessions")
print(f" ✅ Context Pruning — Showcase 2 bloated history was trimmed before LLM call")
breaker_sessions = [s for s in all_sessions if s.status == "halted"]
print(f" ✅ Circuit Breaker — {len(breaker_sessions)} session(s) halted by the breaker")
recovery_sessions = [s for s in all_sessions if "recovery" in s.session_id]
if recovery_sessions:
print(f" ✅ Crash Recovery — Session '{recovery_sessions[0].session_id}' restored and continued")================================================================================
HARDENING THE BACKBONE — SHOWCASE SUMMARY
================================================================================
Session ID Status Turns Tokens Cost Checkpts Halt Reason
──────────────────────────────────────────────────────────────────────────────────────────────────────────────
ckpt-demo running 3 0 $ 0.0120 0 —
showcase-breaker-003 running 2 1310 $ 0.0002 1 —
showcase-normal-001 completed 3 4055 $ 0.0008 3 —
showcase-pruning-002 running 1 1403 $ 0.0003 0 —
showcase-recovery-004 running 3 2062 $ 0.0004 1 —
──────────────────────────────────────────────────────────────────────────────────────────────────────────────
TOTAL 12 8830 $ 0.0137
📋 Patterns Demonstrated:
✅ Checkpointing — 5 checkpoints saved across all sessions
✅ Context Pruning — Showcase 2 bloated history was trimmed before LLM call
✅ Circuit Breaker — 0 session(s) halted by the breaker
✅ Crash Recovery — Session 'showcase-recovery-004' restored and continued清理
# Clean up the temporary checkpoint directory
shutil.rmtree(CHECKPOINT_DIR, ignore_errors=True)
print(f"Cleaned up checkpoint directory: {CHECKPOINT_DIR}")Cleaned up checkpoint directory: /tmp/agent_checkpoints_qu63q9s0推理后端
关于本 notebook
推理后端：冷启动、缓存与部署物理学
当你的 agent 调用一个模型时，它不是在与"一个 API"对话。它是在与一个GPU 进程对话——而那个进程有物理特性。在生成第一个 token 之前，模型必须先加载到 VRAM 中。KV cache 必须为每个新前缀计算。内存必须被分配、管理，有时还要被逐出。
这些就是决定你的 agent 是在 200ms 内还是 12 秒内响应的部署现实：
| 关注点 | 会发生什么 | 对你的 agent 的影响 |
|---|---|---|
| 冷启动 | 模型权重从磁盘/网络加载到 GPU VRAM | 部署后的第一个请求需要 30-120s 而不是 <1s |
| KV Cache | 为 prompt 中的每个 token 计算注意力键/值 | 除非缓存，否则每次调用都会重新计算长系统 prompt |
| 前缀缓存 | 跨请求复用共享 prompt 前缀的 KV cache | 相同的系统 prompt → 跳过重新计算 → TTFT 提升 2-5 倍 |
| 缓存失效 | 前缀改变 → 缓存的 KV 条目变得无用 | 新系统 prompt 或对话分叉 = 冷缓存延迟尖峰 |
| 内存压力 | KV cache 随上下文长度 × 批次大小增长 | 长对话逐出其他请求的缓存，导致尾部延迟尖峰 |
本 notebook 启动一个本地 vLLM 推理服务器，然后用真实请求测量这些效应中的每一个——这样你就能看到数字，而不只是读到它们。
我们还对比了三个主要的开源推理后端——vLLM、TGI 和 SGLang——这样你就知道什么时候该用哪一个。
冷启动是部署税，不是 bug。 每次你重新部署、从零扩容或崩溃后重启，都有一个 30-120s 的窗口，你的 agent 无响应。通过保活探测、预热请求或 min-replica > 0 的扩缩容策略来缓解。
KV cache 是你 agent 的工作记忆——而且它很昂贵。 更长的对话 = 更多 VRAM = 更慢的 TTFT = 更少的并发用户。这就是为什么上下文剪枝（上一个 notebook）不只是成本优化——它是延迟优化。
前缀缓存几乎免费，并且能大幅降低 TTFT。 如果你的 agent 使用稳定的系统 prompt（它应该如此），请启用前缀缓存。但要注意，动态前缀内容（RAG 片段、用户画像）会使缓存失效。设计你的 prompt 模板时，采用稳定前缀和可变后缀。
缓存失效是灵活性的隐藏成本。 每次你更改系统 prompt、分叉对话或在 prompt 顶部注入新上下文，你都要再次付出完整的 KV 计算成本。这就是 agent 灵活性与推理效率之间的权衡。
你的 agent 代码应该与后端无关。 使用 OpenAI SDK 作为你的客户端库。vLLM、TGI、SGLang 以及每个云服务商都说同一种 API。切换后端是配置更改，而不是重构。
Production Deployment Checklist:
☐ Inference backend selected (vLLM / TGI / SGLang)
☐ Prefix caching enabled
☐ System prompt designed with stable prefix
☐ Cold start measured and documented
☐ Health check endpoint monitored
☐ Min replicas > 0 (avoid scale-to-zero cold starts)
☐ Context pruning configured (max history tokens)
☐ GPU memory budget estimated (model + KV cache + overhead)推理后端：vLLM vs. TGI vs. SGLang
这三个都是开源的、GPU 原生的推理服务器，暴露兼容 OpenAI 的 API。你的 agent 代码不需要改变——client.chat.completions.create() 对它们全部适用。改变的是运维画像。
| 特性 | vLLM | TGI (Text Generation Inference) | SGLang |
|---|---|---|---|
| 维护方 | UC Berkeley（开源） | Hugging Face | LMSYS |
| 调度 | PagedAttention（内存高效的 KV） | Continuous batching | RadixAttention（基于树的 KV 复用） |
| 前缀缓存 | --enable-prefix-caching (APC) |
✗（仅手动 prompt 缓存） | 内置（RadixAttention） |
| 多 GPU | 张量并行 + 流水线并行 | 张量并行 | 张量并行 + 专家并行 |
| 投机解码 | ✓（草稿模型或 n-gram） | ✓（medusa heads） | ✓ |
| 量化 | AWQ, GPTQ, FP8, BitsAndBytes | AWQ, GPTQ, EETQ, BitsAndBytes | AWQ, GPTQ, FP8 |
| 结构化输出 | ✓（guided decoding / outlines） | ✓（基于语法） | ✓（压缩 FSM） |
| 视觉模型 | ✓ | ✓ | ✓ |
| 最适合 | 通用、高吞吐 | HF 生态系统、快速部署 | 多轮 agent（树缓存） |
| 冷启动 | ~30-120s（取决于模型） | ~20-90s | ~30-120s |
| API 兼容性 | 兼容 OpenAI | 兼容 OpenAI + TGI 原生 | 兼容 OpenAI + SGLang 原生 |
何时使用哪个：
vLLM：默认选择。文档最好、模型支持最广、经过实战检验的 PagedAttention。除非你有特定理由，否则用它。
TGI：你已经处于 Hugging Face 生态系统中（Inference Endpoints、SageMaker）。与
transformers和 HF Hub 紧密集成。如果你已经是 HF 店铺，这是通往生产的最快路径。SGLang：你的 agent 做大量的多轮推理（思维树、自洽性、分支工具调用）。RadixAttention 跨对话分支复用 KV cache 的效率高于 vLLM 的线性前缀缓存。它还以其压缩的有限状态机方法在结构化输出方面表现出色。
%pip install -q vllm openaiimport json, os, time, subprocess, signal, statistics
from pathlib import Path
from openai import OpenAI
import requests
# ── Model configuration ───────────────────────────────────────────────
# Swap the model for anything your GPU can hold.
# Qwen2.5-3B fits on a single T4 (16 GB). For an A100, try 8B or 14B.
VLLM_MODEL = "Qwen/Qwen2.5-3B-Instruct"
SERVED_NAME = "qwen"
# ── Engine args ───────────────────────────────────────────────────────
# These map directly to vLLM's engine arguments:
# https://docs.vllm.ai/en/stable/configuration/engine_args/
VLLM_PORT = 8001
VLLM_HOST = "0.0.0.0"
TENSOR_PARALLEL_SIZE = 1 # number of GPUs for tensor parallelism
GPU_MEMORY_UTILIZATION = 0.90 # fraction of GPU VRAM vLLM may use
MAX_MODEL_LEN = 4096 # max context window (tokens)
DTYPE = "auto" # "auto" | "float16" | "bfloat16"
SWAP_SPACE = 4 # GiB of CPU swap for KV cache overflow
MAX_NUM_SEQS = 64 # max concurrent sequences in a batch
DISABLE_LOG_STATS = True # quieter logs in notebook
# The OpenAI-compatible client — same interface you already use for
# OpenRouter, Azure, or any hosted API. That's the point: your agent
# code doesn't change when you swap the backend.
local_client = OpenAI(
base_url=f"http://localhost:{VLLM_PORT}/v1",
api_key="not-needed", # local server, no auth
)
print(f"Model: {VLLM_MODEL}")
print(f"Server: http://localhost:{VLLM_PORT}/v1")
print(f"Engine: TP={TENSOR_PARALLEL_SIZE}, GPU mem={GPU_MEMORY_UTILIZATION}, "
f"max_len={MAX_MODEL_LEN}, dtype={DTYPE}")Model: Qwen/Qwen2.5-3B-Instruct
Server: http://localhost:8001/v1
Engine: TP=1, GPU mem=0.9, max_len=4096, dtype=auto1 — 冷启动：从 vllm serve 到第一个 Token
冷启动不仅仅是"服务器启动需要一段时间"。它是一连串阻塞步骤，每一步都会在你的 agent 服务第一个用户之前增加延迟：
┌────────────────┐ ┌──────────────────┐ ┌──────────────────────┐ ┌─────────────┐
│ Download model │ → │ Load into VRAM │ → │ 1st request: alloc │ → │ Warm request │
│ weights (if │ │ (deserialize + │ │ KV cache, compile │ │ (steady │
│ not cached) │ │ shard across │ │ remaining kernels, │ │ state) │
│ │ │ GPUs) │ │ warm scheduler │ │ │
│ ~0-60s │ │ ~15-60s │ │ ~1-5s extra │ │ ~0.2-1s │
└────────────────┘ └──────────────────┘ └──────────────────────┘ └─────────────┘
Phase 1 Phase 2 Phase 3 Phase 4
──────── cold_start_seconds ──────── ─ 1st req ─ ── warm ──我们对所有这些进行端到端测量：从头启动服务器，等待健康，然后计时前几个请求，以查看完整的从冷到热的转变。
我们通过 python -m vllm.entrypoints.openai.api_server 启动 vLLM——这是暴露 /v1/chat/completions 的 OpenAI 兼容入口点（底层与 vllm serve 相同）。这些标志直接映射到 vLLM 的**引擎参数**：
| 标志 | 它控制什么 | 为什么重要 |
|---|---|---|
--model |
HF 模型 ID 或本地路径 | 加载到 GPU VRAM 的内容 |
--tensor-parallel-size |
用于 TP 的 GPU 数量 | 跨 GPU 拆分模型 |
--gpu-memory-utilization |
用于 KV cache 的 VRAM 比例 | 太高 → OOM；太低 → 更少的并发序列 |
--max-model-len |
最大上下文窗口 | 限制每个请求的 KV cache 内存 |
--swap-space |
KV 溢出的 CPU RAM（GiB） | GPU 缓存满时的安全阀 |
--dtype |
计算精度 | auto 根据 GPU 选择 bf16/fp16 |
--max-num-seqs |
最大并发序列数 | 限制内存并控制批处理 |
--enforce-eager |
禁用 CUDA graph 捕获 | 启动更快，吞吐量略低 |
--disable-log-stats |
抑制周期性统计 | 为 notebook 提供更干净的日志 |
import subprocess, time, requests
VLLM_LOG = "vllm_server.log"
def build_vllm_args(
model: str = VLLM_MODEL,
served_name: str = SERVED_NAME,
port: int = VLLM_PORT,
extra_args: list[str] | None = None,
) -> list[str]:
"""Build the vLLM server command using python -m vllm.entrypoints.openai.api_server.
Every flag here maps to a vLLM Engine Argument:
https://docs.vllm.ai/en/stable/configuration/engine_args/
"""
cmd = [
"python", "-m", "vllm.entrypoints.openai.api_server",
f"--host={VLLM_HOST}",
f"--port={port}",
f"--model={model}",
f"--served-model-name={served_name}",
f"--tensor-parallel-size={TENSOR_PARALLEL_SIZE}",
f"--gpu-memory-utilization={GPU_MEMORY_UTILIZATION}",
f"--max-model-len={MAX_MODEL_LEN}",
f"--dtype={DTYPE}",
f"--swap-space={SWAP_SPACE}",
f"--max-num-seqs={MAX_NUM_SEQS}",
"--enforce-eager", # faster startup, skip CUDA graph capture
]
if DISABLE_LOG_STATS:
cmd.append("--disable-log-stats")
if extra_args:
cmd.extend(extra_args)
return cmd
def start_vllm_server(
model: str = VLLM_MODEL,
served_name: str = SERVED_NAME,
port: int = VLLM_PORT,
extra_args: list[str] | None = None,
) -> subprocess.Popen:
"""Start a vLLM server process and return the handle."""
cmd = build_vllm_args(model, served_name, port, extra_args)
log_fh = open(VLLM_LOG, "w")
proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT)
# Pretty-print the launch command
print(f" vLLM PID {proc.pid}")
print(f" cmd: \\\n " + " \\\n ".join(cmd))
return proc
def wait_for_healthy(port: int = VLLM_PORT, timeout: int = 300) -> float:
"""Block until the vLLM health endpoint responds. Returns seconds waited."""
url = f"http://localhost:{port}/health"
t0 = time.time()
while time.time() - t0 < timeout:
try:
r = requests.get(url, timeout=2)
if r.status_code == 200:
elapsed = time.time() - t0
print(f" ✅ Server healthy after {elapsed:.1f}s")
return elapsed
except requests.ConnectionError:
pass
time.sleep(2)
raise TimeoutError(f"vLLM not healthy after {timeout}s — check {VLLM_LOG}")
def stop_vllm_server(proc: subprocess.Popen):
"""Gracefully stop the vLLM server."""
proc.terminate()
try:
proc.wait(timeout=10)
except subprocess.TimeoutExpired:
proc.kill()
print(f" Server stopped (PID {proc.pid})")SYSTEM_PROMPT = (
"You are a customer support triage agent for an e-commerce company. "
"Classify tickets by category and priority. Be concise."
)
TEST_MESSAGE = "I placed order #ORD-8842 three weeks ago and nobody has told me anything. This is unacceptable."
def timed_completion(
client: OpenAI,
messages: list[dict],
model: str = SERVED_NAME,
max_tokens: int = 256,
) -> dict:
"""Send a chat completion and return timing + usage metadata."""
t0 = time.time()
response = client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens,
temperature=0.2,
)
elapsed = time.time() - t0
usage = response.usage
content = response.choices[0].message.content or ""
return {
"latency_s": round(elapsed, 3),
"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,
"content": content,
"tokens_per_sec": round(
(usage.completion_tokens / elapsed) if usage and elapsed > 0 else 0, 1
),
}
# PHASE 1+2: Start server from scratch, measure boot time
print("=" * 70)
print(" COLD START — FULL END-TO-END MEASUREMENT")
print("=" * 70)
t_total_start = time.time()
print("\n Phase 1+2: Starting vLLM server (model download + GPU loading)...\n")
vllm_proc = start_vllm_server()
cold_start_seconds = wait_for_healthy()
print(f"\n ⏱ Server boot (Phase 1+2): {cold_start_seconds:.1f}s")
print(f" Model loaded into VRAM, health endpoint responding.\n")
# PHASE 3+4: First requests — cold GPU → warm steady state
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": TEST_MESSAGE},
]
results = []
labels = [
"1st request (cold GPU) ← Phase 3",
"2nd request (warming)",
"3rd request (warm) ← Phase 4",
"4th request (warm)",
"5th request (warm)",
]
print(" Phase 3+4: Request latencies after boot:")
print(" " + "─" * 66)
for i, label in enumerate(labels):
r = timed_completion(local_client, messages)
results.append(r)
print(
f" {label:<35} │ {r['latency_s']:>6.3f}s │ "
f"{r['completion_tokens']:>4} tok │ "
f"{r['tokens_per_sec']:>6.1f} tok/s"
)
t_total_first_response = time.time() - t_total_start
# Summary
cold_req = results[0]["latency_s"]
warm_avg = statistics.mean(r["latency_s"] for r in results[2:]) # 3rd-5th
total_to_first = cold_start_seconds + cold_req
print(f"\n{'═' * 70}")
print(f" COLD START BREAKDOWN")
print(f"{'─' * 70}")
print(f" Server boot (Phase 1+2): {cold_start_seconds:>7.1f}s model → VRAM")
print(f" 1st request (Phase 3): {cold_req:>7.3f}s KV alloc + kernel compile")
print(f" ────────────────────────── ─────────")
print(f" Total cold start: {total_to_first:>7.1f}s deploy → first answer")
print(f"")
print(f" Warm request (Phase 4): {warm_avg:>7.3f}s steady state")
print(f" Cold/Warm ratio: {total_to_first / warm_avg:>7.0f}× ← your user waits this much longer")
print(f"{'─' * 70}")
print(f" → After every deploy, restart, or scale-from-zero event, your agent")
print(f" is OFFLINE for ~{total_to_first:.0f}s. Plan for it: health checks, warm-up")
print(f" probes, min-replicas > 0, or rolling deploys.")======================================================================
COLD START — FULL END-TO-END MEASUREMENT
======================================================================
Phase 1+2: Starting vLLM server (model download + GPU loading)...
vLLM PID 3941
cmd: \
python \
-m \
vllm.entrypoints.openai.api_server \
--host=0.0.0.0 \
--port=8001 \
--model=Qwen/Qwen2.5-3B-Instruct \
--served-model-name=qwen \
--tensor-parallel-size=1 \
--gpu-memory-utilization=0.9 \
--max-model-len=4096 \
--dtype=auto \
--swap-space=4 \
--max-num-seqs=64 \
--enforce-eager \
--disable-log-stats
✅ Server healthy after 124.1s
⏱ Server boot (Phase 1+2): 124.1s
Model loaded into VRAM, health endpoint responding.
Phase 3+4: Request latencies after boot:
──────────────────────────────────────────────────────────────────
1st request (cold GPU) ← Phase 3 │ 0.458s │ 10 tok │ 21.9 tok/s
2nd request (warming) │ 0.276s │ 10 tok │ 36.2 tok/s
3rd request (warm) ← Phase 4 │ 0.277s │ 10 tok │ 36.1 tok/s
4th request (warm) │ 0.279s │ 10 tok │ 35.8 tok/s
5th request (warm) │ 0.276s │ 10 tok │ 36.2 tok/s
══════════════════════════════════════════════════════════════════════
COLD START BREAKDOWN
──────────────────────────────────────────────────────────────────────
Server boot (Phase 1+2): 124.1s model → VRAM
1st request (Phase 3): 0.458s KV alloc + kernel compile
────────────────────────── ─────────
Total cold start: 124.6s deploy → first answer
Warm request (Phase 4): 0.277s steady state
Cold/Warm ratio: 449× ← your user waits this much longer
──────────────────────────────────────────────────────────────────────
→ After every deploy, restart, or scale-from-zero event, your agent
is OFFLINE for ~125s. Plan for it: health checks, warm-up
probes, min-replicas > 0, or rolling deploys.上下文长度与 KV Cache 压力
注意： 本节特别适用于仅解码器（自回归）模型——GPT、LLaMA、Qwen、Mistral 等——也就是 vLLM、TGI 和 SGLang 所服务的模型。编码器-解码器架构（T5、BART）只计算一次编码器 KV cache，并在所有解码步骤中复用它，因此 prompt 长度对那里的每 token 生成延迟影响要小得多。
在仅解码器模型中，prompt 中的每个 token 都会产生一对 KV，它们必须存储在 GPU 内存中并且在后续每个生成步骤中被注意力机制访问。随着你的 agent 对话历史的增长，会发生三件事：
- TTFT（首 token 时间）增加 —— prefill 阶段必须计算整个 prompt 上的注意力；成本随序列长度二次增长
- 内存压力上升 —— KV cache 随上下文长度线性增长，留给批处理并发请求的空间减少
- 每 token 解码变慢 —— 每个新 token 都要访问之前所有的 KV 对，因此更长的上下文意味着每一步更多的内存读取
这就是为什么上下文剪枝（来自上一个 notebook）不只是关于成本——它是关于保持推理后端响应灵敏。
# ── Build prompts of increasing context length ───────────────────────
FILLER_TURN = (
"The customer previously wrote: 'I ordered a blue widget on January 5th "
"and the tracking says it's stuck in Memphis. Can someone look into this? "
"I also have a question about your return policy for items over $50.'"
)
def build_messages(num_history_turns: int) -> list[dict]:
"""Build a conversation with N filler turns + the real question."""
msgs = [{"role": "system", "content": SYSTEM_PROMPT}]
for i in range(num_history_turns):
msgs.append({"role": "user", "content": f"[Turn {i+1}] {FILLER_TURN}"})
msgs.append({"role": "assistant", "content": f"Acknowledged turn {i+1}."})
msgs.append({"role": "user", "content": TEST_MESSAGE})
return msgs
# ── Measure latency vs context size ──────────────────────────────────
turn_counts = [0, 5, 10, 20, 30]
context_results = []
print("=" * 70)
print(" CONTEXT LENGTH vs. LATENCY")
print("=" * 70)
for n in turn_counts:
msgs = build_messages(n)
r = timed_completion(local_client, msgs, max_tokens=128)
context_results.append({"turns": n, **r})
print(
f" {n:>3} history turns │ {r['prompt_tokens']:>5} prompt tok │ "
f"{r['latency_s']:>6.3f}s │ {r['tokens_per_sec']:>6.1f} tok/s"
)
# ── Compute scaling factor ────────────────────────────────────────────
baseline = context_results[0]["latency_s"]
heaviest = context_results[-1]["latency_s"]
print(f"\n{'─' * 70}")
print(f" Baseline (0 turns): {baseline:.3f}s")
print(f" Heaviest ({turn_counts[-1]} turns): {heaviest:.3f}s")
print(f" Slowdown: {heaviest / baseline:.1f}×")
print(f"\n → Long agent conversations get slower even if the final question is identical.")======================================================================
CONTEXT LENGTH vs. LATENCY
======================================================================
0 history turns │ 62 prompt tok │ 0.282s │ 35.4 tok/s
5 history turns │ 407 prompt tok │ 0.223s │ 31.4 tok/s
10 history turns │ 754 prompt tok │ 0.286s │ 34.9 tok/s
20 history turns │ 1464 prompt tok │ 0.232s │ 30.2 tok/s
30 history turns │ 2174 prompt tok │ 0.309s │ 32.4 tok/s
──────────────────────────────────────────────────────────────────────
Baseline (0 turns): 0.282s
Heaviest (30 turns): 0.309s
Slowdown: 1.1×
→ Long agent conversations get slower even if the final question is identical.前缀缓存：A/B 对比
大多数 agent 对每个请求使用相同的系统 prompt。如果没有前缀缓存，服务器会在每次调用时重新计算该系统 prompt 的 KV cache——纯粹的浪费。
vLLM 的 --enable-prefix-caching（APC——自动前缀缓存）会检测多个请求何时共享公共前缀，并复用缓存的 KV 块：
Request 1: [SYSTEM: "You are a support agent..."] + [USER: "My order is late"]
↑ compute KV, store in cache
Request 2: [SYSTEM: "You are a support agent..."] + [USER: "Refund policy?"]
↑ cache HIT → skip KV computation for this prefix为了看到真正的差异，我们运行完全相同的 5 个问题两次：
- 没有前缀缓存（当前服务器）
- 有
--enable-prefix-caching（重启服务器）
然后并排比较。
# 5 test questions (same system prompt, different user messages) ─
user_messages = [
"I placed order #ORD-8842 three weeks ago and nobody has told me anything.",
"What is your return policy for electronics?",
"I need to cancel order #ORD-1234 immediately.",
"My package arrived damaged — the box was crushed.",
"Can I change the shipping address on my pending order?",
]
def run_prefix_experiment(client, label: str) -> list[dict]:
"""Send the same 5 questions and collect latency results."""
# Warm-up request (not counted) — primes GPU scheduler / KV allocator
_ = timed_completion(client, messages, max_tokens=16)
time.sleep(0.5)
results = []
for i, user_msg in enumerate(user_messages):
msgs = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_msg},
]
r = timed_completion(client, msgs, max_tokens=128)
results.append(r)
print(
f" Q{i+1}: {r['latency_s']:>6.3f}s │ "
f"{r['prompt_tokens']:>4} prompt tok │ "
f"{r['tokens_per_sec']:>6.1f} tok/s │ "
f"{user_msg[:50]}..."
)
return results
# RUN A: WITHOUT prefix caching (current server)
print("=" * 70)
print(" RUN A — WITHOUT prefix caching")
print("=" * 70)
results_no_cache = run_prefix_experiment(local_client, "no-cache")
# Restart server WITH prefix caching
print(f"\n{'─' * 70}")
print("Stopping current server...")
stop_vllm_server(vllm_proc)
time.sleep(3)
print("\nStarting vLLM server WITH --enable-prefix-caching...")
vllm_proc = start_vllm_server(extra_args=["--enable-prefix-caching"])
wait_for_healthy()
# RUN B: WITH prefix caching
print("\n" + "=" * 70)
print(" RUN B — WITH prefix caching")
print("=" * 70)
results_with_cache = run_prefix_experiment(local_client, "with-cache")
# SIDE-BY-SIDE COMPARISON
print("\n" + "=" * 70)
print(" A/B COMPARISON — prefix caching OFF vs ON")
print("=" * 70)
print(f" {'Q':<3} │ {'Without':>9} │ {'With':>9} │ {'Speedup':>8} │ Question")
print(f" {'─'*3}─┼─{'─'*9}─┼─{'─'*9}─┼─{'─'*8}─┼─{'─'*40}")
for i, (no_c, with_c) in enumerate(zip(results_no_cache, results_with_cache)):
speedup = no_c["latency_s"] / with_c["latency_s"] if with_c["latency_s"] > 0 else 0
print(
f" Q{i+1} │ {no_c['latency_s']:>8.3f}s │ {with_c['latency_s']:>8.3f}s │ "
f"{speedup:>7.2f}× │ {user_messages[i][:40]}..."
)
avg_no = statistics.mean(r["latency_s"] for r in results_no_cache)
avg_with = statistics.mean(r["latency_s"] for r in results_with_cache)
avg_speedup = avg_no / avg_with if avg_with > 0 else 0
# Use results from Q2-Q5 for "cached" comparison (Q1 primes the prefix cache)
avg_no_2_5 = statistics.mean(r["latency_s"] for r in results_no_cache[1:])
avg_with_2_5 = statistics.mean(r["latency_s"] for r in results_with_cache[1:])
cached_speedup = avg_no_2_5 / avg_with_2_5 if avg_with_2_5 > 0 else 0
print(f"\n{'─' * 70}")
print(f" Avg (all 5): {avg_no:.3f}s → {avg_with:.3f}s ({avg_speedup:.2f}× speedup)")
print(f" Avg (Q2-Q5): {avg_no_2_5:.3f}s → {avg_with_2_5:.3f}s ({cached_speedup:.2f}× speedup)")
print(f"\n → Q1 primes the prefix cache. From Q2 onward the system prompt KV is")
print(f" reused — that's the steady-state benefit your agent gets in production.")======================================================================
RUN A — WITHOUT prefix caching
======================================================================
Q1: 0.304s │ 58 prompt tok │ 32.9 tok/s │ I placed order #ORD-8842 three weeks ago and nobod...
Q2: 1.033s │ 46 prompt tok │ 35.8 tok/s │ What is your return policy for electronics?...
Q3: 0.228s │ 52 prompt tok │ 30.6 tok/s │ I need to cancel order #ORD-1234 immediately....
Q4: 0.282s │ 48 prompt tok │ 35.4 tok/s │ My package arrived damaged — the box was crushed....
Q5: 0.258s │ 49 prompt tok │ 34.9 tok/s │ Can I change the shipping address on my pending or...
──────────────────────────────────────────────────────────────────────
Stopping current server...
Server stopped (PID 3941)
Starting vLLM server WITH --enable-prefix-caching...
vLLM PID 4853
cmd: \
python \
-m \
vllm.entrypoints.openai.api_server \
--host=0.0.0.0 \
--port=8001 \
--model=Qwen/Qwen2.5-3B-Instruct \
--served-model-name=qwen \
--tensor-parallel-size=1 \
--gpu-memory-utilization=0.9 \
--max-model-len=4096 \
--dtype=auto \
--swap-space=4 \
--max-num-seqs=64 \
--enforce-eager \
--disable-log-stats \
--enable-prefix-caching
✅ Server healthy after 54.0s
======================================================================
RUN B — WITH prefix caching
======================================================================
Q1: 0.286s │ 58 prompt tok │ 35.0 tok/s │ I placed order #ORD-8842 three weeks ago and nobod...
Q2: 0.813s │ 46 prompt tok │ 38.1 tok/s │ What is your return policy for electronics?...
Q3: 0.205s │ 52 prompt tok │ 34.1 tok/s │ I need to cancel order #ORD-1234 immediately....
Q4: 0.283s │ 48 prompt tok │ 35.3 tok/s │ My package arrived damaged — the box was crushed....
Q5: 0.258s │ 49 prompt tok │ 34.9 tok/s │ Can I change the shipping address on my pending or...
======================================================================
A/B COMPARISON — prefix caching OFF vs ON
======================================================================
Q │ Without │ With │ Speedup │ Question
────┼───────────┼───────────┼──────────┼─────────────────────────────────────────
Q1 │ 0.304s │ 0.286s │ 1.06× │ I placed order #ORD-8842 three weeks ago...
Q2 │ 1.033s │ 0.813s │ 1.27× │ What is your return policy for electroni...
Q3 │ 0.228s │ 0.205s │ 1.11× │ I need to cancel order #ORD-1234 immedia...
Q4 │ 0.282s │ 0.283s │ 1.00× │ My package arrived damaged — the box was...
Q5 │ 0.258s │ 0.258s │ 1.00× │ Can I change the shipping address on my ...
──────────────────────────────────────────────────────────────────────
Avg (all 5): 0.421s → 0.369s (1.14× speedup)
Avg (Q2-Q5): 0.450s → 0.390s (1.16× speedup)
→ Q1 primes the prefix cache. From Q2 onward the system prompt KV is
reused — that's the steady-state benefit your agent gets in production.缓存失效：当前缀改变时
前缀缓存很强大——直到前缀改变。在 agent 系统中，这发生在你：
- 更改系统 prompt（A/B 测试、角色切换）
- 注入动态上下文（RAG 检索、用户画像、工具结果）
- 分叉对话（分支 agent 路径）
每次前缀更改都是一次缓存未命中——服务器必须从头重新计算新前缀的 KV cache。这就是"缓存失效税"。
# ── Different system prompts = cache misses ───────────────────────────
SYSTEM_PROMPTS = {
"support_agent": (
"You are a customer support triage agent for an e-commerce company. "
"Classify tickets by category and priority. Be concise."
),
"returns_specialist": (
"You are a returns and refund specialist. Help customers with return "
"eligibility, refund timelines, and exchange options. Be precise about "
"policy details and timelines."
),
"escalation_agent": (
"You are an escalation manager who handles VIP customers and complex "
"cases. Use a formal, empathetic tone. Offer concrete next steps "
"and set clear expectations for resolution timelines."
),
"back_to_support": (
"You are a customer support triage agent for an e-commerce company. "
"Classify tickets by category and priority. Be concise."
),
}
fixed_user_msg = "My order hasn't arrived and I'm really frustrated."
print("=" * 70)
print(" CACHE INVALIDATION — CHANGING THE SYSTEM PROMPT")
print("=" * 70)
invalidation_results = {}
for label, sys_prompt in SYSTEM_PROMPTS.items():
msgs = [
{"role": "system", "content": sys_prompt},
{"role": "user", "content": fixed_user_msg},
]
r = timed_completion(local_client, msgs, max_tokens=128)
invalidation_results[label] = r
cache_status = "MISS" if label != "back_to_support" else "HIT (same as #1)"
print(
f" {label:<22} │ {r['latency_s']:>6.3f}s │ "
f"{r['tokens_per_sec']:>6.1f} tok/s │ cache: {cache_status}"
)
# ── Highlight the cost of switching ──────────────────────────────────
first = invalidation_results["support_agent"]["latency_s"]
back = invalidation_results["back_to_support"]["latency_s"]
print(f"\n{'─' * 70}")
print(f" First 'support_agent': {first:.3f}s (cache miss → compute KV)")
print(f" Return to same prompt: {back:.3f}s (cache hit → reuse KV)")
print(f"\n → Dynamic system prompts (RAG injection, role switching) invalidate")
print(f" the prefix cache. Design for stable prefixes where possible.")======================================================================
CACHE INVALIDATION — CHANGING THE SYSTEM PROMPT
======================================================================
support_agent │ 0.283s │ 35.3 tok/s │ cache: MISS
returns_specialist │ 3.253s │ 39.4 tok/s │ cache: MISS
escalation_agent │ 3.267s │ 39.2 tok/s │ cache: MISS
back_to_support │ 0.315s │ 34.9 tok/s │ cache: HIT (same as #1)
──────────────────────────────────────────────────────────────────────
First 'support_agent': 0.283s (cache miss → compute KV)
Return to same prompt: 0.315s (cache hit → reuse KV)
→ Dynamic system prompts (RAG injection, role switching) invalidate
the prefix cache. Design for stable prefixes where possible.对话增长：多轮中的缓存行为
在真实的 agent 循环中，对话逐轮增长。有了前缀缓存，每一轮新轮次只需要对新 token 进行 KV 计算——整个前缀（系统 prompt + 之前所有轮次）都被缓存。
但是当对话分叉时（例如，agent 用不同的工具调用重试），分叉点会使后缀缓存失效。这就是"线性对话"（缓存友好）与"思维树"（缓存敌对）之间的权衡。
# ── Simulate a growing conversation (linear) ─────────────────────────
conversation = [{"role": "system", "content": SYSTEM_PROMPT}]
turn_pairs = [
("My order #ORD-8842 hasn't arrived.", "I'll look into that for you."),
("It's been three weeks now.", "I understand your frustration. Let me check the status."),
("Can you expedite the shipping?", "I'll escalate this to our logistics team."),
("What about a refund if it doesn't arrive by Friday?", "Our policy allows refunds after 21 business days."),
("Okay, can you also check order #ORD-9901?", "Sure, let me pull up that order as well."),
("That one has the wrong item. I got a red widget instead of blue.", "I'm sorry about that. Let me initiate a return for the incorrect item."),
]
print("=" * 70)
print(" MULTI-TURN CONVERSATION — LINEAR GROWTH (cache-friendly)")
print("=" * 70)
linear_results = []
for i, (user_msg, assistant_msg) in enumerate(turn_pairs):
conversation.append({"role": "user", "content": user_msg})
r = timed_completion(local_client, conversation, max_tokens=128)
linear_results.append({"turn": i + 1, **r})
# Add the assistant response to keep the conversation going
conversation.append({"role": "assistant", "content": assistant_msg})
print(
f" Turn {i+1} │ {r['prompt_tokens']:>5} prompt tok │ "
f"{r['latency_s']:>6.3f}s │ {r['tokens_per_sec']:>6.1f} tok/s"
)
# ── Now fork: same base conversation, different last message ─────────
print(f"\n{'─' * 70}")
print(" CONVERSATION FORK — cache invalidation at the branch point")
print("─" * 70)
# Fork A: continue the original conversation
fork_a = conversation.copy()
fork_a.append({"role": "user", "content": "Actually, forget the return. Just send me a replacement."})
# Fork B: different branch from the same point
fork_b = conversation.copy()
fork_b.append({"role": "user", "content": "I want to speak with a supervisor about both orders."})
r_a = timed_completion(local_client, fork_a, max_tokens=128)
r_b = timed_completion(local_client, fork_b, max_tokens=128)
print(f" Fork A (replacement): {r_a['latency_s']:>6.3f}s │ {r_a['prompt_tokens']} prompt tok")
print(f" Fork B (supervisor): {r_b['latency_s']:>6.3f}s │ {r_b['prompt_tokens']} prompt tok")
print(f"\n → Both forks share the same prefix and differ only in the last message.")
print(f" With prefix caching, the shared history is not recomputed.")======================================================================
MULTI-TURN CONVERSATION — LINEAR GROWTH (cache-friendly)
======================================================================
Turn 1 │ 51 prompt tok │ 0.281s │ 35.6 tok/s
Turn 2 │ 76 prompt tok │ 0.279s │ 35.8 tok/s
Turn 3 │ 104 prompt tok │ 0.525s │ 38.1 tok/s
Turn 4 │ 135 prompt tok │ 0.661s │ 37.8 tok/s
Turn 5 │ 171 prompt tok │ 0.502s │ 37.9 tok/s
Turn 6 │ 208 prompt tok │ 0.272s │ 33.1 tok/s
──────────────────────────────────────────────────────────────────────
CONVERSATION FORK — cache invalidation at the branch point
──────────────────────────────────────────────────────────────────────
Fork A (replacement): 0.343s │ 246 prompt tok
Fork B (supervisor): 0.336s │ 245 prompt tok
→ Both forks share the same prefix and differ only in the last message.
With prefix caching, the shared history is not recomputed.后端可移植性：同一个客户端，不同的服务器
一个关键的部署原则：**你的 agent 代码应该与后端无关。**由于所有三个后端都暴露兼容 OpenAI 的 API，切换是配置更改，而不是代码更改。下面我们展示同一个 OpenAI 客户端构造函数如何适用于每一个。
# ── Backend configurations — same OpenAI client, different base_url ───
#
# Each backend uses its own entrypoint / CLI, but they all expose
# an OpenAI-compatible /v1/chat/completions endpoint.
#
# vLLM engine args reference:
# https://docs.vllm.ai/en/stable/configuration/engine_args/
MODEL_EXAMPLE = "Qwen/Qwen2.5-3B-Instruct"
BACKEND_CONFIGS = {
"vLLM": {
"base_url": "http://localhost:8001/v1",
"start_cmd": [
"python", "-m", "vllm.entrypoints.openai.api_server",
f"--host=0.0.0.0",
f"--port=8001",
f"--model={MODEL_EXAMPLE}",
f"--served-model-name=qwen",
f"--tensor-parallel-size=1",
f"--gpu-memory-utilization=0.90",
f"--max-model-len=4096",
f"--dtype=auto",
f"--swap-space=4",
f"--max-num-seqs=64",
"--enforce-eager",
"--enable-prefix-caching",
"--disable-log-stats",
],
"docs": "https://docs.vllm.ai/en/stable/configuration/engine_args/",
},
"TGI": {
"base_url": "http://localhost:8002/v1",
"start_cmd": [
"text-generation-launcher",
f"--model-id={MODEL_EXAMPLE}",
"--port=8002",
"--hostname=0.0.0.0",
"--max-input-tokens=4096",
"--max-total-tokens=4608",
"--dtype=float16",
],
"docs": "https://huggingface.co/docs/text-generation-inference",
},
"SGLang": {
"base_url": "http://localhost:8003/v1",
"start_cmd": [
"python", "-m", "sglang.launch_server",
f"--model-path={MODEL_EXAMPLE}",
f"--served-model-name=qwen",
"--port=8003",
"--host=0.0.0.0",
],
"docs": "https://docs.sglang.ai/",
},
}
print("=" * 70)
print(" BACKEND PORTABILITY — YOUR AGENT CODE DOESN'T CHANGE")
print("=" * 70)
for name, cfg in BACKEND_CONFIGS.items():
cmd_str = " \\\n ".join(cfg["start_cmd"])
print(f"\n ── {name} ──")
print(f" Docs: {cfg['docs']}")
print(f" Start: {cmd_str}")
print(f" Client: OpenAI(base_url='{cfg['base_url']}', api_key='not-needed')")
print(f" Call: client.chat.completions.create(model='qwen', messages=...)")
print(f"\n{'─' * 70}")
print(" → Same chat.completions.create() call. Swap the base_url and you've")
print(" switched your entire inference backend. No agent code changes.")
print(" This is why using the OpenAI SDK as your client library matters —")
print(" it's the lingua franca of inference APIs.")======================================================================
BACKEND PORTABILITY — YOUR AGENT CODE DOESN'T CHANGE
======================================================================
── vLLM ──
Docs: https://docs.vllm.ai/en/stable/configuration/engine_args/
Start: python \
-m \
vllm.entrypoints.openai.api_server \
--host=0.0.0.0 \
--port=8001 \
--model=Qwen/Qwen2.5-3B-Instruct \
--served-model-name=qwen \
--tensor-parallel-size=1 \
--gpu-memory-utilization=0.90 \
--max-model-len=4096 \
--dtype=auto \
--swap-space=4 \
--max-num-seqs=64 \
--enforce-eager \
--enable-prefix-caching \
--disable-log-stats
Client: OpenAI(base_url='http://localhost:8001/v1', api_key='not-needed')
Call: client.chat.completions.create(model='qwen', messages=...)
── TGI ──
Docs: https://huggingface.co/docs/text-generation-inference
Start: text-generation-launcher \
--model-id=Qwen/Qwen2.5-3B-Instruct \
--port=8002 \
--hostname=0.0.0.0 \
--max-input-tokens=4096 \
--max-total-tokens=4608 \
--dtype=float16
Client: OpenAI(base_url='http://localhost:8002/v1', api_key='not-needed')
Call: client.chat.completions.create(model='qwen', messages=...)
── SGLang ──
Docs: https://docs.sglang.ai/
Start: python \
-m \
sglang.launch_server \
--model-path=Qwen/Qwen2.5-3B-Instruct \
--served-model-name=qwen \
--port=8003 \
--host=0.0.0.0
Client: OpenAI(base_url='http://localhost:8003/v1', api_key='not-needed')
Call: client.chat.completions.create(model='qwen', messages=...)
──────────────────────────────────────────────────────────────────────
→ Same chat.completions.create() call. Swap the base_url and you've
switched your entire inference backend. No agent code changes.
This is why using the OpenAI SDK as your client library matters —
it's the lingua franca of inference APIs.汇总仪表盘
print("=" * 70)
print(" DEPLOYMENT PHYSICS — SUMMARY")
print("=" * 70)
print(f"""
┌─────────────────────────────────────────────────────────────────┐
│ COLD START │
│ Server boot: {cold_start_seconds:>6.1f}s │
│ 1st request: {cold_req:>6.3f}s │
│ Total (boot + 1st req): {total_to_first:>6.1f}s │
│ Warm request avg: {warm_avg:>6.3f}s │
│ Cold/Warm ratio: {total_to_first / warm_avg:>5.0f}× │
│ │
│ CONTEXT LENGTH │
│ 0 turns → {context_results[0]['latency_s']:>6.3f}s │
│ {turn_counts[-1]} turns → {context_results[-1]['latency_s']:>6.3f}s ({context_results[-1]['latency_s'] / context_results[0]['latency_s']:.1f}× slower) │
│ │
│ PREFIX CACHING (A/B) │
│ Without (avg Q2-5): {avg_no_2_5:>6.3f}s │
│ With (avg Q2-5): {avg_with_2_5:>6.3f}s ({cached_speedup:.2f}× faster) │
│ │
│ CACHE INVALIDATION │
│ Same prefix (return): {invalidation_results['back_to_support']['latency_s']:>6.3f}s (cache hit) │
│ New prefix (switch): {invalidation_results['returns_specialist']['latency_s']:>6.3f}s (cache miss) │
│ │
│ MODEL: {VLLM_MODEL:<40} │
└─────────────────────────────────────────────────────────────────┘
""")======================================================================
DEPLOYMENT PHYSICS — SUMMARY
======================================================================
┌─────────────────────────────────────────────────────────────────┐
│ COLD START │
│ Server boot: 124.1s │
│ 1st request: 0.458s │
│ Total (boot + 1st req): 124.6s │
│ Warm request avg: 0.277s │
│ Cold/Warm ratio: 449× │
│ │
│ CONTEXT LENGTH │
│ 0 turns → 0.282s │
│ 30 turns → 0.309s (1.1× slower) │
│ │
│ PREFIX CACHING (A/B) │
│ Without (avg Q2-5): 0.450s │
│ With (avg Q2-5): 0.390s (1.16× faster) │
│ │
│ CACHE INVALIDATION │
│ Same prefix (return): 0.315s (cache hit) │
│ New prefix (switch): 3.253s (cache miss) │
│ │
│ MODEL: Qwen/Qwen2.5-3B-Instruct │
└─────────────────────────────────────────────────────────────────┘清理
# ── Stop the vLLM server ──────────────────────────────────────────────
stop_vllm_server(vllm_proc)
# ── Clean up log file ─────────────────────────────────────────────────
if os.path.exists(VLLM_LOG):
os.remove(VLLM_LOG)
print(f" Removed {VLLM_LOG}")
print(" Done.")Server stopped (PID 4853)
Removed vllm_server.log
Done.模型回退
跨模型的 Structured Output：回退、漂移与 Schema 保险
无论你的主模型有多好，它终究会失败。有时它超时。有时它产生幻觉。有时它产生的输出看起来语法正确但语义错误。在生产环境中，你承担不起假装这不会发生。
每个 agent 系统都需要回退。 但这里有个陷阱：当模型改变时，你的整体 agent 行为也会改变。所以你需要设计系统时就考虑到这一点——而不是在凌晨 2 点第一次宕机之后。你需要不仅用主模型测试你的系统，还要用"第二选择"来测试。
本 notebook 演示一个三层回退策略，并证明每一层为什么重要：
- 第 1 层 —— 严格的 JSON Schema 模式：服务商约束 token 生成以匹配 schema。最好，但不是每个模型都支持，而且它只约束形状，不约束判断。
- 第 2 层 —— Instructor + Pydantic 校验器：我们自己用校验、归一化和自动重试来执行 schema。适用于任何能产生 JSON 的模型。
- 第 3 层 —— 纯 prompt + 规范化：最后的手段。我们在 prompt 中描述 schema，希望模型配合，然后归一化任何返回的内容。
我们通过 OpenRouter 测试四个模型，每个都被要求将同一条客户消息分类为结构化支持工单——并展示即使有了全部三层，模型在优先级和情绪等主观字段上仍然存在分歧。
1. 严格的 JSON schema 模式约束形状，不约束判断
所有支持严格模式的模型都产生了有效的 JSON，但它们对值存在分歧。优先级 p0 与 p1 决定了是否有人会被传呼。情绪 angry 与 frustrated 会改变升级路径。Schema 强制执行结构，而非解释。在带有工具调用和多轮交接的真实 agent 中，即使是结构保证也只适用于单次调用，而不是整个流水线。
2. 并非所有模型都支持严格模式——你需要第 2 层
当你的主模型宕机而你路由到回退时，那个回退可能不支持 response_format=json_schema。没有 Instructor + Pydantic 校验器作为你的第二道防线，你的流水线就会崩溃。Instructor 给你校验和带有错误反馈的自动重试——模型会被告知到底哪里出了问题，并获得另一次机会。
3. Pydantic 校验器是归一化应该放的地方
把归一化映射移入 Pydantic 模型作为 field_validator，意味着每条路径——严格的、Instructor 的和降级的——都受益于相同的归一化。"High" → "p1"、"authentication" → "login"、"anxious" → "frustrated"。这不是可选项；这是 schema 保险。
4. 用整个流水线测试你的回退
不要只测试"这个模型能生成文本吗？"要测试：它的输出能否通过你的三层回退 → 解析 → 归一化 → 校验 → 交接链？如果任何一步失败，回退就没用。
5. 随着你在各层之间回退，延迟会增加
第 1 层（严格）最快，一次 API 调用，无重试。第 2 层（Instructor）在校验失败时可能需要重试。第 3 层（降级 + 规范化）需要额外的 API 调用加上后处理。把这计入你的 SLA 和超时预算。
6. 不同的模型有不同的"判断"
这一点很微妙但至关重要：两个模型看待同一个生气的客户，可能把工单分类为 p0/angry 或 p1/frustrated。如果你的路由逻辑、SLA 计时器或升级规则依赖这些字段，你需要知道回退链中每个模型如何评定它们，以及这种差异对你的用例是否可以接受。这就是为什么你要在加固阶段测试回退模型，而不是在第一次生产宕机之后。
!pip install instructorimport json, re, time, random, requests
from typing import Any, Dict, List, Literal
from pydantic import BaseModel, field_validator
import os
from dotenv import load_dotenv
import instructor
from openai import OpenAI
load_dotenv()
OPENROUTER_API_KEY = os.getenv('OPENROUTER_API_KEY')
BASE = "https://openrouter.ai/api/v1"/usr/local/lib/python3.12/dist-packages/google/colab/_import_hooks/_hook_injector.py:55: FutureWarning:
All support for the `google.generativeai` package has ended. It will no longer be receiving
updates or bug fixes. Please switch to the `google.genai` package as soon as possible.
See README for more details:
https://github.com/google-gemini/deprecated-generative-ai-python/blob/main/README.md
loader.exec_module(module)归一化映射
这些映射是我们防御模型漂移的核心。不同的模型会对同一个语义概念返回不同的值——"High" 而不是 "p1"、"authentication" 而不是 "login"、"anxious" 而不是 "frustrated"。
我们提前定义它们，因为 Pydantic 校验器（第 2 层由 Instructor 使用）和最后手段的 canonicalize() 函数（第 3 层）都依赖它们。
CATEGORY_MAP = {
"authentication": "login",
"authentication/login": "login",
"account access": "login",
"login": "login",
"billing": "billing",
"bug": "bug",
"performance": "performance",
"feature_request": "feature_request",
"other": "other",
}
PRIORITY_MAP = {
"p0": "p0", "p1": "p1", "p2": "p2", "p3": "p3",
"critical": "p0",
"urgent": "p1",
"high": "p1",
"medium": "p2",
"low": "p3",
}
SENTIMENT_MAP = {
"calm": "calm",
"frustrated": "frustrated",
"angry": "angry",
"frustrated/urgent": "frustrated",
"urgent": "frustrated",
"anxious": "frustrated",
}目标 Schema 与 Pydantic 模型
这里两件不同的事情服务两个不同的目的：
- JSON Schema 会交给 API 服务商用于严格模式（第 1 层）。服务商在 token 级别用它进行约束解码。
- Pydantic 模型带有
Literal类型和field_validator，是我们本地的执行。它使用上面的映射归一化输入值，并拒绝任何无法映射的内容。Instructor（第 2 层）使用这个模型来校验响应并在失败时重试。
注意 field_validator：当模型返回 "priority": "High" 时，校验器通过 PRIORITY_MAP 把它映射到 "p1"。如果该值根本无法映射，校验器会抛出 ValueError——Instructor 捕获它，把错误注入 prompt，并带着关于哪里出错的明确反馈给模型另一次机会。
SUPPORT_TICKET_SCHEMA = {
"type": "object",
"properties": {
"ticket_id": {"type": "string"},
"summary": {"type": "string"},
"category": {"type": "string", "enum": ["billing","login","bug","feature_request","performance","other"]},
"priority": {"type": "string", "enum": ["p0","p1","p2","p3"]},
"customer_sentiment": {"type": "string", "enum": ["calm","frustrated","angry"]},
"repro_steps": {"type": "array", "items": {"type": "string"}},
"expected_behavior": {"type": "string"},
"actual_behavior": {"type": "string"},
"suggested_next_action": {"type": "string"},
},
"required": [
"ticket_id","summary","category","priority","customer_sentiment",
"repro_steps","expected_behavior","actual_behavior","suggested_next_action"
],
"additionalProperties": False,
}
class SupportTicket(BaseModel):
ticket_id: str
summary: str
category: Literal["billing", "login", "bug", "feature_request", "performance", "other"]
priority: Literal["p0", "p1", "p2", "p3"]
customer_sentiment: Literal["calm", "frustrated", "angry"]
repro_steps: List[str]
expected_behavior: str
actual_behavior: str
suggested_next_action: str
@field_validator("category", mode="before")
@classmethod
def normalize_category(cls, v: str) -> str:
return CATEGORY_MAP.get(str(v).strip().lower(), "other")
@field_validator("priority", mode="before")
@classmethod
def normalize_priority(cls, v: str) -> str:
mapped = PRIORITY_MAP.get(str(v).strip().lower())
if mapped:
return mapped
raise ValueError(f"Cannot map priority '{v}' to p0/p1/p2/p3. Valid inputs: {list(PRIORITY_MAP.keys())}")
@field_validator("customer_sentiment", mode="before")
@classmethod
def normalize_sentiment(cls, v: str) -> str:
return SENTIMENT_MAP.get(str(v).strip().lower(), "frustrated")Instructor 客户端设置
我们使用 Instructor 包装 OpenAI 客户端，为我们提供经过 Pydantic 校验的、带自动重试的 structured output。当模型返回一个我们的校验器无法归一化的值（如 "priority": "ASAP"）时，校验器抛出异常，Instructor 把错误注入 prompt，模型带着关于哪里出错的明确反馈获得另一次机会。
这就是第 2 层——比 prompt 工程（第 3 层）更强，比严格 schema 模式（第 1 层）更通用。它适用于任何能产生类 JSON 输出的模型。
instructor_client = instructor.from_openai(
OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=OPENROUTER_API_KEY,
),
mode=instructor.Mode.JSON, # works with any model that can produce JSON
)客户消息
一条任何模型都应该能分类的支持工单。客户明显很沮丧，有一个紧急期限（一小时后有演示），并且已经尝试过自助服务（重置了两次密码）。让我们看看不同的模型如何解读同一种情况。
PROMPT = """Turn this into a structured support ticket.
Customer message:
"I cannot log in since yesterday. I reset my password twice and it still says 'invalid credentials'.
I have a demo in one hour. This is ridiculous. Please fix this now."
"""API 辅助函数
call_openrouter 处理 5xx 错误上的重试，使用抖动退避。4xx（错误请求、认证失败）立即抛出。被严格模式和降级模式路径使用（第 1 层和第 3 层）。第 2 层改用 Instructor 客户端。
def call_openrouter(payload: Dict[str, Any], retries: int = 2, timeout_s: int = 25) -> Dict[str, Any]:
last = None
for _ in range(retries):
r = requests.post(
f"{BASE}/chat/completions",
headers={
"Authorization": f"Bearer {OPENROUTER_API_KEY}",
"Content-Type": "application/json",
},
json=payload,
timeout=timeout_s,
)
if 500 <= r.status_code <= 599:
last = f"{r.status_code}: {r.text[:240]}"
time.sleep(0.4 + random.random() * 0.6)
continue
if r.status_code >= 400:
raise RuntimeError(f"{r.status_code}: {r.text[:600]}")
return r.json()
raise RuntimeError(f"OpenRouter failed after retries. Last={last}")三层回退策略
第 1 层：严格 JSON Schema 模式（response_format=json_schema）
这是可用的最强选项。我们把完整的 JSON schema 直接传给 API，使用 "strict": True，让服务商在 token 生成层面强制执行。
但要注意，严格模式只约束单次 LLM 补全调用——模型的即时 token 生成被强制匹配 schema。在生产 agent 系统中，有以下几种场景会导致它失效：
- 中间发生工具调用。 模型生成一个工具调用、拿到结果，然后生成最终的结构化输出。工具结果会注入意外内容，改变模型对 schema 值的"解释"。
- 多轮交接。 如果模型 A 的结构化输出被喂给模型 B（甚至是模型 A 的另一轮）的 prompt，schema 执行只适用于每个单独生成边界，而不是整个链。
- 服务商层面的怪癖。 OpenRouter 在代理底层服务商。"严格"执行取决于服务商是否真正实现了约束解码，而不只是声称如此。
- 值仍然会漂移。 即使严格模式正常工作，你的回退模型也可能不同地解释同一个工单——分配不同的优先级、读出不同的情绪。
所以严格模式为一次调用提供格式良好的 JSON。它不能在模型之间、轮次之间或整个流水线中提供一致的判断。
第 2 层：Instructor + Pydantic 校验器
当严格模式失败时，我们用带 Pydantic 模型的 Instructor。Instructor 要求模型返回 JSON，对照 SupportTicket 校验响应（包括归一化值的 field_validator），如果任何内容失败，把校验错误注入 prompt 重试。这严格比 prompt 工程更强大，因为模型会得到关于哪里出错的明确反馈。
第 3 层：纯 prompt + 规范化（最后手段）
当连 Instructor 都无法获得有效输出时——也许模型根本不支持 JSON 模式，或者它用尽了所有重试——我们回退到纯 prompt 工程。我们在系统消息中描述 schema 约束并解析任何返回内容。然后 canonicalize() 归一化键、映射值并填充默认值。这是最弱的路径，但总比崩溃好。
STRICT_SYSTEM = "Return ONLY valid JSON. No markdown. No code fences."
def run_strict(model: str) -> Dict[str, Any]:
"""Tier 1: Strict JSON Schema mode via the provider."""
payload = {
"model": model,
"messages": [
{"role": "system", "content": STRICT_SYSTEM},
{"role": "user", "content": PROMPT},
],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "support_ticket", "strict": True, "schema": SUPPORT_TICKET_SCHEMA},
},
}
t0 = time.time()
data = call_openrouter(payload)
latency = time.time() - t0
content = data["choices"][0]["message"]["content"]
return {"model": model, "mode": "strict", "latency_s": round(latency, 3), "raw_content": content}第 2 层：Instructor 驱动的 Structured Output
这就是我们从希望模型配合转向强制执行的地方。Instructor 把 Pydantic schema 作为请求的一部分发送，解析响应，运行 field_validator（通过我们的映射归一化值），如果任何内容失败，就把校验错误注入对话重试。
max_retries=2 意味着最多 3 次总尝试（初始 + 2 次重试）。如果模型持续失败，我们落到第 3 层。
def run_instructor(model: str) -> Dict[str, Any]:
"""Tier 2: Instructor + Pydantic validation with automatic retries."""
t0 = time.time()
ticket = instructor_client.chat.completions.create(
model=model,
response_model=SupportTicket,
max_retries=2,
messages=[
{"role": "system", "content": "You are a customer support triage agent. Return a structured support ticket."},
{"role": "user", "content": PROMPT},
],
)
latency = time.time() - t0
return {
"model": model,
"mode": "instructor",
"latency_s": round(latency, 3),
"ticket": ticket,
"raw_content": ticket.model_dump_json(indent=2),
}第 3 层：纯 prompt JSON（最后手段）
当连 Instructor 都无法获得有效输出时——也许模型根本不支持 JSON 模式，或者它用尽了所有重试——我们回退到 prompt 工程。我们在系统消息中描述 schema 约束并要求模型返回纯 JSON。
这本质上是最不可靠的路径。模型可能：
- 把 JSON 包在 ```json 代码围栏中
- 嵌套对象而不是返回扁平结构
- 使用不同的键名（
subject而不是summary） - 选择允许枚举之外的值（
"High"而不是"p1"）
这就是为什么这条路径会送入 canonicalize()——紧急归一化器。
DEGRADED_SYSTEM = """Return ONLY valid JSON. No markdown. No code fences.
Return a SINGLE flat JSON object with exactly these keys:
ticket_id, summary, category, priority, customer_sentiment, repro_steps, expected_behavior, actual_behavior, suggested_next_action.
Rules:
- category must be one of: billing, login, bug, feature_request, performance, other
- priority must be one of: p0, p1, p2, p3
- customer_sentiment must be one of: calm, frustrated, angry
- repro_steps must be a JSON array of strings (even if empty)
- Do not add any other keys
"""
def run_degraded(model: str) -> Dict[str, Any]:
"""Tier 3: Prompt-only JSON — last resort."""
payload = {
"model": model,
"messages": [
{"role": "system", "content": DEGRADED_SYSTEM},
{"role": "user", "content": PROMPT},
],
}
t0 = time.time()
data = call_openrouter(payload)
latency = time.time() - t0
content = data["choices"][0]["message"]["content"]
return {"model": model, "mode": "degraded", "latency_s": round(latency, 3), "raw_content": content}健壮的 JSON 解析（用于第 3 层）
降级模式下的模型经常把它们的 JSON 包在 ```json 代码围栏中，在 JSON 前后添加评论，或返回嵌套对象。我们需要优雅地处理所有这些——因为在生产环境中，你没有机会重新提示。
def strip_code_fences(s: str) -> str:
s = s.strip()
if s.startswith("```"):
s = re.sub(r"^```[a-zA-Z]*\n?", "", s)
s = re.sub(r"\n?```$", "", s)
return s.strip()
def extract_first_json_object(text: str) -> str:
s = text.strip()
start = s.find("{")
if start == -1:
raise ValueError("No JSON object start found")
depth = 0
for i in range(start, len(s)):
if s[i] == "{":
depth += 1
elif s[i] == "}":
depth -= 1
if depth == 0:
return s[start:i+1]
raise ValueError("No complete JSON object found")
def parse_json_robust(content: str) -> Dict[str, Any]:
cleaned = strip_code_fences(content)
try:
return json.loads(cleaned)
except Exception:
return json.loads(extract_first_json_object(cleaned))最后手段的规范化（用于第 3 层）
canonicalize() 函数是模型输出的急诊室。它只在第 1 层（严格）和第 2 层（Instructor）都失败时运行，这意味着输出可能很乱。它会：
- 接受替代键名（
subject→summary、steps_taken→repro_steps） - 使用我们的归一化映射映射非标准值
- 当其他方法都失败时，从上下文推断优先级
- 为缺失字段提供合理的默认值
注意，归一化映射已经在上方定义，并且也被第 2 层中的 Pydantic 校验器使用。这个函数是即使校验器也无能为力时的兜底——通常是因为模型返回的内容严重偏离 schema，以至于 Instructor 的重试都无法修复。
def to_list(x: Any) -> List[str]:
if isinstance(x, list):
return [str(i).strip() for i in x if str(i).strip()]
if isinstance(x, str):
s = x.strip()
if not s:
return []
if "\n" in s:
parts = [p.strip(" -\t") for p in s.split("\n")]
return [p for p in parts if p]
return [s]
return []
def infer_priority(original_prompt: str, summary: str, sentiment: str, category: str) -> str:
"""Heuristic priority when the model gave us nothing usable."""
t = (original_prompt + " " + (summary or "")).lower()
s = (sentiment or "").lower()
cat = (category or "").lower()
urgent_signal = ("demo" in t and ("1 hour" in t or "one hour" in t)) or "urgent" in t or "immediate" in t
blocked_login = (cat == "login") and ("cannot log in" in t or "unable to log in" in t)
if blocked_login and ("angry" in s or urgent_signal):
return "p0"
if urgent_signal:
return "p1"
if blocked_login:
return "p1"
return "p2"
def canonicalize(raw: Dict[str, Any]) -> Dict[str, Any]:
"""Emergency normalizer for Tier 3 output. Maps alternate keys and values."""
ticket_id = str(raw.get("ticket_id") or raw.get("id") or "TCK-0000").strip()
summary = str(raw.get("summary") or raw.get("subject") or raw.get("title") or "").strip()
category_raw = str(raw.get("category") or "other").strip().lower()
category = CATEGORY_MAP.get(category_raw, "other")
sentiment_raw = str(raw.get("customer_sentiment") or raw.get("sentiment") or "frustrated").strip().lower()
sentiment = SENTIMENT_MAP.get(sentiment_raw, "frustrated")
priority_raw = str(raw.get("priority") or "").strip().lower()
priority = PRIORITY_MAP.get(priority_raw, None)
if priority is None:
priority = infer_priority(PROMPT, summary, sentiment, category)
repro_steps = to_list(raw.get("repro_steps") or raw.get("steps_taken") or raw.get("actions_taken"))
expected_behavior = str(raw.get("expected_behavior") or "").strip()
actual_behavior = str(raw.get("actual_behavior") or raw.get("current_behavior") or "").strip()
next_action = str(raw.get("suggested_next_action") or raw.get("requested_action") or "").strip()
return {
"ticket_id": ticket_id if ticket_id else "TCK-0000",
"summary": summary if summary else "Customer cannot access account",
"category": category,
"priority": priority,
"customer_sentiment": sentiment,
"repro_steps": repro_steps if repro_steps else ["Attempt login", "Reset password", "Attempt login again"],
"expected_behavior": expected_behavior if expected_behavior else "User should be able to log in successfully.",
"actual_behavior": actual_behavior if actual_behavior else "User receives invalid credentials error after password reset.",
"suggested_next_action": next_action if next_action else "Check auth logs and account lockout status; escalate if needed.",
}完整的三层流水线
run_model() 实现了完整策略：
- 第 1 层 — 尝试严格模式 → 解析 → 用 Pydantic 校验（校验器归一化）
- 第 2 层 — 尝试 Instructor → Pydantic 校验 + 归一化 → 失败时重试
- 第 3 层 — 纯 prompt → 健壮解析 →
canonicalize()→ 用 Pydantic 校验
每一层都渐进地更不可靠但更普遍兼容。在生产环境中，大多数调用应该在第 1 层成功。如果你的主模型宕机且回退不能做严格模式，第 2 层会接住。如果连那都失败，第 3 层是紧急路径。
问题不是你的主模型会不会失败——而是何时失败，以及你的系统是优雅降级还是崩溃。
def run_model(model: str) -> Dict[str, Any]:
strict_failure = None
instructor_failure = None
# ── Tier 1: Strict JSON Schema mode ──────────────────────────────
try:
out = run_strict(model)
raw = parse_json_robust(out["raw_content"])
ticket = SupportTicket.model_validate(raw)
return {
"model": model,
"mode": "strict",
"tier": 1,
"strict_failed": False,
"instructor_failed": False,
"ok": True,
"latency_s": out["latency_s"],
"priority": ticket.priority,
"category": ticket.category,
"sentiment": ticket.customer_sentiment,
"n_repro_steps": len(ticket.repro_steps),
"strict_failure": None,
"instructor_failure": None,
"note": "Tier 1: Strict JSON Schema succeeded.",
"raw_excerpt": out["raw_content"][:300],
}
except Exception as e_strict:
strict_failure = f"{type(e_strict).__name__}: {str(e_strict)[:220]}"
# ── Tier 2: Instructor + Pydantic validation ─────────────────────
try:
out2 = run_instructor(model)
ticket2 = out2["ticket"]
return {
"model": model,
"mode": "instructor",
"tier": 2,
"strict_failed": True,
"instructor_failed": False,
"ok": True,
"latency_s": out2["latency_s"],
"priority": ticket2.priority,
"category": ticket2.category,
"sentiment": ticket2.customer_sentiment,
"n_repro_steps": len(ticket2.repro_steps),
"strict_failure": strict_failure,
"instructor_failure": None,
"note": "Tier 2: Strict failed \u2192 Instructor + Pydantic validators succeeded.",
"raw_excerpt": out2["raw_content"][:300],
}
except Exception as e_instr:
instructor_failure = f"{type(e_instr).__name__}: {str(e_instr)[:220]}"
# ── Tier 3: Prompt-only + canonicalize ───────────────────────────
try:
out3 = run_degraded(model)
raw3 = parse_json_robust(out3["raw_content"])
canonical = canonicalize(raw3)
ticket3 = SupportTicket.model_validate(canonical)
return {
"model": model,
"mode": "degraded_canonicalized",
"tier": 3,
"strict_failed": True,
"instructor_failed": True,
"ok": True,
"latency_s": out3["latency_s"],
"priority": ticket3.priority,
"category": ticket3.category,
"sentiment": ticket3.customer_sentiment,
"n_repro_steps": len(ticket3.repro_steps),
"strict_failure": strict_failure,
"instructor_failure": instructor_failure,
"note": "Tier 3: Strict + Instructor failed \u2192 prompt-only + canonicalize \u2192 validated.",
"raw_excerpt": out3["raw_content"][:300],
}
except Exception as e_degraded:
return {
"model": model,
"mode": "all_failed",
"tier": 0,
"strict_failed": True,
"instructor_failed": True,
"ok": False,
"latency_s": 0,
"priority": "unknown",
"category": "unknown",
"sentiment": "unknown",
"n_repro_steps": 0,
"strict_failure": strict_failure,
"instructor_failure": instructor_failure,
"note": f"All tiers failed. Last error: {str(e_degraded)[:200]}",
"raw_excerpt": "",
}运行全部 4 个模型 × 2 次试验
我们在相同输入上运行每个模型两次，以检查：
- 跨模型漂移：不同的模型是否产生不同的优先级/情绪？
- 运行间稳定性：同一个模型两次给出相同答案吗？
- 分层分布：哪些模型在第 1 层成功，哪些需要第 2 层或第 3 层？
MODELS = [
"openai/gpt-5.2",
"qwen/qwen3-max-thinking",
"anthropic/claude-haiku-4.5",
"minimax/minimax-m2.5",
]
results = []
for m in MODELS:
for trial in range(2):
print(f" Running {m} (trial {trial+1}/2)...")
results.append(run_model(m))
print(f"\nDone: {len(results)} runs completed.")Running openai/gpt-5.2 (trial 1/2)...
Running openai/gpt-5.2 (trial 2/2)...
Running qwen/qwen3-max-thinking (trial 1/2)...
Running qwen/qwen3-max-thinking (trial 2/2)...
Running anthropic/claude-haiku-4.5 (trial 1/2)...
Running anthropic/claude-haiku-4.5 (trial 2/2)...
Running minimax/minimax-m2.5 (trial 1/2)...
Running minimax/minimax-m2.5 (trial 2/2)...
Done: 8 runs completed.并排对比：每个模型产生了什么？
这是事情变得真实的地方。注意：
- 层级：每个模型落到哪个回退级别——这告诉你什么坏了
- 优先级：
p0与p1—— 决定是有人在凌晨 2 点被传呼，还是工单等到早上 - 情绪：
angry与frustrated—— 改变自动回复的语气和升级路径 - 复现步骤数量：模型从同一条消息中提取了多少细节
- 延迟：落到更低层的真实成本
# ── Per-run detail table ──────────────────────────────────────────────
print(f"{'Model':<30} {'#':>2} {'Tier':>4} {'Mode':<25} {'Pri':>3} {'Sentiment':<12} {'Steps':>5} {'Latency':>8}")
print("\u2500" * 95)
for i, r in enumerate(results):
trial = (i % 2) + 1
print(f"{r['model']:<30} {trial:>2} {r['tier']:>4} {r['mode']:<25} {r['priority']:>3} {r['sentiment']:<12} {r['n_repro_steps']:>5} {r['latency_s']:>7.1f}s")
# ── Highlight cross-model disagreements ───────────────────────────────
priorities = {}
sentiments = {}
tiers = {}
for r in results:
short = r["model"].split("/")[-1]
priorities.setdefault(short, set()).add(r["priority"])
sentiments.setdefault(short, set()).add(r["sentiment"])
tiers.setdefault(short, set()).add(r["tier"])
print()
print("=" * 70)
print(" CROSS-MODEL DRIFT: Same customer, same schema, different judgments")
print("=" * 70)
print()
print("Priority assignments (p0 = page someone, p1 = next morning):")
for m, ps in priorities.items():
marker = " \u2713" if len(ps) == 1 else " \u26a0 UNSTABLE"
print(f" {m:<28} \u2192 {', '.join(sorted(ps))}{marker}")
print()
print("Sentiment readings (angry = escalate, frustrated = empathize):")
for m, ss in sentiments.items():
marker = " \u2713" if len(ss) == 1 else " \u26a0 UNSTABLE"
print(f" {m:<28} \u2192 {', '.join(sorted(ss))}{marker}")
print()
print("Tier distribution (which fallback level did each model land on?):")
for m, ts in tiers.items():
tier_str = ", ".join(f"Tier {t}" for t in sorted(ts))
print(f" {m:<28} \u2192 {tier_str}")
# ── Strict + Instructor failures ─────────────────────────────────────
strict_failures = [r for r in results if r["strict_failed"]]
instructor_failures = [r for r in results if r.get("instructor_failed")]
if strict_failures:
print()
print("=" * 70)
print(f" TIER 1 FAILURES: {len(strict_failures)} of {len(results)} runs failed strict mode")
print("=" * 70)
for r in strict_failures:
print(f" {r['model']}")
print(f" Strict error: {(r['strict_failure'] or '')[:120]}")
if r.get("instructor_failed"):
print(f" Instructor error: {(r.get('instructor_failure') or '')[:120]}")
print(f" Rescued by: Tier 3 (prompt-only + canonicalize)")
else:
print(f" Rescued by: Tier 2 (Instructor + Pydantic)")
# ── Latency comparison by tier ────────────────────────────────────────
print()
print("=" * 70)
print(" LATENCY BY TIER")
print("=" * 70)
for tier_num, tier_name in [(1, "Strict"), (2, "Instructor"), (3, "Degraded+canonicalize")]:
lats = [r["latency_s"] for r in results if r["tier"] == tier_num]
if lats:
print(f" Tier {tier_num} ({tier_name}): avg {sum(lats)/len(lats):.1f}s (n={len(lats)})")Model # Tier Mode Pri Sentiment Steps Latency
───────────────────────────────────────────────────────────────────────────────────────────────
openai/gpt-5.2 1 1 strict p0 angry 7 6.2s
openai/gpt-5.2 2 1 strict p0 angry 7 6.9s
qwen/qwen3-max-thinking 1 1 strict p1 frustrated 4 6.2s
qwen/qwen3-max-thinking 2 1 strict p1 frustrated 4 6.0s
anthropic/claude-haiku-4.5 1 1 strict p0 angry 7 7.0s
anthropic/claude-haiku-4.5 2 1 strict p0 angry 5 5.8s
minimax/minimax-m2.5 1 1 strict p1 frustrated 6 5.5s
minimax/minimax-m2.5 2 1 strict p0 frustrated 3 2.6s
======================================================================
CROSS-MODEL DRIFT: Same customer, same schema, different judgments
======================================================================
Priority assignments (p0 = page someone, p1 = next morning):
gpt-5.2 → p0 ✓
qwen3-max-thinking → p1 ✓
claude-haiku-4.5 → p0 ✓
minimax-m2.5 → p0, p1 ⚠ UNSTABLE
Sentiment readings (angry = escalate, frustrated = empathize):
gpt-5.2 → angry ✓
qwen3-max-thinking → frustrated ✓
claude-haiku-4.5 → angry ✓
minimax-m2.5 → frustrated ✓
Tier distribution (which fallback level did each model land on?):
gpt-5.2 → Tier 1
qwen3-max-thinking → Tier 1
claude-haiku-4.5 → Tier 1
minimax-m2.5 → Tier 1
======================================================================
LATENCY BY TIER
======================================================================
Tier 1 (Strict): avg 5.8s (n=8)原始输出样本
让我们看看模型实际返回了什么。这使漂移具体化——你可以确切地看到不同模型如何不同地格式化相同的信息，以及是哪一层接住了它们。
seen_models = []
for r in results:
if r["model"] not in seen_models:
seen_models.append(r["model"])
for model_name in seen_models:
model_runs = [r for r in results if r["model"] == model_name]
short = model_name.split("/")[-1]
r0 = model_runs[0]
print(f"{'\u2501' * 80}")
print(f" {short} | tier: {r0['tier']} | mode: {r0['mode']} | priority: {r0['priority']} | sentiment: {r0['sentiment']}")
print(f"{'\u2501' * 80}")
# Show key fields from the raw output
raw = r0["raw_excerpt"]
try:
# Try to parse (close truncated JSON if needed)
parsed = json.loads(raw + ("}" * max(0, raw.count("{") - raw.count("}"))))
except Exception:
parsed = None
if parsed:
for key in ["ticket_id", "summary", "category", "priority", "customer_sentiment", "repro_steps"]:
if key in parsed:
val = parsed[key]
if isinstance(val, list):
print(f" {key}: [{len(val)} items] {', '.join(str(v)[:40] for v in val[:3])}{'...' if len(val) > 3 else ''}")
else:
print(f" {key}: {val}")
else:
print(f" {raw[:300]}")
# Show if the two trials agreed
if len(model_runs) > 1:
r1 = model_runs[1]
diffs = []
for field in ["priority", "sentiment", "n_repro_steps", "tier"]:
if r0.get(field) != r1.get(field):
diffs.append(f"{field}: {r0.get(field)} vs {r1.get(field)}")
if diffs:
print(f" \u26a0 Trial 1 vs 2 differ: {'; '.join(diffs)}")
else:
print(f" \u2713 Both trials agree on priority, sentiment, steps, tier")
print()量化漂移分析
让我们给漂移加上数字。漂移分数综合了：
- 优先级漂移（如果模型在试验之间分配了不同的优先级，1 分）
- 步骤漂移（归一化：复现步骤计数变化的程度）
- 层级惩罚（第 1 层 0.0，第 2 层 0.25，第 3 层 0.5）
漂移分数 0 = 在第 1 层完全稳定。更高 = 更不可预测。
import pandas as pd
df = pd.DataFrame(results).copy()
# Add trial index per model
df["trial"] = df.groupby("model").cumcount() + 1
# ── Per-run summary table ────────────────────────────────────────────
summary_cols = ["model", "trial", "tier", "mode", "latency_s",
"priority", "category", "sentiment", "n_repro_steps", "note"]
print("Per-Run Results")
print("=" * 40)
display(df[summary_cols])Per-Run Results
========================================model trial tier mode latency_s priority \
0 openai/gpt-5.2 1 1 strict 6.193 p0
1 openai/gpt-5.2 2 1 strict 6.881 p0
2 qwen/qwen3-max-thinking 1 1 strict 6.182 p1
3 qwen/qwen3-max-thinking 2 1 strict 6.003 p1
4 anthropic/claude-haiku-4.5 1 1 strict 7.000 p0
5 anthropic/claude-haiku-4.5 2 1 strict 5.772 p0
6 minimax/minimax-m2.5 1 1 strict 5.471 p1
7 minimax/minimax-m2.5 2 1 strict 2.600 p0
category sentiment n_repro_steps note
0 login angry 7 Tier 1: Strict JSON Schema succeeded.
1 login angry 7 Tier 1: Strict JSON Schema succeeded.
2 login frustrated 4 Tier 1: Strict JSON Schema succeeded.
3 login frustrated 4 Tier 1: Strict JSON Schema succeeded.
4 login angry 7 Tier 1: Strict JSON Schema succeeded.
5 login angry 5 Tier 1: Strict JSON Schema succeeded.
6 login frustrated 6 Tier 1: Strict JSON Schema succeeded.
7 login frustrated 3 Tier 1: Strict JSON Schema succeeded.# ── Per-model delta: what changed between trials? ─────────────────────
def delta_summary(g: pd.DataFrame) -> pd.Series:
diffs = []
for field in ["priority", "category", "sentiment", "n_repro_steps", "mode", "tier"]:
if g[field].nunique(dropna=True) > 1:
diffs.append(field)
change = ("Changed across trials: " + ", ".join(diffs) + ".") if diffs else "Stable across both trials."
return pd.Series({
"trials": len(g),
"changed_fields": ", ".join(diffs) if diffs else "none",
"delta_note": change,
"latency_min": round(float(g["latency_s"].min()), 3),
"latency_max": round(float(g["latency_s"].max()), 3),
})
delta = df.groupby("model").apply(delta_summary, include_groups=False).reset_index()
print("\nRun-to-Run Stability (same model, same input)")
print("=" * 50)
display(delta)
# ── Drift scoring (updated for three tiers) ──────────────────────────
priority_rank = {"p0": 3, "p1": 2, "p2": 1, "p3": 0}
tier_penalty_map = {1: 0.0, 2: 0.25, 3: 0.5, 0: 1.0}
def drift_for_model(g: pd.DataFrame) -> pd.Series:
pr = g["priority"].map(priority_rank)
priority_drift = int(pr.nunique(dropna=True) > 1)
steps = pd.to_numeric(g["n_repro_steps"], errors="coerce")
steps_min = steps.min() if steps.notna().any() else None
steps_max = steps.max() if steps.notna().any() else None
steps_drift = float(steps_max - steps_min) if steps_min is not None else 0.0
steps_drift_norm = min(1.0, steps_drift / 10.0)
# Tier penalty: worst tier used across trials
worst_tier = int(g["tier"].max()) if g["tier"].notna().any() else 0
tier_penalty = tier_penalty_map.get(worst_tier, 0.5)
drift_score = priority_drift + steps_drift_norm + tier_penalty
return pd.Series({
"trials": len(g),
"tiers_used": ", ".join(f"T{int(t)}" for t in sorted(g["tier"].dropna().unique())),
"worst_tier": worst_tier,
"priority_values": ", ".join(sorted(g["priority"].dropna().unique().tolist())),
"steps_min": int(steps_min) if steps_min is not None else None,
"steps_max": int(steps_max) if steps_max is not None else None,
"priority_drift": priority_drift,
"steps_drift": round(steps_drift, 2),
"tier_penalty": tier_penalty,
"drift_score": round(drift_score, 2),
"avg_latency_s": round(float(g["latency_s"].mean()), 3),
"p95_latency_s": round(float(g["latency_s"].quantile(0.95)), 3),
})
drift = df.groupby("model").apply(drift_for_model, include_groups=False).reset_index()
print("\nDrift Score by Model (lower = more stable)")
print("=" * 50)
display(drift)Run-to-Run Stability (same model, same input)
==================================================model trials changed_fields \
0 anthropic/claude-haiku-4.5 2 n_repro_steps
1 minimax/minimax-m2.5 2 priority, n_repro_steps
2 openai/gpt-5.2 2 none
3 qwen/qwen3-max-thinking 2 none
delta_note latency_min latency_max
0 Changed across trials: n_repro_steps. 5.772 7.000
1 Changed across trials: priority, n_repro_steps. 2.600 5.471
2 Stable across both trials. 6.193 6.881
3 Stable across both trials. 6.003 6.182Drift Score by Model (lower = more stable)
==================================================model trials tiers_used worst_tier priority_values \
0 anthropic/claude-haiku-4.5 2 T1 1 p0
1 minimax/minimax-m2.5 2 T1 1 p0, p1
2 openai/gpt-5.2 2 T1 1 p0
3 qwen/qwen3-max-thinking 2 T1 1 p1
steps_min steps_max priority_drift steps_drift tier_penalty \
0 5 7 0 2.0 0.0
1 3 6 1 3.0 0.0
2 7 7 0 0.0 0.0
3 4 4 0 0.0 0.0
drift_score avg_latency_s p95_latency_s
0 0.2 6.386 6.939
1 1.3 4.035 5.327
2 0.0 6.537 6.847
3 0.0 6.093 6.173# ── Overall summary ──────────────────────────────────────────────────
tier_counts = df["tier"].value_counts().to_dict()
overall = {
"models_tested": int(df["model"].nunique()),
"total_runs": int(len(df)),
"tier_1_runs": int(tier_counts.get(1, 0)),
"tier_2_runs": int(tier_counts.get(2, 0)),
"tier_3_runs": int(tier_counts.get(3, 0)),
"all_failed_runs": int(tier_counts.get(0, 0)),
"avg_latency_s": round(float(df["latency_s"].mean()), 3),
"p95_latency_s": round(float(df["latency_s"].quantile(0.95)), 3),
"priority_values_seen": sorted(df["priority"].dropna().unique().tolist()),
"sentiment_values_seen": sorted(df["sentiment"].dropna().unique().tolist()),
"steps_range": (
int(df["n_repro_steps"].min()) if df["n_repro_steps"].notna().any() else None,
int(df["n_repro_steps"].max()) if df["n_repro_steps"].notna().any() else None,
),
}
print("Overall Summary")
print("=" * 40)
for k, v in overall.items():
print(f" {k}: {v}")
# Highlight the key finding
pri_set = set(df["priority"].dropna().unique())
sent_set = set(df["sentiment"].dropna().unique())
if len(pri_set) > 1 or len(sent_set) > 1:
print()
print("\u26a0 KEY FINDING: Same customer message, same schema, but models disagree:")
if len(pri_set) > 1:
print(f" Priority: {sorted(pri_set)} \u2192 This changes who gets paged and when.")
if len(sent_set) > 1:
print(f" Sentiment: {sorted(sent_set)} \u2192 This changes the auto-response tone and escalation path.")Overall Summary
========================================
models_tested: 4
total_runs: 8
tier_1_runs: 8
tier_2_runs: 0
tier_3_runs: 0
all_failed_runs: 0
avg_latency_s: 5.763
p95_latency_s: 6.958
priority_values_seen: ['p0', 'p1']
sentiment_values_seen: ['angry', 'frustrated']
steps_range: (3, 7)
⚠ KEY FINDING: Same customer message, same schema, but models disagree:
Priority: ['p0', 'p1'] → This changes who gets paged and when.
Sentiment: ['angry', 'frustrated'] → This changes the auto-response tone and escalation path.