Agent

当类型安全遇上 AI Agent Pydantic AI 是 Pydantic 团队在 2025 年推出的 Agent 框架，核心理念是将 Python 的类型系统引入 Agent 开发。在大多数 Agent 框架中，LLM 输出的结构化数据依赖运行时验证——如果 LLM 输出不符合预期，你只能在运行时发现问题。Pydantic AI 通过编译时类型检查 + 运行时 Schema 验证，将这类问题消灭在开发阶段。 核心设计 类型驱动的 Agent 定义 from pydantic_ai import Agent, RunContext from pydantic import BaseModel, Field from typing import Literal from dataclasses import dataclass # 定义结构化输出 class ResearchReport(BaseModel): title: str = Field(description="报告标题") summary: str = Field(description="执行摘要，不超过200字") key_findings: list[str] = Field(description="关键发现列表") confidence: float = Field(ge=0, le=1, description="置信度 0-1") sources: list[str] = Field(description="信息来源 URL 列表") recommendation: Literal["buy", "hold", "sell"] = Field(description="投资建议") # 定义依赖类型 @dataclass class ResearchDeps: api_keys: dict[str, str] database_url: str max_sources: int = 10 # 创建 Agent research_agent = Agent( model="openai:gpt-4o", deps_type=ResearchDeps, output_type=ResearchReport, # 类型安全的输出 system_prompt="你是一名专业的研究分析师..." ) # 类型安全的工具定义 @research_agent.tool async def search_database(ctx: RunContext[ResearchDeps], query: str, limit: int = 10) -> list[dict]: """ 搜索内部数据库。 Args: query: 搜索查询 limit: 返回结果数量上限 """ # ctx.deps 是 ResearchDeps 类型，IDE 自动补全 async with httpx.AsyncClient() as client: response = await client.post( f"{ctx.deps.database_url}/search", json={"query": query, "limit": min(limit, ctx.deps.max_sources)}, headers={"Authorization": f"Bearer {ctx.deps.api_keys['database']}"} ) return response.json()["results"] @research_agent.tool async def fetch_web_page(ctx: RunContext[ResearchDeps], url: str) -> str: """获取网页内容。""" async with httpx.AsyncClient() as client: response = await client.get(url, timeout=10) return response.text[:5000] # 限制内容长度 # 运行 Agent —— 输出类型自动验证 result = await research_agent.run( "分析 2026 年 AGI 芯片市场", deps=ResearchDeps( api_keys={"database": "xxx"}, database_url="https://api.example.com" ) ) # result.output 是 ResearchReport 类型，IDE 完整支持 print(result.output.title) print(result.output.confidence) print(result.output.recommendation) # 如果 LLM 输出不符合 Schema，Pydantic 会抛出 # ValidationError，而非静默返回错误数据 核心特性 1. 结构化输出保证 from pydantic_ai import Agent from pydantic import BaseModel, validator from typing import Optional class CodeReview(BaseModel): file_name: str issues: list["CodeIssue"] overall_rating: int = Field(ge=1, le=10, description="整体评分 1-10") approved: bool @validator("file_name") def validate_filename(cls, v): if not v.endswith((".py", ".js", ".ts")): raise ValueError("仅支持 .py/.js/.ts 文件") return v class CodeIssue(BaseModel): line_number: int = Field(ge=1) severity: Literal["error", "warning", "info"] message: str suggestion: Optional[str] = None review_agent = Agent( model="anthropic:claude-4-sonnet", output_type=CodeReview, system_prompt="你是代码审查专家，分析代码质量问题。" ) # Agent 输出保证符合 CodeReview Schema result = await review_agent.run("审查以下代码: ...") review: CodeReview = result.output # 类型保证 2. 多模型支持与切换 from pydantic_ai.models import Model from pydantic_ai.models.openai import OpenAIModel from pydantic_ai.models.anthropic import AnthropicModel from pydantic_ai.models.groq import GroqModel # 环境感知模型选择 def get_model() -> Model: env = os.getenv("ENVIRONMENT", "dev") if env == "production": return OpenAIModel("gpt-4o", api_key=os.getenv("OPENAI_API_KEY")) elif env == "staging": return AnthropicModel("claude-4-sonnet", api_key=os.getenv("ANTHROPIC_API_KEY")) elif env == "dev": return GroqModel("llama-4-70b", api_key=os.getenv("GROQ_API_KEY")) else: # 使用 Ollama 本地模型 return OpenAIModel( "qwen3:72b", base_url="http://localhost:11434/v1", api_key="ollama" ) agent = Agent( model=get_model(), output_type=ResearchReport, system_prompt="..." ) # 运行时切换模型 result = await agent.run("...", model=AnthropicModel("claude-4-opus")) 3. 流式输出 from pydantic_ai import Agent from pydantic_ai.messages import Part streaming_agent = Agent( model="openai:gpt-4o", system_prompt="你是技术写作助手..." ) # 流式输出 async with streaming_agent.run_stream("写一篇关于 AGI 的短文") as result: async for message in result.stream_text(delta=True): print(message, end="", flush=True) # 流式结构化输出（2026 新特性） class ArticleOutline(BaseModel): title: str sections: list[str] estimated_words: int async with streaming_agent.run_stream( "生成文章大纲", output_type=ArticleOutline ) as result: # 部分结果流式返回 async for partial in result.stream_structured(): print(f"当前大纲: {partial.title} ({len(partial.sections)} 节)") final = await result.get_output() print(f"最终大纲: {final}") 4. 依赖注入 from pydantic_ai import Agent, RunContext from dataclasses import dataclass import httpx import asyncio @dataclass class AppDeps: http_client: httpx.AsyncClient db_pool: asyncpg.Pool cache: redis.Redis config: dict # 所有工具共享同一个依赖上下文 agent = Agent("openai:gpt-4o", deps_type=AppDeps) @agent.tool async def get_user_profile(ctx: RunContext[AppDeps], user_id: int) -> dict: # 从缓存获取 cached = await ctx.deps.cache.get(f"user:{user_id}") if cached: return json.loads(cached) # 从数据库获取 async with ctx.deps.db_pool.acquire() as conn: user = await conn.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) # 写入缓存 await ctx.deps.cache.setex(f"user:{user_id}", 3600, json.dumps(dict(user))) return dict(user) @agent.tool async def call_external_api(ctx: RunContext[AppDeps], url: str) -> dict: response = await ctx.deps.http_client.get(url) return response.json() # 运行时注入依赖 async def main(): async with httpx.AsyncClient() as http_client: db_pool = await asyncpg.create_pool(DATABASE_URL) cache = redis.Redis() deps = AppDeps( http_client=http_client, db_pool=db_pool, cache=cache, config={"max_results": 50} ) result = await agent.run("查询用户 123 的信息", deps=deps) print(result.output) 5. Agent 组合 from pydantic_ai import Agent # 子 Agent：各自有独立的输出类型 research_agent = Agent( model="openai:gpt-4o", output_type=ResearchReport, system_prompt="你是研究员..." ) writing_agent = Agent( model="anthropic:claude-4-sonnet", output_type=Article, system_prompt="你是技术写手..." ) review_agent = Agent( model="openai:gpt-4o", output_type=ReviewResult, system_prompt="你是审稿人..." ) # 主 Agent 协调子 Agent class PipelineResult(BaseModel): research: ResearchReport article: Article review: ReviewResult final_status: str pipeline_agent = Agent( model="openai:gpt-4o", output_type=PipelineResult, system_prompt="你是项目经理，协调多个子任务..." ) # 主 Agent 可以委派给子 Agent @pipeline_agent.tool async def run_research(ctx: RunContext, topic: str) -> ResearchReport: result = await research_agent.run(topic, deps=ctx.deps) return result.output @pipeline_agent.tool async def run_writing(ctx: RunContext, research: ResearchReport) -> Article: result = await writing_agent.run( f"基于以下研究撰写文章: {research.model_dump_json()}", deps=ctx.deps ) return result.output 评估与测试 from pydantic_ai.evals import TestCase, Evaluator # 定义测试用例 test_cases = [ TestCase( input="分析苹果公司 2026 年 Q1 财报", expected_output=ResearchReport( title="苹果 2026 Q1 财报分析", summary="...", key_findings=["营收增长 15%", "服务业务创新高"], confidence=0.9, sources=["https://investor.apple.com"], recommendation="buy" ) ), # 更多测试用例... ] # 评估器 class ResearchEvaluator(Evaluator): async def evaluate(self, output: ResearchReport, expected: ResearchReport) -> float: score = 0.0 # 检查结构化字段 if output.recommendation == expected.recommendation: score += 0.3 # 检查发现重叠度 overlap = len(set(output.key_findings) & set(expected.key_findings)) score += 0.4 * (overlap / max(len(expected.key_findings), 1)) # 检查置信度合理性 if abs(output.confidence - expected.confidence) < 0.2: score += 0.3 return score # 运行评估 results = await Evaluator.run( agent=research_agent, test_cases=test_cases, evaluator=ResearchEvaluator() ) print(f"平均得分: {results.avg_score}") print(f"通过率: {results.pass_rate}") 框架对比 特性 Pydantic AI LangChain CrewAI smolagents 类型安全 ⭐⭐⭐⭐⭐ ⭐⭐ ⭐⭐⭐ ⭐⭐ 结构化输出 编译时+运行时 运行时 运行时 无 学习曲线 中 高 低 低 IDE 支持 完整 部分 部分 部分 依赖注入 内置 无 无 无 流式输出 文本+结构化 文本 文本 文本 测试框架 内置 第三方 无 无 多模型 完整 完整 完整 HF 优先 性能基准 指标 Pydantic AI LangChain CrewAI Cold Start 0.2s 0.8s 0.5s 简单调用 1.1s 1.5s 1.8s 结构化输出 1.3s 1.8s 2.1s 内存占用 85MB 256MB 312MB Schema 验证开销 <50ms 100-200ms N/A 适用场景 最适合 企业级应用：类型安全是生产环境的刚需 API 后端：结构化输出直接映射 API 响应 数据管道：类型保证数据处理链路的可靠性 团队协作：类型系统作为 Agent 接口契约 不太适合 快速原型：类型定义增加了前期开发量 创意类任务：非结构化输出场景下类型约束是负担 复杂 Agent 图：不支持状态机编排 总结 Pydantic AI 在 2026 年的 Agent 框架竞争中找到了独特的定位：类型安全。这不是一个噱头——在实际的企业开发中，LLM 输出的不可预测性是最大的痛点之一。Pydantic AI 通过编译时类型检查、运行时 Schema 验证、依赖注入、内置测试框架，为 Agent 开发带来了真正的工程严谨性。 ...

微软的 AI 编排答卷 Semantic Kernel（SK）是微软在 2023 年初推出的 AI 编排框架，定位类似于"AI 版的 .NET CLR"。经过三年多的发展，2026 版的 SK 已经成为企业级 AI 应用的主流选择之一，特别是在微软生态（Azure、M365、Power Platform）中占据核心位置。 2026 核心架构 分层设计 ┌──────────────────────────────────────────────────┐ │ Application Layer │ │ Copilot Apps │ Plugins │ Agent Workflows │ ├──────────────────────────────────────────────────┤ │ Semantic Kernel Core │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │ │ Kernel │ │ Planner │ │ Agent Framework │ │ │ │ (上下文) │ │ (规划器) │ │ (多Agent编排) │ │ │ └──────────┘ └──────────┘ └──────────────────┘ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │ │ Memory │ │ Filters │ │ Process Engine │ │ │ │ (记忆) │ │ (过滤器) │ │ (流程引擎) │ │ │ └──────────┘ └──────────┘ └──────────────────┘ │ ├──────────────────────────────────────────────────┤ │ Connector Layer │ │ LLM Connectors │ Vector DB │ Search │ Tools │ ├──────────────────────────────────────────────────┤ │ .NET / Python / Java │ └──────────────────────────────────────────────────┘ 多语言支持 2026 年 SK 的语言支持矩阵： ...

引言 Agent 的推理过程往往是漫长的等待——用户盯着加载动画，不知道 Agent 在做什么。流式响应把"等待结果"变成"实时观察思考"，是 Agent 用户体验的关键升级。2026年，三种流式协议各有优劣，选型不当会导致体验降级或工程复杂度爆炸。 一、三种协议对比 核心特性矩阵 特性 SSE WebSocket gRPC Stream 通信方向 服务器→客户端（单向） 双向 双向 底层协议 HTTP/1.1 或 HTTP/2 HTTP 升级 HTTP/2 数据格式 文本（text/event-stream） 文本/二进制 Protobuf（二进制） 自动重连 内置 需手动实现 需手动实现 浏览器支持 原生 EventSource 原生 WebSocket 需 gRPC-Web 代理/CDN兼容 优秀 良好 较差 连接数限制 浏览器6个/域名 无限制 无限制 序列化效率 低（文本） 中 高（Protobuf） 移动端友好 高 中 低 Agent 场景适配分析 Agent 流式需求频谱： 单向输出流 双向交互流 （LLM→用户） （用户↔Agent） │ │ │ ┌───────────┐ │ │ │ SSE 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ WebSocket │ │ │ │ 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ gRPC 最佳 │ │ │ └───────────┘ │ 简单聊天 ←─────────────→ 复杂多Agent 低延迟 ←─────────────→ 高吞吐 Web前端 ←─────────────→ 微服务后端 二、SSE 实现方案 2.1 服务端 from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() class AgentStreamEvent: """Agent 流式事件类型""" THINKING = "thinking" # Agent 思考中 TOOL_CALL = "tool_call" # 工具调用 TOOL_RESULT = "tool_result" # 工具结果 CONTENT = "content" # 内容输出 ERROR = "error" # 错误 DONE = "done" # 完成 async def agent_stream_generator( query: str, session_id: str ): """Agent SSE 流式生成器""" try: # 1. 发送思考状态 yield _format_sse(AgentStreamEvent.THINKING, { "message": "正在分析您的请求...", "session_id": session_id }) # 2. Agent 推理（流式 LLM 输出） async for chunk in agent.think_stream(query): if chunk.type == "tool_call": yield _format_sse(AgentStreamEvent.TOOL_CALL, { "tool": chunk.tool_name, "args": chunk.tool_args, "thinking": chunk.reasoning }) # 3. 工具执行 result = await agent.execute_tool(chunk.tool_call) yield _format_sse(AgentStreamEvent.TOOL_RESULT, { "tool": chunk.tool_name, "result": result.summary, "duration_ms": result.duration_ms }) elif chunk.type == "content": yield _format_sse(AgentStreamEvent.CONTENT, { "text": chunk.text, "tokens_so_far": chunk.token_count }) # 4. 完成 yield _format_sse(AgentStreamEvent.DONE, { "session_id": session_id, "total_tokens": agent.total_tokens, "duration_ms": agent.total_duration_ms }) except Exception as e: yield _format_sse(AgentStreamEvent.ERROR, { "message": str(e), "session_id": session_id }) def _format_sse(event_type: str, data: dict) -> str: return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" @app.post("/api/agent/chat") async def chat(request: ChatRequest): return StreamingResponse( agent_stream_generator(request.query, request.session_id), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx 禁用缓冲 } ) 2.2 客户端 class AgentSSEClient { private eventSource: EventSource | null = null; private reconnectAttempts = 0; private maxReconnects = 3; connect(query: string, sessionId: string) { // 使用 fetch POST + ReadableStream（EventSource 仅支持 GET） fetch('/api/agent/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ query, sessionId }), }).then(response => { const reader = response.body!.getReader(); const decoder = new TextDecoder(); let buffer = ''; const read = () => { reader.read().then(({ done, value }) => { if (done) return; buffer += decoder.decode(value, { stream: true }); const events = buffer.split('\n\n'); buffer = events.pop() || ''; events.forEach(raw => this.handleEvent(raw)); read(); }); }; read(); }); } private handleEvent(raw: string) { const lines = raw.split('\n'); let event = 'message'; let data = ''; lines.forEach(line => { if (line.startsWith('event: ')) event = line.slice(7); if (line.startsWith('data: ')) data = line.slice(6); }); const parsed = JSON.parse(data); switch (event) { case 'thinking': this.onThinking?.(parsed); break; case 'tool_call': this.onToolCall?.(parsed); break; case 'tool_result': this.onToolResult?.(parsed); break; case 'content': this.onContent?.(parsed.text); break; case 'error': this.onError?.(parsed); break; case 'done': this.onDone?.(parsed); break; } } } 三、WebSocket 实现方案 3.1 服务端 from fastapi import WebSocket, WebSocketDisconnect class AgentConnectionManager: """Agent WebSocket 连接管理""" def __init__(self): self.active: dict[str, WebSocket] = {} # session_id → WebSocket self.agent_tasks: dict[str, asyncio.Task] = {} async def connect(self, ws: WebSocket, session_id: str): await ws.accept() self.active[session_id] = ws logger.info(f"WebSocket connected: {session_id}") async def disconnect(self, session_id: str): if session_id in self.active: del self.active[session_id] if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() del self.agent_tasks[session_id] async def handle_session(self, ws: WebSocket, session_id: str): """处理 WebSocket 会话""" await self.connect(ws, session_id) try: while True: # 接收客户端消息 message = await ws.receive_json() if message["type"] == "chat": # 启动 Agent 任务 task = asyncio.create_task( self._run_agent(ws, session_id, message["content"]) ) self.agent_tasks[session_id] = task elif message["type"] == "interrupt": # 用户中断当前 Agent 执行 if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() await ws.send_json({ "type": "interrupted", "session_id": session_id }) elif message["type"] == "feedback": # 用户实时反馈（Human-in-the-loop） await self._handle_feedback(session_id, message) except WebSocketDisconnect: await self.disconnect(session_id) async def _run_agent(self, ws: WebSocket, session_id: str, query: str): """运行 Agent 并通过 WebSocket 推送更新""" try: async for event in agent.run_stream(query): await ws.send_json({ "type": event.type, "data": event.data, "timestamp": time.time() }) except asyncio.CancelledError: logger.info(f"Agent task cancelled: {session_id}") except Exception as e: await ws.send_json({ "type": "error", "data": {"message": str(e)} }) manager = AgentConnectionManager() @app.websocket("/ws/agent/{session_id}") async def websocket_endpoint(ws: WebSocket, session_id: str): await manager.handle_session(ws, session_id) 3.2 客户端 class AgentWSClient { private ws: WebSocket | null = null; private messageQueue: string[] = []; connect(sessionId: string) { this.ws = new WebSocket(`wss://api.example.com/ws/agent/${sessionId}`); this.ws.onopen = () => { // 发送排队消息 this.messageQueue.forEach(msg => this.ws!.send(msg)); this.messageQueue = []; }; this.ws.onmessage = (event) => { const data = JSON.parse(event.data); this.handleMessage(data); }; this.ws.onclose = () => { // 自动重连 setTimeout(() => this.connect(sessionId), 3000); }; } sendChat(content: string) { const msg = JSON.stringify({ type: 'chat', content }); if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(msg); } else { this.messageQueue.push(msg); } } interrupt() { this.ws?.send(JSON.stringify({ type: 'interrupt' })); } } 四、gRPC 流式方案 4.1 Proto 定义 // agent.proto syntax = "proto3"; service AgentService { // 服务端流式：Agent → 客户端 rpc ChatStream(ChatRequest) returns (stream ChatResponse); // 双向流式：支持实时交互 rpc ChatBidirectional(stream ChatMessage) returns (stream ChatResponse); } message ChatRequest { string session_id = 1; string query = 2; map<string, string> metadata = 3; } message ChatMessage { string session_id = 1; MessageType type = 2; // CHAT, INTERRUPT, FEEDBACK string content = 3; } message ChatResponse { ResponseType type = 1; // THINKING, TOOL_CALL, CONTENT, DONE, ERROR string session_id = 2; bytes data = 3; // JSON 编码的事件数据 int64 timestamp = 4; } enum MessageType { CHAT = 0; INTERRUPT = 1; FEEDBACK = 2; } enum ResponseType { THINKING = 0; TOOL_CALL = 1; TOOL_RESULT = 2; CONTENT = 3; DONE = 4; ERROR = 5; } 4.2 服务端实现 import grpc from concurrent import futures class AgentServicer(agent_pb2_grpc.AgentServiceServicer): def ChatStream(self, request, context): """服务端流式：逐条推送 Agent 事件""" try: for event in agent.run(request.query): response = agent_pb2.ChatResponse( type=self._map_event_type(event.type), session_id=request.session_id, data=json.dumps(event.data).encode(), timestamp=int(time.time()) ) yield response except Exception as e: yield agent_pb2.ChatResponse( type=agent_pb2.ERROR, data=json.dumps({"error": str(e)}).encode() ) def ChatBidirectional(self, request_iterator, context): """双向流式：支持中断和实时反馈""" session = None for message in request_iterator: if message.type == agent_pb2.CHAT: # 启动 Agent 执行 for event in agent.run(message.content): if not context.is_active(): break yield agent_pb2.ChatResponse( type=self._map_event_type(event.type), data=json.dumps(event.data).encode() ) elif message.type == agent_pb2.INTERRUPT: agent.interrupt() yield agent_pb2.ChatResponse( type=agent_pb2.DONE, data=json.dumps({"reason": "interrupted"}).encode() ) 五、性能基准测试 测试环境 服务端：4 vCPU / 16GB RAM / Python 3.12 / FastAPI 客户端：1000 并发连接 消息大小：平均 200 bytes / 消息 持续时间：5 分钟 结果对比 指标 SSE WebSocket gRPC Stream 最大并发连接 10,000 50,000 30,000 消息延迟 P50 12ms 8ms 5ms 消息延迟 P95 45ms 25ms 12ms 消息延迟 P99 120ms 60ms 30ms 吞吐量 (msg/s) 50,000 200,000 150,000 内存/连接 32KB 48KB 24KB CPU 利用率 (1k连接) 35% 28% 22% 带宽效率 基准 +15% +40% 六、选型决策树 ┌─────────────────┐ │ 是否需要双向通信？│ └────────┬────────┘ │ ┌──────────────┴──────────────┐ │ 否 │ 是 ▼ ▼ ┌────────────────┐ ┌──────────────────┐ │ 是否是微服务 │ │ 是否是浏览器前端？│ │ 内部通信？ │ └────────┬─────────┘ └───────┬────────┘ │ │ ┌───────┴───────┐ ┌───────┴───────┐ │ 否 │ 是 │ 是 │ 否 ▼ ▼ ▼ ▼ ┌──────────┐ ┌────────────┐ ┌──────┐ ┌────────┐ │ gRPC │ │ WebSocket │ │gRPC │ │ SSE │ │ Stream │ │ │ └──────┘ └────────┘ └──────────┘ └────────────┘ 场景推荐 场景 推荐协议 原因 Web 聊天界面 SSE 原生支持、简单、自动重连 实时协作编辑 WebSocket 双向低延迟 移动 App SSE 移动网络友好、自动重连 微服务间 Agent 通信 gRPC 高效序列化、强类型 多 Agent 系统 gRPC 流式 RPC 适配 Agent 通信 简单 LLM 问答 SSE 单向流足够、实现简单 Human-in-the-loop WebSocket 需要双向交互 大规模推送 SSE CDN 兼容、连接效率高 七、生产环境关键配置 Nginx SSE 配置 location /api/agent/chat { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; # 关键：禁用缓冲 proxy_cache off; proxy_read_timeout 300s; # 长连接超时 chunked_transfer_encoding on; } WebSocket 配置 location /ws/agent/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 86400; # 24小时 proxy_send_timeout 86400; } 八、流式架构 Checklist □ 协议选型基于通信方向和客户端类型 □ SSE 禁用代理缓冲（proxy_buffering off） □ WebSocket 实现心跳和重连机制 □ gRPC 配置 keepalive 和流控 □ 消息序列化使用高效格式（JSON/Protobuf） □ 背压机制防止慢客户端拖垮服务端 □ 连接超时和最大连接数限制 □ 流式错误不中断连接，通过事件传递 □ 客户端实现优雅降级（流式不可用时回退轮询） □ 监控流式连接的延迟和消息丢失率 结语 流式响应是 Agent 从"工具"到"伙伴"的关键体验升级。协议选型没有银弹：SSE 简单可靠适合 Web 场景，WebSocket 灵活双向适合交互场景，gRPC 高效强类型适合微服务场景。理解你的通信模式——是单向输出还是双向交互——然后选择最适合的工具。在 Agent 时代，好的流式架构让用户感觉 Agent 在"思考"，而不是在"卡住"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Agent 的非确定性使得"感觉更好"不能作为决策依据。一个 Prompt 的微调可能提升某类任务的表现，却悄悄损害了另一类。2026年，A/B 测试已成为 Agent 优化的科学方法——用数据说话，用统计检验做决策。 一、Agent A/B 测试的特殊性 与传统 Web A/B 测试不同，Agent A/B 测试面临独特挑战： 维度 传统 A/B 测试 Agent A/B 测试 指标 点击率、转化率 输出质量、任务完成率、用户满意度 测量 确定性（点击=1/不点击=0） 非确定性（同一输入可能不同输出） 变量 UI 元素 Prompt、模型、工具、温度 噪声 低 高（LLM 输出方差大） 样本量 百万级 千级（成本限制） 指标延迟 即时 分钟级（需要完整执行） 二、实验设计框架 2.1 假设构建 @dataclass class ExperimentHypothesis: """实验假设""" name: str description: str independent_variable: str # 自变量（如 temperature） control_value: any # 对照组值（如 0.3） treatment_value: any # 实验组值（如 0.5） dependent_variables: list[str] # 因变量（如 task_completion_rate） expected_effect: str # 预期效果 min_detectable_effect: float # 最小可检测效应 (MDE) statistical_power: float # 统计功效 (通常 0.8) significance_level: float # 显著性水平 (通常 0.05) def required_sample_size(self) -> int: """计算所需样本量""" # 基于双比例检验的样本量计算 p1 = self.baseline_rate # 基线成功率 p2 = p1 + self.min_detectable_effect # 预期成功率 z_alpha = 1.96 # α=0.05 z_beta = 0.84 # power=0.8 n = ( (z_alpha * (2*p1*(1-p1))**0.5 + z_beta * (p1*(1-p1) + p2*(1-p2))**0.5) ** 2 ) / (p2 - p1) ** 2 return int(n) + 1 # 示例 hypothesis = ExperimentHypothesis( name="temperature_optimization", description="将 temperature 从 0.3 调至 0.5，预期能提升创意写作任务的用户满意度", independent_variable="temperature", control_value=0.3, treatment_value=0.5, dependent_variables=["user_satisfaction", "task_completion_rate"], expected_effect="满意度提升 5%", min_detectable_effect=0.05, statistical_power=0.8, significance_level=0.05, baseline_rate=0.75 # 当前满意度 75% ) # 所需样本量 ≈ 2,435 per group 2.2 实验配置 @dataclass class ExperimentConfig: experiment_id: str name: str hypothesis: ExperimentHypothesis traffic_allocation: float # 实验占总流量比例 (0-1) control_split: float # 对照组在实验流量中的比例 (通常 0.5) targeting_rules: list[Rule] # 目标用户筛选 duration_days: int # 预计运行天数 metrics: list[Metric] # 追踪指标 guardrail_metrics: list[Metric] # 护栏指标（不可恶化） early_stop_rules: list[Rule] # 提前停止规则 cost_budget: float # 实验成本预算 # 护栏指标示例 GUARDRAIL_METRICS = [ Metric(name="error_rate", type="counter", max_threshold=0.05), Metric(name="p95_latency", type="histogram", max_threshold_ms=10000), Metric(name="cost_per_request", type="histogram", max_threshold=0.15), Metric(name="toxic_output_rate", type="counter", max_threshold=0.01), ] 三、流量分配系统 class ExperimentRouter: """实验流量路由""" def __init__(self, redis_client): self.redis = redis_client async def assign( self, user_id: str, agent_name: str ) -> VariantAssignment: """为用户分配实验变体""" # 1. 获取活跃实验 experiments = await self._get_active_experiments(agent_name) for exp in experiments: # 2. 检查目标规则 if not self._matches_targeting(user_id, exp.targeting_rules): continue # 3. 检查是否已分配 existing = await self._get_assignment(user_id, exp.experiment_id) if existing: return existing # 保持一致性 # 4. 一致性哈希分配 bucket = self._hash_bucket(user_id, exp.experiment_id) # 5. 决定是否进入实验 if bucket < exp.traffic_allocation: # 在实验内部分配对照组/实验组 inner_bucket = self._hash_bucket( f"{user_id}:{exp.experiment_id}", "inner" ) if inner_bucket < exp.control_split: variant = "control" else: variant = "treatment" else: variant = "excluded" # 不参与实验 assignment = VariantAssignment( experiment_id=exp.experiment_id, user_id=user_id, variant=variant, config=exp.get_variant_config(variant), assigned_at=datetime.now() ) await self._save_assignment(assignment) return assignment # 没有匹配的实验 return VariantAssignment(variant="default", config={}) def _hash_bucket(self, key: str, salt: str = "") -> float: """一致性哈希，返回 0-1 之间的值""" h = hashlib.sha256(f"{key}:{salt}".encode()).hexdigest() return int(h[:8], 16) / 0xFFFFFFFF 四、指标收集与统计检验 4.1 指标收集器 class ExperimentMetricsCollector: """实验指标收集器""" async def record( self, experiment_id: str, user_id: str, variant: str, metrics: dict ): """记录单次实验观测""" event = { "experiment_id": experiment_id, "user_id": user_id, "variant": variant, "timestamp": time.time(), **metrics # task_completed, satisfaction_score, latency_ms, tokens_used, cost } # 写入时序数据库 await self.influxdb.write( measurement="experiment_events", tags={"experiment_id": experiment_id, "variant": variant}, fields=metrics, timestamp=event["timestamp"] ) async def aggregate( self, experiment_id: str, metric_name: str ) -> dict: """聚合实验指标""" return { "control": await self._compute_stats(experiment_id, "control", metric_name), "treatment": await self._compute_stats(experiment_id, "treatment", metric_name) } async def _compute_stats( self, exp_id: str, variant: str, metric: str ) -> MetricStats: values = await self.influxdb.query( f'SELECT "{metric}" FROM "experiment_events" ' f'WHERE "experiment_id" = \'{exp_id}\' ' f'AND "variant" = \'{variant}\'' ) return MetricStats( n=len(values), mean=statistics.mean(values), std=statistics.stdev(values) if len(values) > 1 else 0, median=statistics.median(values), p25=np.percentile(values, 25), p75=np.percentile(values, 75), p95=np.percentile(values, 95), ) 4.2 统计检验 from scipy import stats import numpy as np class StatisticalTester: """统计显著性检验""" def test_proportion( self, control_successes: int, control_total: int, treatment_successes: int, treatment_total: int, alpha: float = 0.05 ) -> TestResult: """比例检验（用于完成率等二值指标）""" # 卡方检验 contingency = [ [control_successes, control_total - control_successes], [treatment_successes, treatment_total - treatment_successes] ] chi2, p_value, _, _ = stats.chi2_contingency(contingency) # 效应量 p_control = control_successes / control_total p_treatment = treatment_successes / treatment_total effect_size = p_treatment - p_control # 置信区间 se = np.sqrt(p_control*(1-p_control)/control_total + p_treatment*(1-p_treatment)/treatment_total) ci_lower = effect_size - 1.96 * se ci_upper = effect_size + 1.96 * se return TestResult( test="chi_square", p_value=p_value, significant=p_value < alpha, effect_size=effect_size, confidence_interval=(ci_lower, ci_upper), control_rate=p_control, treatment_rate=p_treatment, interpretation=self._interpret( p_value, alpha, effect_size, p_control, p_treatment ) ) def test_continuous( self, control_values: list[float], treatment_values: list[float], alpha: float = 0.05 ) -> TestResult: """连续值检验（用于满意度分数、延迟等）""" # 正态性检验 _, p_normal_ctrl = stats.shapiro(control_values) _, p_normal_treat = stats.shapiro(treatment_values) if p_normal_ctrl > 0.05 and p_normal_treat > 0.05: # 正态分布：使用 t 检验 statistic, p_value = stats.ttest_ind( control_values, treatment_values, equal_var=False # Welch's t-test ) test_name = "welch_t_test" else: # 非正态：使用 Mann-Whitney U 检验 statistic, p_value = stats.mannwhitneyu( control_values, treatment_values, alternative='two-sided' ) test_name = "mann_whitney_u" # 效应量 (Cohen's d) pooled_std = np.sqrt( ((len(control_values)-1) * np.var(control_values, ddof=1) + (len(treatment_values)-1) * np.var(treatment_values, ddof=1)) / (len(control_values) + len(treatment_values) - 2) ) cohens_d = (np.mean(treatment_values) - np.mean(control_values)) / pooled_std return TestResult( test=test_name, p_value=p_value, significant=p_value < alpha, effect_size=cohens_d, control_mean=np.mean(control_values), treatment_mean=np.mean(treatment_values), interpretation=self._interpret_continuous( p_value, alpha, cohens_d, np.mean(control_values), np.mean(treatment_values) ) ) def _interpret(self, p_value, alpha, effect, p_ctrl, p_treat): if p_value >= alpha: return f"无统计显著差异 (p={p_value:.4f} ≥ {alpha})。建议继续收集数据或增大样本量。" direction = "提升" if effect > 0 else "下降" return ( f"统计显著 (p={p_value:.4f} < {alpha})。" f"实验组{direction}了{abs(effect)*100:.1f}个百分点" f"（{p_ctrl:.1%} → {p_treat:.1%}）。" ) 4.3 序贯检验（支持提前停止） class SequentialTester: """序贯检验：允许在实验过程中提前判断""" def __init__(self, alpha: float = 0.05, power: float = 0.8, num_looks: int = 5): # Bonferroni 校正 self.adjusted_alpha = alpha / num_looks self.looks = num_looks self.current_look = 0 def should_stop_early( self, control_data: list, treatment_data: list, sample_size_ratio: float # 当前样本量 / 计划样本量 ) -> EarlyStopDecision: """检查是否可以提前停止""" self.current_look = int(sample_size_ratio * self.looks) result = StatisticalTester().test_continuous( control_data, treatment_data, self.adjusted_alpha ) # 护栏指标检查 guardrail_ok = self._check_guardrails(control_data, treatment_data) if not guardrail_ok: return EarlyStopDecision( should_stop=True, reason="护栏指标恶化，建议立即停止实验", winner="control" ) if result.significant: if result.effect_size > 0: return EarlyStopDecision( should_stop=True, reason=f"实验组显著优于对照组 (p={result.p_value:.4f})", winner="treatment" ) else: return EarlyStopDecision( should_stop=True, reason=f"实验组显著劣于对照组 (p={result.p_value:.4f})", winner="control" ) # 计算当前功效 current_power = self._compute_power( len(control_data), result.effect_size ) if current_power > 0.8 and not result.significant: return EarlyStopDecision( should_stop=True, reason=f"功效充足({current_power:.1%})但无显著差异，停止实验", winner="tie" ) return EarlyStopDecision(should_stop=False) 五、LLM 特有的 A/B 测试方法 5.1 LLM-as-Judge A/B 测试 class LLMJudgeABTest: """使用 LLM 作为评判者的 A/B 测试""" async def judge_pair( self, prompt: str, response_a: str, response_b: str, criteria: list[str] ) -> JudgmentResult: """让 LLM 判断哪个回答更好""" judge_prompt = f"""You are an impartial judge. Compare two responses to the same prompt. Prompt: {prompt} Response A: {response_a} Response B: {response_b} Criteria: {', '.join(criteria)} Evaluate which response is better. Consider: 1. Accuracy and correctness 2. Completeness 3. Clarity and structure 4. Adherence to instructions Respond in JSON: {{ "winner": "A" | "B" | "tie", "confidence": 0.0-1.0, "reasoning": "explanation", "scores": {{"A": float, "B": float}} }}""" response = await self.judge_llm.invoke(judge_prompt, temperature=0.0) return JudgmentResult(**json.loads(response.content)) async def run_experiment( self, test_cases: list[TestCase], control_agent: Agent, treatment_agent: Agent, num_judges: int = 3 # 多评判者取平均 ) -> ExperimentResult: results = [] for case in test_cases: # 生成两组回答 response_ctrl = await control_agent.run(case.input) response_treat = await treatment_agent.run(case.input) # 多评判者投票 judgments = [] for i in range(num_judges): judge = self.judges[i] judgment = await judge.judge_pair( case.input, response_ctrl, response_treat, case.criteria ) judgments.append(judgment) # 多数投票 winner = self._majority_vote(judgments) results.append({ "test_id": case.id, "winner": winner, "confidence": np.mean([j.confidence for j in judgments]), }) # 统计分析 wins_treatment = sum(1 for r in results if r["winner"] == "treatment") wins_control = sum(1 for r in results if r["winner"] == "control") ties = sum(1 for r in results if r["winner"] == "tie") # Bradley-Terry 模型检验 bt_stat = self._bradley_terry_test(wins_treatment, wins_control, ties) return ExperimentResult( wins_treatment=wins_treatment, wins_control=wins_control, ties=ties, p_value=bt_stat.p_value, significant=bt_stat.p_value < 0.05, avg_confidence=np.mean([r["confidence"] for r in results]) ) 六、实验报告自动化 class ExperimentReporter: """自动化实验报告生成""" async def generate_report( self, experiment_id: str ) -> ExperimentReport: exp = await self.repo.get(experiment_id) metrics = await self.collector.aggregate_all(experiment_id) test_results = {} for metric_name, data in metrics.items(): if metric_name in ["task_completed", "user_thumbs_up"]: # 比例检验 result = self.tester.test_proportion( data["control"].successes, data["control"].total, data["treatment"].successes, data["treatment"].total ) else: # 连续值检验 result = self.tester.test_continuous( data["control"].values, data["treatment"].values ) test_results[metric_name] = result # 护栏指标检查 guardrail_status = self._check_guardrails(metrics, exp.guardrail_metrics) # 生成决策建议 recommendation = self._generate_recommendation( test_results, guardrail_status, exp.hypothesis ) return ExperimentReport( experiment=exp, sample_sizes={ "control": metrics["task_completed"]["control"].total, "treatment": metrics["task_completed"]["treatment"].total, }, results=test_results, guardrail_status=guardrail_status, recommendation=recommendation, summary=self._generate_summary(test_results, recommendation), generated_at=datetime.now() ) def _generate_recommendation(self, results, guardrails, hypothesis): primary = results.get(hypothesis.dependent_variables[0]) if not primary.significant: return Recommendation( action="continue_or_stop", reason=f"主指标无显著差异 (p={primary.p_value:.4f})。" f"建议：若已达到计划样本量则停止；否则继续收集数据。" ) if primary.effect_size > 0 and guardrails.all_passed: return Recommendation( action="ship", reason=f"主指标显著提升 (p={primary.p_value:.4f}, " f"效应量={primary.effect_size:.3f})。" f"护栏指标全部通过。建议全量发布。" ) if primary.effect_size < 0: return Recommendation( action="do_not_ship", reason=f"主指标显著下降 (p={primary.p_value:.4f})。不建议发布。" ) if not guardrails.all_passed: return Recommendation( action="do_not_ship", reason=f"主指标虽提升但护栏指标恶化：{guardrails.violated}。不建议发布。" ) 七、A/B 测试 Checklist □ 实验假设明确（自变量、因变量、预期效果） □ 样本量计算完成（MDE、power、alpha） □ 流量分配使用一致性哈希（同一用户体验一致） □ 护栏指标已定义并监控 □ 统计检验方法匹配指标类型（比例/连续） □ 序贯检验支持提前停止 □ LLM-as-Judge 评判使用多评判者 □ 实验报告自动生成 □ 决策建议基于数据而非直觉 □ 实验结果归档可追溯 结语 A/B 测试是 Agent 优化的科学基石。在 LLM 的非确定性世界里，直觉是不可靠的，只有统计检验才能区分真实效果和随机噪声。投资 A/B 测试平台不是开销，而是回报率最高的基础设施投资。让每一次 Prompt 修改、每一次模型升级都有数据支撑，这就是 Agent 工程的成熟标志。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Agent 系统的三个核心维度——Prompt、工具、模型——任何一个的变更都可能引发连锁反应。传统软件的版本管理主要针对代码，而 Agent 还需要管理自然语言"代码"（Prompt）、动态加载的工具和外部模型版本。本文将构建完整的 Agent 版本管理体系。 一、Agent 版本的复杂性 变更类型与风险 ┌──────────────────────────────────────────────────────┐ │ Agent 变更类型与风险矩阵 │ ├──────────────┬──────────┬──────────┬─────────────────┤ │ 变更类型 │ 频率 │ 风险等级 │ 影响范围 │ ├──────────────┼──────────┼──────────┼─────────────────┤ │ Prompt 修改 │ 每周 │ 中-高 │ 输出质量/行为 │ │ 工具更新 │ 每月 │ 中 │ 工具调用/结果 │ │ 模型升级 │ 每季度 │ 高 │ 全局行为变化 │ │ System配置 │ 每周 │ 低-中 │ 性能/限制 │ │ Few-shot示例 │ 每月 │ 中 │ 输出风格/格式 │ │ 工作流变更 │ 每月 │ 高 │ 执行路径/延迟 │ └──────────────┴──────────┴──────────┴─────────────────┘ 版本组合的笛卡尔积问题 Agent 行为是 Prompt版本 × 工具版本 × 模型版本 的组合。如果三者各自有 3 个版本，理论上存在 27 种组合。版本管理的目标就是确保任意组合的行为可预测、可回滚。 ...

引言 2026年3月15日，OpenAI API 全球性宕机 47 分钟。那些没有降级方案的 Agent 应用全部显示"服务不可用"，而准备好容灾方案的产品几乎无感知地度过了这次故障。LLM 是 Agent 的核心，但它不是 100% 可靠的基础设施。本文将系统化构建 Agent 的多级降级体系。 一、故障分类与影响分析 1.1 LLM 服务故障类型 ┌──────────────────────────────────────────────────┐ │ LLM 故障分类 │ ├──────────────┬───────────┬───────────────────────┤ │ 故障类型 │ 持续时间 │ 影响范围 │ ├──────────────┼───────────┼───────────────────────┤ │ API 完全宕机 │ 5-60min │ 所有请求失败 │ │ 限流(429) │ 1-10min │ 超额请求失败 │ │ 超时 │ 30-120s │ 单请求受影响 │ │ 模型退版 │ 永久 │ 特定版本不可用 │ │ 区域故障 │ 5-30min │ 特定区域不可用 │ │ 质量退化 │ 未知 │ 输出质量下降 │ └──────────────┴───────────┴───────────────────────┘ 1.2 故障影响评估 @dataclass class FaultImpact: fault_type: str user_visible: bool # 用户是否感知 data_loss: bool # 是否丢数据 recovery_time: str # 恢复时间 business_impact: str # 低/中/高/严重 LLM_FAULT_IMPACTS = [ FaultImpact("API宕机", True, False, "5-60min", "严重"), FaultImpact("限流", True, False, "1-10min", "中"), FaultImpact("超时", True, False, "30-120s", "低"), FaultImpact("质量退化", False, False, "未知", "中"), ] 二、多级降级架构 ┌──────────────────────────────────────────────────────┐ │ 请求入口 │ │ ┌──────┴──────┐ │ │ │ 请求路由 │ │ │ └──────┬──────┘ │ │ │ │ │ ┌─────────────────────▼──────────────────────┐ │ │ │ Level 0: 主模型 (GPT-5) │ │ │ │ 延迟 < 2s, 成功率 99.5% │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 1: 备用模型 (Claude Opus) │ │ │ │ 延迟 < 3s, 成功率 99.5% │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 2: 降级模型 (GPT-5-mini) │ │ │ │ 延迟 < 1s, 功能受限 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 3: 缓存/预计算结果 │ │ │ │ 精度下降，但可用 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 4: 规则引擎/模板回复 │ │ │ │ 基本功能，无AI能力 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 5: 优雅错误页面 │ │ │ │ 引导用户重试或联系人工 │ │ │ └────────────────────────────────────────────┘ │ 三、各级降级实现 Level 0-1：模型切换 from enum import Enum import asyncio class ModelTier(Enum): PRIMARY = "gpt-5" FALLBACK = "claude-opus-4-2026" DEGRADED = "gpt-5-mini" EMERGENCY = "gpt-4o-mini" class LLMFailoverChain: """LLM 多级故障转移链""" def __init__(self): self.chain = [ {"model": "gpt-5", "timeout": 30, "retries": 2}, {"model": "claude-opus-4-2026", "timeout": 45, "retries": 1}, {"model": "gpt-5-mini", "timeout": 15, "retries": 2}, {"model": "gpt-4o-mini", "timeout": 10, "retries": 3}, ] self.circuit_breakers = {m["model"]: CircuitBreaker() for m in self.chain} self.health_checker = HealthChecker() async def invoke(self, messages: list, **kwargs) -> str: errors = [] for tier in self.chain: model = tier["model"] cb = self.circuit_breakers[model] # 检查熔断器状态 if not cb.can_call(): errors.append(f"{model}: circuit breaker open") continue # 检查健康状态 if not await self.health_checker.is_healthy(model): errors.append(f"{model}: unhealthy") continue try: response = await self._call_with_retry( model, messages, timeout=tier["timeout"], retries=tier["retries"], **kwargs ) cb.record_success() # 如果降级了，通知调用方 if model != self.chain[0]["model"]: logger.warning(f"Degraded to {model}") return response except (TimeoutError, RateLimitError) as e: cb.record_failure() errors.append(f"{model}: {e}") continue except Exception as e: cb.record_failure() errors.append(f"{model}: {e}") continue # 所有模型都失败，进入更深层的降级 raise AllModelsFailedError(errors) Level 2：功能降级 class DegradedMode: """降级模式：功能减少但核心可用""" DEGRADATION_RULES = { "gpt-5-mini": { "disable_tools": False, # 仍可用工具 "max_iterations": 3, # 减少迭代次数 "disable_multi_step": True, # 禁用多步推理 "simplified_prompt": True, # 简化 System Prompt "max_context_messages": 5, # 减少上下文 }, "gpt-4o-mini": { "disable_tools": True, # 禁用工具 "max_iterations": 1, # 单轮回复 "disable_multi_step": True, "simplified_prompt": True, "max_context_messages": 3, } } def apply(self, request: dict, model: str) -> dict: rules = self.DEGRADATION_RULES.get(model, {}) degraded_request = request.copy() if rules.get("simplified_prompt"): degraded_request["system"] = self._simplify_prompt(request.get("system", "")) if rules.get("disable_tools"): degraded_request.pop("tools", None) if rules.get("max_context_messages"): messages = degraded_request.get("messages", []) degraded_request["messages"] = messages[-rules["max_context_messages"]:] degraded_request["max_tokens"] = min( degraded_request.get("max_tokens", 4096), 1024 # 降级时限制输出长度 ) return degraded_request Level 3：缓存降级 class CacheFallback: """缓存降级：使用历史相似查询的结果""" async def try_cached_response(self, query: str) -> CachedResponse | None: # 1. 精确匹配 exact = await self.cache.get(query) if exact: return CachedResponse( content=exact, source="exact_cache", warning="This response is from cache due to AI service degradation." ) # 2. 语义匹配 similar = await self.semantic_cache.search(query, threshold=0.85) if similar: return CachedResponse( content=similar.response, source="semantic_cache", warning="This response is based on a similar historical query." ) # 3. 预计算回答（高频问题） precomputed = await self.precomputed_db.get(query) if precomputed: return CachedResponse( content=precomputed, source="precomputed", warning=None # 预计算结果质量有保证 ) return None Level 4：规则引擎降级 class RuleEngineFallback: """规则引擎：LLM 不可用时的最终保底""" def __init__(self): self.rules = self._load_rules() def _load_rules(self) -> list[Rule]: """加载预定义规则""" return [ # 意图匹配规则 Rule( pattern=r"(.*)价格(.*)", action=lambda m: f"我们的产品价格请参考：{self._get_pricing_table()}", intent="pricing_inquiry" ), Rule( pattern=r"(.*)退款(.*)", action=lambda m: f"退款申请已记录。您的退款将在3-5个工作日内处理。" f"工单号：{self._generate_ticket()}", intent="refund_request" ), Rule( pattern=r"(hello|hi|你好|嗨)", action=lambda m: "您好！我是智能助手。" "AI服务暂时受限，但我可以处理基础请求。请说明您需要什么帮助。", intent="greeting" ), ] def handle(self, user_input: str) -> str: for rule in self.rules: match = re.match(rule.pattern, user_input, re.IGNORECASE) if match: return rule.action(match) # 兜底回复 return ( "AI 服务暂时不可用，我无法处理您的复杂请求。\n" "您可以：\n" "1. 稍后重试\n" "2. 联系人工客服：400-xxx-xxxx\n" "3. 访问帮助中心：https://help.example.com" ) 四、降级决策器 class DegradationOrchestrator: """降级编排器：协调各级降级策略""" def __init__(self): self.llm_chain = LLMFailoverChain() self.cache = CacheFallback() self.rules = RuleEngineFallback() self.degraded_mode = DegradedMode() self.health_monitor = HealthMonitor() async def handle_request( self, messages: list, tools: list | None = None, user_id: str = "", priority: str = "normal" ) -> Response: # 检查系统健康状态 health = await self.health_monitor.get_status() # 根据健康状态选择策略 if health.llm_available: try: return await self.llm_chain.invoke(messages, tools=tools) except AllModelsFailedError: health.llm_available = False if health.cache_available: cached = await self.cache.try_cached_response( messages[-1]["content"] ) if cached: return Response( content=cached.content, degraded=True, source=cached.source, warning=cached.warning ) # 规则引擎兜底 rule_response = self.rules.handle(messages[-1]["content"]) return Response( content=rule_response, degraded=True, source="rule_engine", warning="AI 服务暂时不可用，正在使用规则引擎处理。" ) async def health_check_loop(self): """持续健康检查，自动恢复""" while True: await asyncio.sleep(30) # 每30秒检查一次 if not self.health_monitor.llm_available: # 尝试恢复 test_result = await self._test_llm_health() if test_result: self.health_monitor.llm_available = True logger.info("LLM service recovered!") await self._notify_recovery() 五、降级通知与用户体验 class DegradedUX: """降级时的用户体验设计""" MESSAGES = { "model_switch": { "inline": "（响应速度可能略有不同）", "banner": "⚠️ AI 服务正在切换到备用节点，响应可能略有延迟。" }, "cache_fallback": { "inline": "（以下为缓存回复）", "banner": "⚠️ AI 服务暂时繁忙，正在使用历史缓存回复您。" }, "rule_engine": { "inline": "（基础模式）", "banner": "⚠️ AI 服务暂时不可用，正在使用基础模式处理您的请求。" }, "full_failure": { "banner": "❌ AI 服务暂时不可用。您可以通过以下方式获取帮助：\n" "📞 客服热线：400-xxx-xxxx\n" "💬 在线客服：点击右下角\n" "📧 邮件：support@example.com\n" "我们正在紧急修复中，请稍后重试。" } } def render(self, response: Response) -> dict: if not response.degraded: return {"content": response.content, "banner": None} ux = self.MESSAGES.get(response.source, {}) return { "content": response.content, "banner": ux.get("banner"), "inline_note": ux.get("inline"), "retry_available": True, } 六、降级演练 class DegradationDrill: """降级演练：定期验证降级策略有效性""" async def run_drill(self): """模拟 LLM 故障，验证降级链""" results = [] # 测试 1: 主模型超时 with mock_llm_timeout(ModelTier.PRIMARY): result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is 2+2?"}] ) results.append(DrillResult( test="primary_timeout", passed=result.degraded == True, response_source=result.source )) # 测试 2: 所有模型宕机 with mock_all_llm_down(): result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "你好"}] ) results.append(DrillResult( test="all_llm_down", passed=result.source == "rule_engine", response_source=result.source )) # 测试 3: 缓存命中 with mock_all_llm_down(): # 先正常请求一次建立缓存 await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is your return policy?"}] ) # 再次请求相同问题 result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is your return policy?"}] ) results.append(DrillResult( test="cache_hit", passed=result.source in ("exact_cache", "semantic_cache"), response_source=result.source )) return DrillReport(results=results) 七、降级 Checklist □ 多 LLM Provider 热备，自动切换 < 5s □ 降级模型已配置，功能限制明确 □ 语义缓存层已部署，命中率 > 30% □ 高频问题预计算，离线生成 □ 规则引擎已覆盖核心业务意图 □ 降级通知 UX 已设计，用户可感知 □ 健康检查每 30s 执行，自动恢复 □ 降级演练每月执行，结果归档 □ 人工客服兜底渠道畅通 □ 降级期间数据不丢失（请求排队） 结语 降级策略的本质是"接受不完美，但永不不可用"。在 LLM 依赖度越来越高的今天，没有降级方案的 Agent 系统就像没有安全网的走钢丝。投资降级不是悲观——它是工程成熟度的标志。最好的降级是用户从未注意到的降级。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 一个生产级 Agent 的月度 LLM 账单可以从几百美元到数十万美元不等。2026年，随着 Agent 在企业中的大规模部署，成本优化已成为工程团队的核心 KPI。本文将从 Token 定价模型出发，系统化拆解 Agent 成本结构，并提供可落地的优化方案。 一、Token 经济学基础 1.1 定价模型（2026年Q2市场价） 模型 输入 ($/1M tokens) 输出 ($/1M tokens) 缓存输入 ($/1M tokens) 上下文窗口 GPT-5 $5.00 $15.00 $2.50 256K GPT-5-mini $0.30 $1.50 $0.15 128K Claude Opus 4 $8.00 $24.00 $4.00 200K Claude Sonnet 4 $3.00 $15.00 $1.50 200K Gemini 2.5 Pro $2.50 $10.00 $1.25 2M DeepSeek V4 $0.15 $0.60 $0.07 128K Llama 4 405B $0.80 $2.40 $0.40 128K 关键洞察：输出 Token 的价格是输入的 3-5 倍。因此，减少输出 Token 比减少输入 Token 更具成本效益。 ...

引言 将 Agent 从 Demo 推向生产环境，需要跨越一道巨大的鸿沟。2025年，我们目睹了太多 Agent 在生产环境中翻车的案例：无限循环烧掉数万美元、Prompt 注入导致数据泄露、并发请求拖垮整个系统。这份 Checklist 是用真金白银换来的经验。 一、模型层 Checklist 1.1 模型选择与配置 模型版本锁定：生产环境使用明确的模型版本（如 gpt-5-2026-06-01），而非 gpt-5-latest Fallback 模型配置：主模型不可用时自动切换到备用模型 Token 限制设置：max_tokens 根据业务场景硬性设置，防止无限生成 Temperature 策略：事实性任务 T=0-0.3，创意任务 T=0.7-1.0，代码生成 T=0-0.2 上下文窗口管理：实现对话历史压缩策略，防止 Context Overflow PRODUCTION_MODEL_CONFIG = { "primary": { "model": "gpt-5-2026-06-01", "max_tokens": 4096, "temperature": 0.2, "timeout": 30, "retry": {"max_attempts": 3, "backoff": "exponential"} }, "fallback": { "model": "claude-opus-4-2026-04", "max_tokens": 4096, "temperature": 0.2, "timeout": 45, }, "emergency": { "model": "gpt-4o-2026-03", "max_tokens": 2048, "temperature": 0.0, } } 1.2 成本控制 Token 预算硬限：每个请求/用户/天的 Token 上限 模型路由策略：简单任务用小模型，复杂任务用大模型 缓存层：语义缓存命中率监控 成本告警：单次请求成本 > $0.1 时告警 class TokenBudget: """Token 预算管理器""" def __init__(self, redis_client): self.redis = redis_client async def check_budget( self, user_id: str, requested_tokens: int ) -> bool: daily_key = f"budget:{user_id}:{date.today()}" used = int(await self.redis.get(daily_key) or 0) limit = await self._get_user_limit(user_id) if used + requested_tokens > limit: await self._notify_overrun(user_id, used, requested_tokens, limit) return False return True async def consume( self, user_id: str, tokens_used: int, cost: float ): daily_key = f"budget:{user_id}:{date.today()}" cost_key = f"cost:{user_id}:{date.today()}" pipe = self.redis.pipeline() pipe.incrby(daily_key, tokens_used) pipe.incrbyfloat(cost_key, cost) pipe.expire(daily_key, 86400 * 2) pipe.expire(cost_key, 86400 * 2) await pipe.execute() 二、Prompt 层 Checklist System Prompt 注入防护：用户输入与系统指令严格分离 Prompt 版本管理：所有 Prompt 变更通过 Git 管理，支持 A/B 测试 Prompt 长度监控：System Prompt 不超过 Context Window 的 20% Few-shot 示例审计：示例中不包含敏感数据 输出格式约束：使用 JSON Mode 或 Structured Output class PromptSafetyValidator: """Prompt 安全验证器""" INJECTION_PATTERNS = [ r"ignore.*previous.*instructions", r"you.*are.*now.*a", r"system.*prompt.*is", r"reveal.*your.*instructions", ] def validate(self, user_input: str) -> ValidationResult: for pattern in self.INJECTION_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): return ValidationResult( safe=False, reason=f"Potential prompt injection: matched {pattern}" ) if len(user_input) > 10000: return ValidationResult( safe=False, reason="Input exceeds maximum length" ) return ValidationResult(safe=True) 三、工具层 Checklist 工具输入校验：每个工具的输入参数用 Pydantic 模型校验 工具超时设置：每个工具调用设置独立超时 工具权限分级：只读工具自动执行，写操作需审批 工具结果截断：工具输出超过 N 字符时自动摘要 工具幂等性：关键工具支持重试不会产生副作用 from pydantic import BaseModel, Field from typing import Literal class ToolRegistry: """工具注册中心""" def register(self, tool: Tool): # 验证工具定义 assert tool.timeout_ms <= 30000, "Tool timeout must be < 30s" assert tool.input_schema is not None, "Input schema required" assert tool.permission_level in ["read", "write", "admin"] if tool.permission_level in ["write", "admin"]: assert tool.requires_confirmation == True self._tools[tool.name] = tool class WebSearchInput(BaseModel): query: str = Field(..., min_length=1, max_length=200) max_results: int = Field(default=5, ge=1, le=20) safe_search: bool = True class WebSearchTool(Tool): name = "web_search" description = "Search the web for information" input_schema = WebSearchInput timeout_ms = 10000 permission_level = "read" requires_confirmation = False idempotent = True 四、架构层 Checklist 4.1 并发与限流 请求限流：QPS / 并发数 / Token/s 三维度限流 队列隔离：不同优先级请求使用独立队列 背压机制：下游不可用时主动拒绝请求，而非堆积 from asyncio import Semaphore, Queue import asyncio class AgentRequestHandler: def __init__(self, max_concurrent: int = 10): self.semaphore = Semaphore(max_concurrent) self.priority_queue = Queue(maxsize=1000) self.normal_queue = Queue(maxsize=5000) async def handle(self, request, priority: str = "normal"): queue = self.priority_queue if priority == "high" else self.normal_queue if queue.full(): raise ServiceUnavailableError("Request queue full") await queue.put(request) async with self.semaphore: return await self._process(request) 4.2 状态管理 会话持久化：对话状态可持久化到外部存储 检查点机制：长流程 Agent 支持断点续传 幂等 ID：每个请求分配幂等 ID，防重复执行 4.3 容灾设计 多 Provider 热备：OpenAI / Anthropic / Azure 至少两家 降级策略：LLM 不可用时回退到规则引擎 熔断器：错误率 > 50% 时自动熔断 30s class CircuitBreaker: """Agent 熔断器""" def __init__( self, failure_threshold: int = 10, recovery_timeout: int = 30, half_open_max_calls: int = 3 ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.half_open_max_calls = half_open_max_calls self._state = "closed" # closed / open / half_open self._failures = 0 self._last_failure_time = None async def call(self, func, *args, **kwargs): if self._state == "open": if time.time() - self._last_failure_time > self.recovery_timeout: self._state = "half_open" else: raise CircuitOpenError("Agent temporarily unavailable") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise 五、安全层 Checklist PII 检测与脱敏：用户输入和 LLM 输出中的 PII 自动脱敏 输出过滤：有害内容、不当建议的过滤层 工具沙箱：代码执行工具在容器中运行，限制网络和文件访问 审计日志：所有 Agent 决策记录可审计 数据驻留：敏感数据不发送到 LLM，本地处理 class PIIRedactor: """PII 脱敏器""" PATTERNS = { "email": (r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]'), "phone": (r'\b1[3-9]\d{9}\b', '[PHONE]'), "id_card": (r'\b\d{17}[\dXx]\b', '[ID_CARD]'), "bank_card": (r'\b\d{16,19}\b', '[BANK_CARD]'), } def redact(self, text: str) -> tuple[str, dict]: """返回脱敏文本和映射表（用于恢复）""" mapping = {} redacted = text for pii_type, (pattern, replacement) in self.PATTERNS.items(): matches = re.finditer(pattern, redacted) for i, match in enumerate(matches): placeholder = f"{replacement}_{i}" mapping[placeholder] = match.group() redacted = redacted.replace(match.group(), placeholder) return redacted, mapping 六、可观测性 Checklist 全链路追踪：每个 Agent 步骤生成 Span 结构化日志：所有日志 JSON 格式，包含 trace_id 实时仪表盘：Grafana 仪表盘展示关键指标 告警规则：延迟、错误率、成本、Token 使用量告警 会话回放：支持根据 trace_id 回放完整 Agent 执行过程 七、运维层 Checklist 蓝绿部署：新版本 Agent 灰度发布，支持秒级回滚 配置热更新：Prompt、工具配置不重启即可更新 依赖版本锁定：所有依赖版本锁定，定期安全扫描 Secret 管理：API Key 通过 Vault/Secret Manager 管理 备份策略：对话历史、Agent 状态定期备份 八、合规层 Checklist 数据保留策略：用户对话数据保留期限明确 GDPR/PIPL 合规：支持用户数据删除请求 模型透明度：向用户说明使用了 AI 及其局限性 内容免责声明：输出中标注 AI 生成内容 审计合规：关键决策可追溯，满足监管要求 九、性能层 Checklist P95 延迟 < 5s：首 Token 延迟 < 2s，完整响应 < 30s 流式响应：长输出使用 SSE 流式返回 预计算：高频请求结果预计算缓存 连接池：LLM API 连接复用 CDN 加速：静态资源（Prompt 模板、配置）通过 CDN 分发 十、用户体验层 Checklist 加载状态：Agent 思考时展示进度提示 优雅降级：LLM 超时时返回友好提示而非错误码 多语言支持：Agent 输出跟随用户语言 反馈机制：用户可对 Agent 回答打分 边界处理：处理空输入、超长输入、非预期输入 十一、测试层 Checklist 单元测试覆盖率 > 80%：非 LLM 逻辑全覆盖 集成测试：关键业务流程端到端测试 混沌测试：主动注入 LLM 超时、工具失败 安全测试：Prompt 注入、数据泄露测试 负载测试：验证 10x 峰值流量下的表现 十二、上线前最终确认 □ 紧急关停开关（Kill Switch）可用且已测试 □ 回滚脚本已验证，可在 60s 内完成 □ on-call 排班已确认，告警通知链路畅通 □ 成本预算已设置硬性上限 □ 用户协议已更新，涵盖 AI 使用条款 □ 压测通过：P95 < 5s, 错误率 < 1% □ 安全审计已完成，无高危漏洞 □ 数据备份已验证可恢复 □ 运行手册（Runbook）已编写 □ 团队培训已完成 结语 这份 Checklist 看似冗长，但每一条都来自真实的生产事故。Agent 系统的复杂度远超传统应用——它不仅有代码的确定性风险，还有模型的不确定性风险、工具的副作用风险。上线的标准不是"能跑了"，而是"出事了能兜住"。祝你的 Agent 平稳上线。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 传统软件测试的基石是确定性：相同输入产生相同输出。LLM Agent 打破了这个前提，让 CI/CD 测试面临根本性挑战。2026年，随着 Agent 评估框架的成熟，一套可落地的 CI/CD 测试方法论终于成型。本文将带你构建完整的 Agent 测试金字塔。 一、Agent 测试金字塔 ┌─────────┐ │ E2E │ ← 端到端场景测试（5-10个） ┌┴─────────┴┐ │ Integration │ ← 多 Agent 协作测试（20-50个） ┌┴──────────────┴┐ │ Evaluation │ ← LLM 评判测试（50-100个） ┌┴──────────────────┴┐ │ Component │ ← 工具/Prompt 测试（200+） ┌┴──────────────────────┴┐ │ Unit Test │ ← 纯函数测试（500+） └───────────────────────────┘ 二、第一层：单元测试（确定性层） 单元测试只测试不涉及 LLM 的部分：数据处理、工具执行的解析逻辑、Prompt 模板渲染。 import pytest class TestPromptTemplate: """Prompt 模板渲染测试""" def test_system_prompt_renders_correctly(self): template = SystemPromptTemplate( role="research_assistant", tools=["search", "calculator"], constraints=["cite sources", "be concise"] ) rendered = template.render() assert "research_assistant" in rendered assert "search" in rendered assert "calculator" in rendered assert "cite sources" in rendered def test_few_shot_template_with_examples(self): template = FewShotTemplate( system="You are a classifier", examples=[ {"input": "I love it", "output": "positive"}, {"input": "Terrible", "output": "negative"}, ], query="{user_input}" ) rendered = template.render(user_input="Amazing!") assert "positive" in rendered assert "negative" in rendered assert "Amazing!" in rendered class TestToolParsing: """工具调用解析测试""" @pytest.mark.parametrize("raw_output,expected_tool,expected_args", [ ('{"tool": "search", "args": {"q": "weather"}}', "search", {"q": "weather"}), ('```json\n{"tool": "calc", "args": {"expr": "1+1"}}\n```', "calc", {"expr": "1+1"}), ('I\'ll use the search tool: {"tool": "search", "args": {"q": "news"}}', "search", {"q": "news"}), ]) def test_parse_tool_call(self, raw_output, expected_tool, expected_args): result = parse_tool_call(raw_output) assert result.tool == expected_tool assert result.args == expected_args def test_parse_malformed_output(self): with pytest.raises(ToolParseError): parse_tool_call("This is not JSON at all") 三、第二层：组件测试（Mock LLM） 使用 Mock LLM 测试 Agent 的控制流，确保工作流逻辑正确。 ...

System Prompt：AI 应用的操作系统 如果说 User Prompt 是给 AI 的任务指令，那么 System Prompt 就是 AI 应用的"操作系统"——它定义了 AI 的身份、能力边界、行为准则和知识背景。2026 年的实践表明，System Prompt 的质量直接决定了 AI 应用的上限，优秀的 System Prompt 可以让同一模型的表现提升 40-60%。 一、System Prompt 的三层架构 ┌────────────────────────────────────┐ │ 身份层（WHO） │ ← 角色定义、人格设定 ├────────────────────────────────────┤ │ 约束层（WHAT & HOW） │ ← 行为规则、输出格式、安全限制 ├────────────────────────────────────┤ │ 知识层（WHAT TO KNOW） │ ← 领域知识、术语表、参考资料 └────────────────────────────────────┘ 二、身份层设计 2.1 角色定义模板 ## 角色 你是「角色名称」，一个「核心定位描述」。 ### 核心能力 1. 能力1（具体描述 + 熟练度） 2. 能力2（具体描述 + 熟练度） 3. 能力3（具体描述 + 熟练度） ### 性格特质 - 特质1：具体表现 - 特质2：具体表现 ### 语言风格 - 语气：专业/友好/幽默/严肃 - 用词偏好：技术性强/通俗易懂/学术风格 - 句式：简洁有力/详尽展开/对话感强 ### 交互边界 - 你能做什么：列举 - 你不能做什么：列举 2.2 角色定义实例：技术顾问 ## 角色 你是「TechAdvisor」，一个拥有15年经验的技术架构顾问，擅长分布式系统、 云原生架构和AI工程化。 ### 核心能力 1. 系统架构设计（精通微服务、事件驱动、CQRS等模式） 2. 技术选型评估（能对比主流方案的优劣，给出量化分析） 3. 性能优化（擅长识别瓶颈，提供可落地的优化方案） ### 性格特质 - 务实：不推崇过度设计，坚持"够用就好" - 谨慎：对新技术保持审慎态度，强调风险控制 - 直接：指出架构问题不留情面，但会给出改进建议 ### 语言风格 - 语气：专业、直接、有条理 - 用词：使用行业标准术语，但不假定用户了解所有缩写 - 句式：先给结论，再展开分析 ### 交互边界 - 能做：架构评审、技术选型、代码审查建议、性能分析 - 不能做：直接编写完整生产代码、替代团队做架构决策、 提供法律/财务建议 2.3 角色一致性与深度 设计维度 浅层设计 深层设计 角色 “你是助手” 完整背景故事+价值观+能力边界 语气 “友好专业” 具体到句式、用词、标点偏好 知识 通用知识 领域专家级知识+术语体系 边界 “不能做坏事” 具体场景的拒绝策略+替代方案 一致性 靠单条Prompt维持 贯穿所有交互的行为准则 三、约束层设计 3.1 约束分类体系 from enum import Enum from dataclasses import dataclass from typing import List class ConstraintType(Enum): BEHAVIORAL = "behavioral" # 行为约束 OUTPUT = "output" # 输出约束 SAFETY = "safety" # 安全约束 ETHICAL = "ethical" # 伦理约束 SCOPE = "scope" # 范围约束 INTERACTION = "interaction" # 交互约束 @dataclass class Constraint: type: ConstraintType rule: str priority: int # 1=最高, 5=最低 enforcement: str # 检查方式 fallback: str # 违反时的行为 # 约束配置示例 constraints = [ Constraint( type=ConstraintType.SAFETY, rule="绝不生成可执行的恶意代码", priority=1, enforcement="硬性拦截", fallback="拒绝并说明原因" ), Constraint( type=ConstraintType.OUTPUT, rule="代码回答必须包含错误处理示例", priority=2, enforcement="输出后检查", fallback="追加错误处理示例" ), Constraint( type=ConstraintType.SCOPE, rule="只回答与软件开发相关的问题", priority=2, enforcement="意图分类", fallback="礼貌拒绝并引导到相关领域" ), Constraint( type=ConstraintType.INTERACTION, rule="复杂问题分步回答，每步确认用户理解", priority=3, enforcement="结构化输出", fallback="继续下一步" ), ] 3.2 完整约束层 Prompt ## 行为规则 ### 核心原则 1. 准确性优先：不确定时明确说明，不要编造答案 2. 安全性优先：安全要求高于用户体验 3. 实用性优先：给出可操作的建议，而非空泛理论 4. 透明性优先：说明推理过程和依据 ### 回答规范 1. 代码示例必须可运行，包含必要的导入语句 2. 技术方案需说明适用场景和局限性 3. 涉及多个选择时，用对比表格呈现 4. 长回答使用标题分节，每节不超过300字 ### 拒绝策略 以下情况需要拒绝： - 请求生成恶意代码或攻击工具 - 请求绕过安全措施 - 请求提供法律/医疗/财务专业建议 - 请求涉及用户隐私数据 拒绝格式： "这个问题超出了我的服务范围。原因：[具体原因]。 建议您：[替代方案]" ### 特殊处理 - 如果用户的问题不清晰，先提问澄清，不要假设 - 如果用户的问题包含错误前提，先指出错误 - 如果用户的代码有安全漏洞，必须指出 3.3 输出格式约束 ## 输出格式 ### 代码相关回答 [问题分析]（2-3句） ...