实践指南

为什么要多渠道集成？ Agent的价值在于随时可用。如果用户在Telegram上聊天、在Discord里协作、在Signal中沟通，你的Agent就需要在这些渠道都能响应。 OpenClaw支持20+消息渠道集成，是开源Agent框架中渠道覆盖最广的。 OpenClaw支持的渠道 渠道 类型 支持功能 配置难度 Telegram IM 文本/图片/文件/按钮 ⭐ Discord 社区 文本/图片/线程/反应 ⭐⭐ Signal IM 文本/图片 ⭐⭐⭐ WhatsApp IM 文本/图片/语音 ⭐⭐⭐ Slack 协作 文本/图片/线程/应用 ⭐⭐ 飞书 协作 文本/卡片/机器人 ⭐⭐ 微信 IM 文本/图片 ⭐⭐⭐ Matrix IM 文本/图片 ⭐⭐ IRC IM 文本 ⭐ iMessage IM 文本/图片 ⭐⭐⭐ SMS 通信 文本 ⭐⭐ QQ Bot IM 文本/图片 ⭐⭐ Telegram集成实战 1. 创建Bot # 在Telegram中找 @BotFather # /newbot → 输入名称 → 获取Token # Token示例：7812345678:AAExxxxxxxxxxxxxxxxxxxxxxxxxxxx 2. 配置OpenClaw // openclaw.json { "channels": { "telegram": { "enabled": true, "token": "7812345678:AAExxxxxxxxxxxxxxxxxxxxxxxxxxxx", "allowedUsers": ["your_user_id"], "commands": { "/summary": "总结最近对话", "/reset": "重置会话", "/status": "查看状态" } } } } 3. 高级功能 # 内联按钮 def send_with_buttons(chat_id): bot.send_message( chat_id=chat_id, text="请选择操作：", reply_markup={ "inline_keyboard": [[ {"text": "📊 查看报告", "callback_data": "report"}, {"text": "⚙️ 设置", "callback_data": "settings"} ]] } ) # 群组处理 def handle_group_message(message): if message.chat.type in ["group", "supergroup"]: # 群组中只在@提及或回复时响应 if is_mentioned(message) or is_reply_to_bot(message): return process_message(message) return HEARTBEAT_OK # 否则保持安静 4. 实际效果 用户: @MyAgent 帮我查一下服务器状态 Agent: 🔍 正在检查... ✅ 阿里云 ECS (182.92.245.108) - CPU: 12% | 内存: 1.2/2GB - 磁盘: 21% | 运行时间: 45天 ✅ 腾讯云 (43.165.198.197) - CPU: 8% | 内存: 320/957MB - 磁盘: 21% | 论坛正常 [📊 详细报告] [🔄 刷新] Discord集成实战 1. 创建Bot # 1. 访问 Discord Developer Portal # 2. 创建 Application → Bot # 3. 获取Token # 4. 开启MESSAGE CONTENT INTENT # 5. 生成邀请链接，邀请到服务器 2. 配置OpenClaw { "channels": { "discord": { "enabled": true, "token": "MTIxxxxxxxxxxxxxxxxxxx", "prefix": "!", "allowedChannels": ["1234567890"], "reactToMessages": true, "threadSupport": true } } } 3. 群组行为规则 class DiscordBehavior: def should_respond(self, message): # 回应条件 if message.mentions_bot: # @提及 return True if message.is_reply_to_bot: # 回复bot return True if self.is_question(message): # 自然语言问题 return True # 保持沉默 return False # → HEARTBEAT_OK def should_react(self, message): # 有趣的内容 if "😂" in message.content or "lol" in message.content.lower(): return "😂" if message.is_interesting: return "🤔" if message.is_helpful: return "👍" return None 4. 线程支持 用户: @Agent 分析一下这个错误日志 [附件] Agent: 正在分析... [创建线程"错误日志分析"] 在线程中回复详细分析结果 Signal集成实战 1. 安装signal-cli # Linux服务器 sudo apt install signal-cli # 注册号码 signal-cli -u +8613789892969 register signal-cli -u +8613789892969 verify <验证码> 2. 配置OpenClaw { "channels": { "signal": { "enabled": true, "phoneNumber": "+8613789892969", "signalCliPath": "/usr/bin/signal-cli", "trustedContacts": ["+8613800138000"] } } } Signal更注重隐私，适合需要端到端加密的场景。 ...

Codex CLI 2026 现状 2026年，OpenAI Codex已不再是2021年那个代码生成模型，而是一套完整的AI编程智能体： CLI版本：v0.130.0（2026-05-08） GitHub Star：83,200+ 核心能力：读代码、改代码、跑命令、做审查 支持模型：gpt-5-codex、gpt-5.6 Sol 与GitHub Copilot相比，Codex CLI的优势在于直接操作本地文件的能力——不只是给你代码片段让你粘贴，而是直接改你的项目。 环境搭建 系统要求 要求 最低版本 推荐版本 操作系统 macOS 12+ / Ubuntu 20.04+ / Win11(WSL2) macOS 14+ / Ubuntu 22.04+ Node.js 18.0+ 22.0+ npm 10.0+ 最新版 Git 2.23+ 2.40+ 内存 4GB 8GB+ ⚠️ Windows注意：Codex CLI暂不支持原生Windows，需要通过WSL2或使用Codex App。 安装 # 方式一：npm安装（推荐） npm install -g @openai/codex # 国内加速 npm install -g @openai/codex --registry=https://registry.npmmirror.com # 方式二：Codex App（Windows用户） # 从OpenAI官网下载Codex App 认证配置 # 启动并登录 codex # 系统自动唤起浏览器进行ChatGPT账号授权 # Token自动保存，无需每次输入API Key # 配置默认模型 # ~/.codex/config.toml [model] default = "gpt-5-codex" reasoning = "high" # low/medium/high 三种使用模式 1. CLI交互模式 $ codex > 帮我审查src/auth目录下的代码，重点关注安全问题 # Codex会： # 1. 读取src/auth/下所有文件 # 2. 分析代码逻辑和安全风险 # 3. 输出审查报告和修改建议 # 4. 可选择直接应用修改 2. 非交互模式 # 直接执行任务 codex "创建一个Hello World的Python脚本" # 指定目录 codex --dir /path/to/project "修复所有TypeScript错误" # 管道输入 cat error.log | codex "分析这个错误日志，找出根本原因" 3. 云端网页版 访问 chatgpt.com/codex： ...

LLM缓存的价值 LLM推理是昂贵的——每次调用消耗GPU算力和API费用。但很多请求是重复或高度相似的。缓存可以让"已经算过的不再重算"，是投入产出比最高的优化手段。 三层缓存架构 请求 → L1: 精确缓存 → L2: 语义缓存 → L3: 前缀缓存 → LLM L1：精确缓存 import redis.asyncio as redis class ExactCache: """精确匹配缓存：相同输入返回相同输出""" def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) def cache_key(self, model, messages, temperature, max_tokens): """生成缓存键""" content = json.dumps({ "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, }, sort_keys=True, ensure_ascii=False) return hashlib.sha256(content.encode()).hexdigest() async def get(self, model, messages, **params): key = self.cache_key(model, messages, params.get("temperature", 0.7), params.get("max_tokens", 2048)) cached = await self.redis.get(key) return json.loads(cached) if cached else None async def set(self, model, messages, response, ttl=3600, **params): key = self.cache_key(model, messages, params.get("temperature", 0.7), params.get("max_tokens", 2048)) await self.redis.setex(key, ttl, json.dumps(response, ensure_ascii=False)) 适用场景 temperature=0的确定性输出 常见问题FAQ 系统提示词相同的请求 L2：语义缓存 class SemanticCache: """语义缓存：相似查询命中缓存""" def __init__(self, vector_store, embed_model, threshold=0.95): self.store = vector_store self.embed_model = embed_model self.threshold = threshold async def get(self, query, model="default", **params): # 向量化查询 query_embedding = await self.embed_model.embed(query) # 搜索相似缓存 results = await self.store.search( query_embedding, filter={"model": model}, top_k=1 ) if results and results[0]["score"] >= self.threshold: cached = json.loads(results[0]["document"]) # 检查参数兼容性 if self.params_compatible(cached["params"], params): return cached["response"] return None async def set(self, query, response, model="default", ttl=3600, **params): embedding = await self.embed_model.embed(query) await self.store.add( id=str(uuid.uuid4()), embedding=embedding, document=json.dumps({ "query": query, "response": response, "model": model, "params": params, "timestamp": time.time(), }, ensure_ascii=False), ttl=ttl ) def params_compatible(self, cached_params, request_params): """检查缓存参数是否兼容当前请求""" # temperature差异较大则不兼容 if abs(cached_params.get("temperature", 0.7) - request_params.get("temperature", 0.7)) > 0.1: return False return True 语义缓存的阈值选择 阈值 命中率 准确率 适用场景 0.99 低 极高 严格场景（医疗、法律） 0.95 中 高 通用场景 0.90 高 中 FAQ类场景 0.85 很高 低 不推荐（风险大） L3：前缀缓存（KV Cache共享） class PrefixCacheManager: """前缀缓存：共享相同前缀的KV Cache""" def __init__(self): self.prefix_cache = {} # prefix_hash -> kv_cache def compute_prefix_hash(self, messages): """计算消息前缀的哈希""" # 系统提示 + 历史对话通常构成共享前缀 prefix_text = json.dumps(messages[:-1]) # 除最后一条消息 return hashlib.sha256(prefix_text.encode()).hexdigest() async def get_kv_cache(self, messages): """获取前缀的KV Cache""" prefix_hash = self.compute_prefix_hash(messages) return self.prefix_cache.get(prefix_hash) async def store_kv_cache(self, messages, kv_cache): """存储前缀的KV Cache""" prefix_hash = self.compute_prefix_hash(messages) self.prefix_cache[prefix_hash] = kv_cache vLLM内置前缀缓存支持： ...

护栏：AI安全的最后一道防线 AI护栏（Guardrails）是在LLLM输入和输出之间设置的安全过滤层。它在模型生成的响应到达用户之前，检查并修改不安全的内容。 三层护栏架构 用户输入 → [输入护栏] → LLM → [输出护栏] → 用户 ↓ ↓ [拦截/修改] [拦截/修改] 输入护栏 class InputGuardrail: def __init__(self): self.checks = [ self.check_prompt_injection, self.check_pii_input, self.check_toxic_input, self.check_topic_restriction, ] async def validate(self, user_input, context=None): results = [] for check in self.checks: result = await check(user_input, context) results.append(result) if not result["passed"]: # 严重违规直接拦截 if result["severity"] == "high": return results, False return results, True async def check_prompt_injection(self, text, context): """检测提示注入攻击""" injection_patterns = [ r"忽略.{0,10}(指令|规则|限制)", r"(ignore|disregard).{0,10}(previous|above|all)", r"(system|admin|root)\s*(prompt|instruction)", r"你的.{0,5}(指令|提示|prompt)", r"reveal.{0,10}(system|hidden|secret)", ] for pattern in injection_patterns: if re.search(pattern, text, re.IGNORECASE): return { "check": "prompt_injection", "passed": False, "severity": "high", "error": "检测到提示注入攻击" } return {"check": "prompt_injection", "passed": True} async def check_pii_input(self, text, context): """检测输入中的PII""" pii_patterns = { "phone": r'1[3-9]\d{9}', "id_card": r'\d{17}[\dXx]', "bank_card": r'\d{16,19}', "email": r'[\w.-]+@[\w.-]+\.\w+', } found = {} for pii_type, pattern in pii_patterns.items(): matches = re.findall(pattern, text) if matches: found[pii_type] = matches if found: return { "check": "pii_input", "passed": False, "severity": "medium", "error": f"输入包含敏感信息: {list(found.keys())}" } return {"check": "pii_input", "passed": True} async def check_topic_restriction(self, text, context): """话题限制检查""" restricted_topics = ["政治敏感", "暴力恐怖", "色情"] # 使用分类模型检测 for topic in restricted_topics: if await self.topic_classifier(text, topic): return { "check": "topic_restriction", "passed": False, "severity": "high", "error": f"话题不在允许范围内: {topic}" } return {"check": "topic_restriction", "passed": True} 输出护栏 class OutputGuardrail: def __init__(self): self.checks = [ self.check_toxicity, self.check_pii_leak, self.check_hallucination, self.check_format_safety, ] async def validate(self, response, context=None): for check in self.checks: result = await check(response, context) if not result["passed"]: if result["action"] == "block": return False, self.safe_fallback(result["error"]) elif result["action"] == "sanitize": response = self.sanitize(response, result) return True, response async def check_toxicity(self, text, context): """毒性检测""" # 使用毒性分类器 toxicity_score = await self.toxicity_model(text) if toxicity_score > 0.7: return { "check": "toxicity", "passed": False, "action": "block", "error": f"输出包含有害内容 (score={toxicity_score:.2f})" } return {"check": "toxicity", "passed": True} async def check_pii_leak(self, text, context): """检查输出是否泄露PII""" # 如果输入包含PII，检查输出是否回显 if context and "input_pii" in context: for pii in context["input_pii"]: if pii in text: return { "check": "pii_leak", "passed": False, "action": "sanitize", "error": "输出包含输入中的PII" } return {"check": "pii_leak", "passed": True} async def check_hallucination(self, text, context): """幻觉检测""" if context and "retrieved_docs" in context: # 基于检索文档检查幻觉 is_grounded = await self.fact_check(text, context["retrieved_docs"]) if not is_grounded: return { "check": "hallucination", "passed": False, "action": "block", "error": "输出可能与事实不符" } return {"check": "hallucination", "passed": True} def safe_fallback(self, error): """安全兜底响应""" return f"抱歉，我无法回答这个问题。({error})" def sanitize(self, text, check_result): """清理不安全内容""" # 替换敏感信息 if check_result["check"] == "pii_leak": for pii in check_result.get("pii_list", []): text = text.replace(pii, "***") return text 完整护栏管线 class GuardrailPipeline: def __init__(self, input_guardrail, output_guardrail, llm): self.input_gr = input_guardrail self.output_gr = output_guardrail self.llm = llm async def process(self, user_input, conversation_context=None): # 1. 输入护栏 input_results, input_passed = await self.input_gr.validate( user_input, conversation_context ) if not input_passed: blocked_check = next(r for r in input_results if not r["passed"]) return { "response": f"抱歉，您的请求无法处理。{blocked_check['error']}", "blocked": True, "reason": blocked_check["check"] } # 2. LLM生成 try: response = await self.llm.generate(user_input) except Exception as e: return {"response": "服务暂时不可用", "error": str(e)} # 3. 输出护栏 passed, safe_response = await self.output_gr.validate( response, {"input": user_input, **conversation_context or {}} ) return { "response": safe_response, "blocked": not passed, "guardrail_checks": { "input": input_results, } } 护栏日志 class GuardrailLogger: def __init__(self): self.logger = structlog.get_logger() def log_block(self, check, error, user_input, severity): self.logger.warning("guardrail_block", check=check, error=error, severity=severity, input_preview=user_input[:100], timestamp=datetime.now().isoformat() ) 结语 AI护栏是保障LLM安全输出的关键基础设施。输入护栏防止恶意输入和不当话题，输出护栏检测毒性、PII泄露和幻觉。三层管线（输入检查→LLM生成→输出检查）配合日志监控，构建起完整的AI安全防线。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

AI系统的错误特征 AI系统的错误与传统软件不同——不仅有网络超时、服务不可用等基础设施错误，还有模型幻觉、输出格式错误、安全违规等AI特有错误。需要针对不同类型的错误设计不同的处理策略。 错误分类 from enum import Enum class ErrorType(Enum): # 基础设施错误 TIMEOUT = "timeout" RATE_LIMIT = "rate_limit" SERVICE_UNAVAILABLE = "service_unavailable" # 模型错误 HALLUCINATION = "hallucination" REFUSAL = "refusal" # 模型拒绝回答 TRUNCATION = "truncation" # 输出被截断 INVALID_FORMAT = "invalid_format" # 格式不合规 # 安全错误 PROMPT_INJECTION = "prompt_injection" TOXIC_OUTPUT = "toxic_output" PII_LEAK = "pii_leak" 重试模式 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class AIRetryPolicy: @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30), retry=retry_if_exception_type((TimeoutError, ConnectionError)), retry_error_callback=lambda _: {"error": "Max retries exceeded"} ) async def call_with_retry(self, llm, messages, **kwargs): try: response = await asyncio.wait_for( llm.ainvoke(messages, **kwargs), timeout=30 ) return response except TimeoutError: logger.warning("LLM call timeout, retrying...") raise async def call_with_smart_retry(self, llm, messages, **kwargs): """智能重试：根据错误类型调整策略""" for attempt in range(3): try: response = await llm.ainvoke(messages, **kwargs) # 检查输出质量 if self.is_truncated(response): # 截断错误：增加max_tokens重试 kwargs['max_tokens'] = kwargs.get('max_tokens', 2048) * 2 continue if self.is_refusal(response): # 拒绝回答：调整prompt重试 messages = self.adjust_prompt_for_refusal(messages) continue return response except RateLimitError: await asyncio.sleep(2 ** attempt) except TimeoutError: if attempt < 2: continue raise return {"error": "All retries failed"} 降级模式 class GracefulDegradation: """逐级降级策略""" def __init__(self, models): self.models = models # 按优先级排序的模型列表 async def generate(self, messages, **kwargs): errors = [] for model in self.models: try: response = await model.generate(messages, **kwargs) # 验证响应质量 if self.validate_response(response): return response else: errors.append(f"{model.name}: invalid response") except Exception as e: errors.append(f"{model.name}: {str(e)}") logger.warning(f"Falling back from {model.name}: {e}") continue # 所有模型都失败，返回兜底响应 logger.error(f"All models failed: {errors}") return self.fallback_response(messages) def fallback_response(self, messages): """兜底响应""" return { "content": "抱歉，服务暂时遇到问题。请稍后重试。", "status": "degraded", "timestamp": datetime.now().isoformat() } 断路器模式 class CircuitBreaker: """断路器：防止持续请求故障服务""" def __init__(self, failure_threshold=5, recovery_timeout=60): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.failure_count = 0 self.last_failure_time = None self.state = "closed" # closed, open, half-open 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("Circuit breaker is open") try: result = await func(*args, **kwargs) self.on_success() return result except Exception as e: self.on_failure() raise def on_success(self): self.failure_count = 0 self.state = "closed" def on_failure(self): self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = "open" logger.error("Circuit breaker opened") 输出验证 class OutputValidator: """验证LLM输出的合法性和质量""" async def validate(self, response, expected_format=None): checks = [ self.check_empty(response), self.check_length(response), self.check_safety(response), self.check_format(response, expected_format), ] for check in checks: if not check["passed"]: return {"valid": False, "error": check["error"]} return {"valid": True} def check_safety(self, response): # 检查是否包含敏感信息 sensitive_patterns = [ r'\d{18}', # 身份证号 r'\d{16,19}', # 银行卡号 r'password\s*[:=]', # 密码 ] for pattern in sensitive_patterns: if re.search(pattern, response, re.IGNORECASE): return {"passed": False, "error": "Sensitive info detected"} return {"passed": True} def check_format(self, response, expected_format): if expected_format == "json": try: json.loads(response) except: return {"passed": False, "error": "Invalid JSON"} return {"passed": True} 错误监控 class ErrorMonitor: def __init__(self): self.error_counts = defaultdict(int) self.error_history = deque(maxlen=1000) def record(self, error_type, context=None): self.error_counts[error_type] += 1 self.error_history.append({ "type": error_type, "timestamp": datetime.now().isoformat(), "context": context or {}, }) # 错误率突增告警 if self.error_counts[error_type] > 100: self.alert(f"Error {error_type} exceeded 100 occurrences") def health_check(self): """系统健康检查""" total_errors = sum(self.error_counts.values()) recent_errors = [ e for e in self.error_history if (datetime.now() - datetime.fromisoformat(e["timestamp"])).seconds < 300 ] return { "total_errors": total_errors, "recent_errors_5min": len(recent_errors), "top_errors": dict(sorted(self.error_counts.items(), key=lambda x: x[1], reverse=True)[:5]), } 实践建议 区分错误类型：基础设施错误用重试，模型错误用降级，安全错误用阻断 设置合理超时：LLM调用超时建议30-60秒 指数退避：重试间隔使用指数退避，避免雪崩 兜底方案：始终准备兜底响应，确保用户体验 错误分类上报：不同类型错误分别统计，便于定位问题 结语 AI系统的错误处理需要比传统软件更精细的设计——不仅要处理基础设施故障，还要处理模型输出质量问题和安全风险。重试、降级、断路器和输出验证的组合使用，可以构建出既健壮又优雅的AI系统。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

量化部署决策树 是否在H100上部署？ ├─ 是 → FP8量化（原生加速，精度最佳） └─ 否 → 显存是否足够FP16？ ├─ 是 → FP16（无精度损失） └─ 否 → INT8还是INT4？ ├─ 精度要求高 → INT8 (W8A8) └─ 显存优先 → INT4 (GPTQ/AWQ) GPTQ量化 # 使用AutoGPTQ进行量化 from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig # 量化配置 quantize_config = BaseQuantizeConfig( bits=4, # 4-bit量化 group_size=128, # 分组大小 desc_act=False, # 是否按激活降序排列 ) # 准备校准数据 calibration_data = load_calibration_dataset(n_samples=128, max_length=512) # 加载模型并量化 model = AutoGPTQForCausalLM.from_pretrained( "Qwen/Qwen3-32B", quantize_config, ) model.quantize(calibration_data) # 保存量化模型 model.save_quantized("./qwen3-32b-gptq-4bit") AWQ量化 # 使用AutoAWQ进行量化 from awq import AutoAWQForCausalLM model = AutoAWQForCausalLM.from_pretrained("Qwen/Qwen3-32B") quant_config = { "zero_point": True, "q_group_size": 128, "w_bit": 4, "version": "GEMM" } model.quantize( calibration_data, quant_config=quant_config ) model.save_quantized("./qwen3-32b-awq-4bit") vLLM部署量化模型 # GPTQ模型部署 vllm serve ./qwen3-32b-gptq-4bit \ --quantization gptq \ --dtype float16 \ --tensor-parallel-size 1 \ --max-model-len 8192 \ --gpu-memory-utilization 0.9 # AWQ模型部署 vllm serve ./qwen3-32b-awq-4bit \ --quantization awq \ --dtype float16 \ --max-model-len 8192 # FP8模型部署（H100） vllm serve ./model \ --quantization fp8 \ --kv-cache-dtype fp8 Ollama部署 # 从GGUF文件创建Ollama模型 cat > Modelfile <<EOF FROM ./qwen3-32b-q4_k_m.gguf PARAMETER num_ctx 8192 PARAMETER temperature 0.7 EOF ollama create my-qwen -f Modelfile ollama run my-qwen 量化效果对比 以Qwen3-32B为例（单卡A100 80GB）： ...

延迟的两个关键指标 LLM服务的延迟分为两部分： TTFT（Time To First Token）：从请求到第一个token返回的时间 TPOT（Time Per Output Token）：每个后续token的生成时间 用户感知延迟 = TTFT + (输出token数 × TPOT)。优化需要分别针对这两个指标。 TTFT优化 预填充优化 TTFT主要由预填充（处理输入prompt）时间决定： # 1. 分块预填充：避免长prompt阻塞短请求 vllm serve model --enable-chunked-prefill --max-num-batched-tokens 4096 # 2. 前缀缓存：共享系统提示词的KV Cache vllm serve model --enable-prefix-caching # 3. 减少输入长度：精简系统提示词 # 差：500 token的系统提示词 # 好：150 token的精简系统提示词 模型预热 async def warmup_model(model_name): """服务启动时预热模型""" # 预加载模型到GPU dummy_input = "warmup" await llm.generate(dummy_input, max_tokens=1) # 预填充常见前缀 common_prompts = [ "你是一个专业助手", "请根据以下信息回答问题", ] for prompt in common_prompts: await llm.generate(prompt, max_tokens=1) TPOT优化 推测解码 # vLLM启用推测解码 vllm serve model \ --speculative-model /models/draft-model \ --num-speculative-tokens 5 \ --speculative-draft-tensor-parallel-size 1 量化 # INT8量化（2倍加速，精度损失<1%） vllm serve model --quantization awq --dtype float16 # INT4量化（3倍加速，精度损失3-5%） vllm serve model --quantization gptq --dtype float16 KV Cache优化 # FP8 KV Cache（减少显存带宽压力） vllm serve model --kv-cache-dtype fp8 网络层优化 流式响应 # 流式响应让用户更早看到输出 @app.post("/chat") async def chat(): async def stream(): yield "data: " # 立即发送头部 async for token in llm.astream(messages): yield json.dumps({"content": token}) + "\n" return StreamingResponse(stream()) 连接复用 # 使用HTTP/2或WebSocket减少连接建立开销 import httpx # 全局复用客户端 client = httpx.AsyncClient( http2=True, limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), keepalive_expiry=30, ) 调度优化 优先级调度 class PriorityScheduler: """交互请求优先于批处理请求""" async def schedule(self, request): if request.type == "interactive": # 交互请求：立即处理 return await self.process_immediately(request) else: # 批处理请求：低峰期处理 return await self.queue_for_later(request) 请求预取 class RequestPrefetcher: """预测用户可能的下一步请求，提前计算""" async def on_user_typing(self, session_id): """用户正在输入时，预计算可能的请求""" likely_queries = await self.predict_queries(session_id) for query in likely_queries[:2]: # 预计算2个最可能的查询 cache_key = f"prefetch:{session_id}:{query}" if not await self.cache.exists(cache_key): result = await self.llm.generate(query) await self.cache.setex(cache_key, 30, result) 端到端优化清单 优化项 TTFT改善 TPOT改善 实现难度 分块预填充 30-50% — 低 前缀缓存 40-60% — 低 量化 10-20% 50-100% 中 推测解码 — 50-100% 中 流式响应 感知80%↓ — 低 模型分层 20-40% 30-50% 中 监控 # 延迟分解监控 class LatencyBreakdown: def record(self, request_start, first_token, request_end): ttft = first_token - request_start total = request_end - request_start tpot = (total - ttft) / max(output_tokens - 1, 1) metrics = { "ttft_ms": ttft * 1000, "tpot_ms": tpot * 1000, "total_ms": total * 1000, } # 告警 if ttft > 2.0: # TTFT > 2秒 alert("High TTFT") if tpot > 0.05: # TPOT > 50ms alert("High TPOT") 结语 LLM服务延迟优化需要从TTFT和TPOT两个维度系统推进。分块预填充和前缀缓存是最有效的TTFT优化手段，量化和推测解码是TPOT的核心优化手段。流式响应虽然不改变实际延迟，但显著改善用户感知。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

AI成本的结构 AI系统的成本主要由以下几部分构成： API调用费：按token计费的LLM API费用（通常占60-70%） 基础设施：GPU服务器/云服务费用（自部署场景） 向量数据库：存储和检索费用 数据标注：人工标注和评估成本 工程开发：系统开发和维护的人力成本 模型分层策略 class ModelTierRouter: """根据任务复杂度路由到不同级别的模型""" def __init__(self): self.tiers = { "simple": {"model": "qwen3-7b", "cost_per_1k": 0.0005}, "medium": {"model": "qwen3-32b", "cost_per_1k": 0.002}, "complex": {"model": "qwen3-72b", "cost_per_1k": 0.008}, } def route(self, query, context=None): # 简单规则路由 if len(query) < 50 and "?" in query: return self.tiers["simple"] # 基于意图分类 complexity = self.estimate_complexity(query, context) if complexity < 0.3: return self.tiers["simple"] elif complexity < 0.7: return self.tiers["medium"] else: return self.tiers["complex"] def estimate_complexity(self, query, context): """估计查询复杂度""" score = 0 # 长查询更复杂 score += min(len(query) / 1000, 0.3) # 需要推理的关键词 reasoning_words = ["分析", "比较", "为什么", "推导", "计算"] score += 0.2 * any(w in query for w in reasoning_words) # 多轮上下文更复杂 if context and len(context) > 5: score += 0.2 # 代码生成更复杂 if "代码" in query or "函数" in query: score += 0.3 return min(score, 1.0) 缓存策略 class SemanticCache: """语义缓存：相似查询命中缓存""" def __init__(self, vector_store, similarity_threshold=0.95): self.store = vector_store self.threshold = similarity_threshold async def get(self, query): """查询缓存""" query_embedding = await self.embed(query) results = await self.store.search( query_embedding, top_k=1 ) if results and results[0].score > self.threshold: # 缓存命中 return json.loads(results[0].document) return None async def set(self, query, response, ttl=3600): """写入缓存""" embedding = await self.embed(query) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps({ "query": query, "response": response, "timestamp": time.time(), }), ttl=ttl ) Token优化 class TokenOptimizer: """减少token消耗的各种策略""" def compress_history(self, messages): """压缩对话历史""" if len(messages) <= 4: return messages # 保留最近2轮 + 摘要 recent = messages[-4:] old = messages[:-4] summary = self.summarize(old) return [{"role": "system", "content": f"历史摘要：{summary}"}] + recent def optimize_prompt(self, prompt): """精简提示词""" # 移除冗余空格和换行 prompt = re.sub(r'\n{3,}', '\n\n', prompt) prompt = re.sub(r' {2,}', ' ', prompt) return prompt.strip() def truncate_context(self, documents, max_tokens=2000): """智能截断文档""" result = [] current_tokens = 0 for doc in documents: doc_tokens = len(doc) // 4 if current_tokens + doc_tokens > max_tokens: # 截断最后一个文档而非完全丢弃 remaining = max_tokens - current_tokens if remaining > 100: result.append(doc[:remaining * 4]) break result.append(doc) current_tokens += doc_tokens return result 批处理优化 class BatchProcessor: """合并多个请求降低API调用次数""" async def batch_generate(self, requests): """将多个独立请求合并为一个""" # 合并多个查询为一个prompt combined_prompt = "请依次回答以下问题：\n\n" for i, req in enumerate(requests): combined_prompt += f"问题{i+1}：{req['query']}\n" combined_prompt += "\n请按问题编号分别回答。" response = await self.llm.generate(combined_prompt) # 解析响应 answers = self.parse_batch_response(response, len(requests)) return answers 成本监控 class CostMonitor: def __init__(self): self.daily_costs = defaultdict(float) self.user_costs = defaultdict(float) self.model_costs = defaultdict(float) def record(self, user_id, model, input_tokens, output_tokens): pricing = { "qwen3-7b": {"input": 0.0003, "output": 0.0006}, "qwen3-32b": {"input": 0.001, "output": 0.002}, "qwen3-72b": {"input": 0.004, "output": 0.008}, } rates = pricing.get(model, {"input": 0.001, "output": 0.002}) cost = (input_tokens / 1000 * rates["input"] + output_tokens / 1000 * rates["output"]) today = datetime.now().strftime("%Y-%m-%d") self.daily_costs[today] += cost self.user_costs[user_id] += cost self.model_costs[model] += cost def alert_if_over_budget(self, daily_budget=100): today = datetime.now().strftime("%Y-%m-%d") if self.daily_costs[today] > daily_budget: self.send_alert( f"日成本超预算：${self.daily_costs[today]:.2f} / ${daily_budget}" ) 成本优化效果 策略 节省比例 实现难度 模型分层 30-50% 低 语义缓存 20-40% 中 Token优化 10-20% 低 批处理 15-25% 中 本地部署替代 50-80% 高 结语 AI成本优化是一个持续过程。模型分层让简单任务用小模型，语义缓存避免重复计算，Token优化减少浪费，批处理提升效率。组合使用这些策略，可以在不降低用户体验的前提下将AI成本降低50-70%。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

多轮对话的核心挑战 多轮对话不是简单的消息拼接——它需要管理对话状态、控制上下文窗口长度、处理话题切换、维护一致性。一个好的对话管理系统是LLM助手的"大脑"。 对话状态管理 from enum import Enum from dataclasses import dataclass, field class DialogState(Enum): GREETING = "greeting" INFORMATION_SEEKING = "information_seeking" PROBLEM_SOLVING = "problem_solving" CLARIFICATION = "clarification" CLOSING = "closing" @dataclass class ConversationContext: session_id: str user_id: str state: DialogState = DialogState.GREETING topic: str = "" entities: dict = field(default_factory=dict) # 提取的实体 history: list = field(default_factory=list) # 消息历史 user_preferences: dict = field(default_factory=dict) pending_clarification: str = "" class DialogManager: def __init__(self, llm): self.llm = llm self.sessions = {} # session_id -> ConversationContext def get_or_create_session(self, session_id, user_id): if session_id not in self.sessions: self.sessions[session_id] = ConversationContext( session_id=session_id, user_id=user_id ) return self.sessions[session_id] async def process_message(self, session_id, user_id, message): ctx = self.get_or_create_session(session_id, user_id) # 1. 状态检测 ctx.state = await self.detect_state(message, ctx) # 2. 实体提取 new_entities = await self.extract_entities(message) ctx.entities.update(new_entities) # 3. 构建上下文 context = self.build_context(ctx, message) # 4. 生成回复 response = await self.llm.generate(context) # 5. 更新历史 ctx.history.append({"role": "user", "content": message}) ctx.history.append({"role": "assistant", "content": response}) # 6. 上下文窗口管理 self.manage_window(ctx) return response 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4096, reserve_tokens=1024): self.max_tokens = max_tokens self.reserve_tokens = reserve_tokens # 为生成预留 def manage(self, context): """管理上下文窗口大小""" available = self.max_tokens - self.reserve_tokens current_tokens = self.estimate_tokens(context.history) if current_tokens <= available: return # 在限制内，无需处理 # 策略1：摘要旧消息 while current_tokens > available and len(context.history) > 4: # 取最早的2条消息进行摘要 old_msgs = context.history[:2] summary = self.summarize(old_msgs) # 替换为摘要 context.history = [ {"role": "system", "content": f"[早期对话摘要]: {summary}"} ] + context.history[2:] current_tokens = self.estimate_tokens(context.history) # 策略2：如果仍然过长，截断 if current_tokens > available: # 保留system prompt和最近的消息 while current_tokens > available and len(context.history) > 2: context.history.pop(0 if context.history[0]["role"] != "system" else 1) current_tokens = self.estimate_tokens(context.history) def estimate_tokens(self, messages): return sum(len(m["content"]) // 4 for m in messages) def summarize(self, messages): """摘要旧消息""" text = "\n".join(f"{m['role']}: {m['content']}" for m in messages) # 使用LLM摘要 return f"[{len(messages)}条消息的摘要]" 话题管理 class TopicManager: def __init__(self, llm): self.llm = llm async def detect_topic_shift(self, current_message, history): """检测话题是否切换""" if len(history) < 2: return False, None recent_topic = await self.extract_topic(history[-2:]) current_topic = await self.extract_topic([{"content": current_message}]) similarity = await self.compute_similarity(recent_topic, current_topic) if similarity < 0.3: return True, current_topic # 话题切换 return False, recent_topic async def handle_topic_shift(self, ctx, new_topic): """处理话题切换""" # 保存当前话题的摘要 old_summary = await self.summarize_topic(ctx) ctx.topic_summaries = ctx.get("topic_summaries", []) ctx.topic_summaries.append({"topic": ctx.topic, "summary": old_summary}) # 切换到新话题 ctx.topic = new_topic ctx.state = DialogState.INFORMATION_SEEKING 意图跟踪 class IntentTracker: def __init__(self, llm): self.llm = llm self.intent_history = [] async def track(self, message, context): """跟踪用户意图""" prompt = f"""分析用户意图。 用户消息：{message} 对话历史摘要：{self.get_history_summary()} 当前状态：{context.state.value} 输出JSON： {{ "intent": "具体意图", "confidence": 0.0-1.0, "requires_clarification": true/false, "clarification_question": "如需澄清的问题" }}""" result = await self.llm.generate(prompt) intent = json.loads(result) self.intent_history.append(intent) return intent 会话持久化 import redis.asyncio as redis class SessionPersistence: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) async def save_session(self, session_id, context): """持久化会话""" await self.redis.setex( f"session:{session_id}", 86400, # 24小时过期 json.dumps({ "session_id": context.session_id, "user_id": context.user_id, "state": context.state.value, "topic": context.topic, "entities": context.entities, "history": context.history[-20:], # 只存最近20条 "user_preferences": context.user_preferences, }, ensure_ascii=False) ) async def load_session(self, session_id): """加载会话""" data = await self.redis.get(f"session:{session_id}") if data: d = json.loads(data) return ConversationContext( session_id=d["session_id"], user_id=d["user_id"], state=DialogState(d["state"]), topic=d["topic"], entities=d["entities"], history=d["history"], user_preferences=d["user_preferences"], ) return None 实践建议 会话超时：设置合理的会话过期时间（如30分钟无交互自动关闭） 上下文压缩：定期摘要旧对话，而非简单截断 意图确认：低置信度意图主动询问用户 多模态状态：不仅跟踪文本，还跟踪用户的情绪、满意度 A/B测试：对话策略的变更需要A/B测试验证效果 结语 多轮对话管理是LLM助手从"问答工具"升级为"对话伙伴"的关键。状态管理、上下文窗口控制、话题跟踪和会话持久化的协同工作，让Agent能够维持连贯、智能的多轮对话。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

Agent记忆的三层架构 Agent需要像人类一样管理不同类型的记忆： 工作记忆：当前对话的上下文（短期） 情景记忆：过去交互的具体经历（中期） 语义记忆：从交互中提炼的知识（长期） 工作记忆：对话窗口管理 from collections import deque class WorkingMemory: def __init__(self, max_messages=20, max_tokens=4096): self.messages = deque(maxlen=max_messages) self.max_tokens = max_tokens self.token_counter = 0 def add(self, role, content): tokens = len(content) // 4 self.messages.append({"role": role, "content": content, "tokens": tokens}) self.token_counter += tokens # 超出token限制时移除最早的消息 while self.token_counter > self.max_tokens and len(self.messages) > 2: old = self.messages.popleft() self.token_counter -= old["tokens"] def get_context(self): return list(self.messages) def summarize_old_context(self): """当记忆过长时，摘要旧对话""" if len(self.messages) < 10: return old_messages = list(self.messages)[:len(self.messages)//2] summary = await self.summarize(old_messages) # 用摘要替换旧消息 for _ in range(len(old_messages)): self.messages.popleft() self.messages.appendleft({"role": "system", "content": f"之前的对话摘要：{summary}"}) 情景记忆：交互历史存储 class EpisodicMemory: def __init__(self, vector_store, llm): self.store = vector_store # 向量数据库 self.llm = llm async def save_interaction(self, user_id, user_message, assistant_response, metadata=None): """保存一次交互记录""" interaction = { "user_id": user_id, "user_message": user_message, "assistant_response": assistant_response, "timestamp": datetime.now().isoformat(), "metadata": metadata or {}, } # 生成摘要用于检索 summary = await self.llm.generate( f"用一句话概括这次交互：\n用户：{user_message}\n助手：{assistant_response}" ) # 向量化并存储 embedding = await self.llm.embed(summary) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps(interaction, ensure_ascii=False), metadata={"user_id": user_id, "summary": summary} ) async def recall(self, user_id, query, top_k=5): """检索相关的历史交互""" query_embedding = await self.llm.embed(query) results = await self.store.search( query_embedding, filter={"user_id": user_id}, top_k=top_k ) return [json.loads(r.document) for r in results] 语义记忆：知识图谱 class SemanticMemory: def __init__(self, graph_store): self.graph = graph_store # 图数据库（如Neo4j） async def learn(self, user_id, fact): """从交互中提取并存储知识""" # 使用LLM提取结构化知识 extracted = await self.extract_facts(fact) for fact_data in extracted: await self.graph.add_triple( subject=fact_data["subject"], predicate=fact_data["predicate"], object=fact_data["object"], metadata={"user_id": user_id, "source": "interaction"} ) async def query(self, user_id, entity): """查询关于某实体的知识""" triples = await self.graph.query( "MATCH (s)-[p]->(o) WHERE s.name = $entity RETURN s, p, o", {"entity": entity} ) knowledge = [] for triple in triples: knowledge.append(f"{triple['s']} {triple['p']} {triple['o']}") return knowledge async def extract_facts(self, text): """从文本中提取三元组""" prompt = f"""从以下文本中提取知识三元组（主语-谓语-宾语）： 文本：{text} 输出JSON数组： [{{"subject": "...", "predicate": "...", "object": "..."}}]""" result = await self.llm.generate(prompt) return json.loads(result) 记忆整合 class AgentMemorySystem: def __init__(self, working, episodic, semantic): self.working = working # 工作记忆 self.episodic = episodic # 情景记忆 self.semantic = semantic # 语义记忆 async def build_context(self, user_id, current_message): """构建完整的记忆上下文""" context_parts = [] # 1. 工作记忆（当前对话） working_ctx = self.working.get_context() context_parts.append({"type": "working", "messages": working_ctx}) # 2. 情景记忆（相关历史交互） episodic_results = await self.episodic.recall( user_id, current_message, top_k=3 ) if episodic_results: context_parts.append({ "type": "episodic", "memories": [r["summary"] for r in episodic_results] }) # 3. 语义记忆（相关知识） entities = await self.extract_entities(current_message) for entity in entities: knowledge = await self.semantic.query(user_id, entity) if knowledge: context_parts.append({ "type": "semantic", "entity": entity, "facts": knowledge }) return self.format_context(context_parts) def format_context(self, parts): """格式化记忆上下文""" context = "" for part in parts: if part["type"] == "working": context += "## 当前对话\n" for msg in part["messages"]: context += f"{msg['role']}: {msg['content']}\n" elif part["type"] == "episodic": context += "\n## 相关历史\n" for mem in part["memories"]: context += f"- {mem}\n" elif part["type"] == "semantic": context += f"\n## 关于{part['entity']}的知识\n" for fact in part["facts"]: context += f"- {fact}\n" return context 遗忘机制 class ForgettingMechanism: def __init__(self, decay_rate=0.01): self.decay_rate = decay_rate async def decay(self, memory_store): """时间衰减：降低旧记忆的重要性""" now = datetime.now() memories = await memory_store.get_all() for mem in memories: age_days = (now - mem["timestamp"]).days importance = mem.get("importance", 1.0) importance *= (1 - self.decay_rate) ** age_days if importance < 0.1: await memory_store.delete(mem["id"]) else: await memory_store.update(mem["id"], importance=importance) async def consolidate(self, memory_store): """记忆整合：将频繁出现的情景记忆转为语义记忆""" # 找出高频出现的事实 memories = await memory_store.get_all(min_importance=0.5) # 聚类相似记忆 clusters = self.cluster_memories(memories) for cluster in clusters: if len(cluster) >= 3: # 出现3次以上的模式转为知识 summary = await self.summarize_cluster(cluster) await self.semantic_memory.learn(summary) # 降低原始记忆的重要性 for mem in cluster: await memory_store.update(mem["id"], importance=mem["importance"] * 0.3) 存储选择 记忆类型 推荐存储 特点 工作记忆 内存（Redis） 快速读写，无需持久化 情景记忆 向量数据库 语义检索，按相似度召回 语义记忆 图数据库 关系查询，知识推理 结语 Agent记忆系统是构建长期智能助手的核心基础设施。工作记忆处理当前对话，情景记忆保存交互历史，语义记忆积累结构化知识。三层记忆的协同工作加上遗忘和整合机制，让Agent像人类一样"记得住该记的，忘得掉该忘的"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...