生产

引言：像管理团队一样管理 AI CrewAI 的核心理念是：将 AI Agent 组织成一个"团队"（Crew），每个成员有明确的角色、目标和背景故事，通过结构化的流程协作完成任务。2026 年的 CrewAI 3.0 版本在稳定性、可观测性和企业级特性上有了长足进步。 CrewAI 核心概念 三要素模型 Crew = Agents + Tasks + Process 概念 说明 类比 Agent 有角色和能力的 AI 成员 团队成员 Task 需要完成的具体任务 工作任务 Crew Agent 和 Task 的编排单元 项目团队 Process 任务执行方式（顺序/层次） 工作流程 Tool Agent 可使用的工具 工具软件 快速构建第一个 Crew 安装与基础 # pip install crewai crewai-tools from crewai import Agent, Task, Crew, Process from crewai.tools import tool from langchain_openai import ChatOpenAI # 配置 LLM llm = ChatOpenAI(model="gpt-5", temperature=0.7) 定义 Agent # 定义一个内容创作团队 strategist = Agent( role='内容策略师', goal='制定内容策略，确保内容符合品牌调性和目标受众需求', backstory=""" 你是一位拥有 10 年经验的内容策略师，曾在多家科技公司 负责内容规划。你擅长分析受众画像、制定内容日历和优化 内容分发策略。 """, llm=llm, verbose=True, memory=True, # 启用记忆 allow_delegation=False, ) writer = Agent( role='技术写作专家', goal='根据策略师的要求创作高质量的技术内容', backstory=""" 你是一位资深技术写作专家，擅长将复杂的技术概念 转化为通俗易懂的文章。你精通 Markdown 格式， 善于使用代码示例和图表来增强理解。 """, llm=llm, verbose=True, memory=True, ) editor = Agent( role='内容编辑', goal='审核并优化内容，确保质量和准确性', backstory=""" 你是一位严格但公正的内容编辑，关注语法准确性、 逻辑连贯性和事实正确性。你会提出具体的修改建议， 而不是泛泛而谈。 """, llm=llm, verbose=True, ) 定义 Task strategy_task = Task( description=""" 为一篇关于「AI Agent 在企业中的应用」的文章制定内容策略。 要求： 1. 目标受众：技术决策者（CTO、技术总监） 2. 内容长度：3000-5000 字 3. 风格：专业但不晦涩 4. 需要覆盖的关键点： - AI Agent 的定义和价值 - 主流框架对比 - 企业落地案例 - ROI 分析 输出：详细的内容大纲和写作要点。 """, expected_output="一份详细的内容大纲，包含章节结构和每章的要点", agent=strategist, ) writing_task = Task( description=""" 根据策略师的大纲，撰写完整的技术文章。 要求： - 使用 Markdown 格式 - 包含代码示例 - 包含对比表格 - 适当的图表说明 """, expected_output="一篇完整的 Markdown 格式技术文章", agent=writer, context=[strategy_task], # 依赖前一个任务的输出 ) editing_task = Task( description=""" 审查文章并优化： 1. 检查事实准确性 2. 优化段落过渡 3. 修正语法错误 4. 确保风格一致性 5. 输出最终版本 """, expected_output="审查意见 + 最终优化版文章", agent=editor, context=[writing_task], output_file='output/final_article.md', # 保存到文件 ) 组建 Crew 并执行 # 创建 Crew content_crew = Crew( agents=[strategist, writer, editor], tasks=[strategy_task, writing_task, editing_task], process=Process.sequential, # 顺序执行 verbose=True, memory=True, # 团队级记忆 planning=True, # 执行前先规划 # 层次模式配置 # process=Process.hierarchical, # manager_llm=llm, ) # 执行 result = content_crew.kickoff() print(result.raw) 自定义工具集成 创建自定义工具 from crewai.tools import BaseTool from pydantic import BaseModel, Field import requests import json class WebSearchInput(BaseModel): query: str = Field(..., description="搜索关键词") max_results: int = Field(default=5, description="最大返回结果数") class WebSearchTool(BaseTool): name: str = "web_search" description: str = "搜索互联网获取最新信息" args_schema: type[BaseModel] = WebSearchInput def _run(self, query: str, max_results: int = 5) -> str: # 使用搜索 API response = requests.get( "https://api.search.brave.com/res/v1/web/search", headers={"X-Subscription-Token": "YOUR_API_KEY"}, params={"q": query, "count": max_results} ) results = response.json().get("web", {}).get("results", []) formatted = [] for r in results: formatted.append(f"标题: {r['title']}\n链接: {r['url']}\n摘要: {r['description']}\n") return "\n".join(formatted) if formatted else "未找到相关结果" class DatabaseQueryInput(BaseModel): sql: str = Field(..., description="SQL 查询语句") class DatabaseQueryTool(BaseTool): name: str = "database_query" description: str = "查询企业数据库获取业务数据" args_schema: type[BaseModel] = DatabaseQueryInput def _run(self, sql: str) -> str: import sqlite3 conn = sqlite3.connect("data/company.db") cursor = conn.execute(sql) columns = [d[0] for d in cursor.description] rows = cursor.fetchall() conn.close() # 格式化为表格 result = [", ".join(columns)] for row in rows[:20]: # 限制返回行数 result.append(", ".join(str(v) for v in row)) return "\n".join(result) # 将工具分配给 Agent researcher = Agent( role='市场研究员', goal='收集和分析市场数据', backstory='你是市场研究专家，擅长数据收集和分析。', llm=llm, tools=[WebSearchTool(), DatabaseQueryTool()], ) 层次化团队模式 # 层次模式：管理者 Agent 负责任务分配 from crewai import Process manager = Agent( role='项目经理', goal='协调团队成员，确保任务按时高质量完成', backstory='你是一位经验丰富的项目经理，擅长资源分配和进度管理。', llm=llm, allow_delegation=True, # 允许委派任务 ) hierarchical_crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, writing_task, editing_task], process=Process.hierarchical, manager_agent=manager, # 指定管理者 manager_llm=llm, verbose=True, ) 生产级配置 错误处理与重试 from crewai import Crew from crewai.utilities import Logger class ProductionCrew: def __init__(self, config: dict): self.config = config self.max_retries = config.get("max_retries", 3) self.logger = Logger(verbose=True) def run_with_retry(self, crew: Crew, inputs: dict) -> dict: """带重试机制的 Crew 执行""" for attempt in range(self.max_retries): try: result = crew.kickoff(inputs=inputs) return { "status": "success", "result": result.raw, "token_usage": result.token_usage, } except Exception as e: self.logger.log(f"尝试 {attempt+1} 失败: {e}") if attempt < self.max_retries - 1: # 调整重试策略 crew.llm.temperature = min(crew.llm.temperature + 0.1, 1.0) continue return { "status": "failed", "error": str(e), "attempt": attempt + 1, } 监控与可观测性 # 使用 LangSmith 或自定义回调进行监控 from langchain.callbacks import get_openai_callback def monitored_crew_run(crew: Crew, task: str): """带监控的 Crew 执行""" # Token 使用追踪 with get_openai_callback() as cb: result = crew.kickoff(task) metrics = { "total_tokens": cb.total_tokens, "prompt_tokens": cb.prompt_tokens, "completion_tokens": cb.completion_tokens, "total_cost": cb.total_cost, "execution_time": result.execution_time, } # 发送到监控系统 send_metrics_to_dashboard(metrics) # 输出执行报告 print(f""" ═══════════════════════════════════ Crew 执行报告 ═══════════════════════════════════ 任务: {task[:50]}... 状态: 成功 Token 使用: {metrics['total_tokens']:,} 成本: ${metrics['total_cost']:.4f} 耗时: {metrics['execution_time']:.1f}s ═══════════════════════════════════ """) return result, metrics 缓存与优化 from crewai.utilities import CacheHandler import hashlib import json import os class FileCacheHandler(CacheHandler): """文件系统缓存""" def __init__(self, cache_dir: str = "./.crew_cache"): super().__init__() self.cache_dir = cache_dir os.makedirs(cache_dir, exist_ok=True) def _hash_key(self, agent_name: str, input_text: str) -> str: return hashlib.md5(f"{agent_name}:{input_text}".encode()).hexdigest() def get(self, agent_name: str, input_text: str): key = self._hash_key(agent_name, input_text) path = os.path.join(self.cache_dir, f"{key}.json") if os.path.exists(path): with open(path) as f: return json.load(f) return None def set(self, agent_name: str, input_text: str, value): key = self._hash_key(agent_name, input_text) path = os.path.join(self.cache_dir, f"{key}.json") with open(path, "w") as f: json.dump(value, f, ensure_ascii=False) # 使用缓存 crew = Crew( agents=[...], tasks=[...], cache_handler=FileCacheHandler(), # 启用嵌入缓存 embedder={ "provider": "openai", "config": { "model": "text-embedding-3-small", } }, ) 成本优化策略 策略 效果 实现难度 任务分配合适的模型 节省 40-60% 低 启用缓存 节省 20-30% 低 减少冗余上下文 节省 15-25% 中 并行化独立任务 节省 30-50% 时间 中 流式输出 改善用户体验 低 # 模型分层策略 models = { "planning": "gpt-5", # 规划用强模型 "execution": "gpt-4o-mini", # 执行用快速模型 "review": "gpt-5", # 审查用强模型 "summary": "gpt-4o-mini", # 总结用快速模型 } # 为不同 Agent 分配不同模型 strategist = Agent(role='策略师', llm=ChatOpenAI(model=models["planning"])) writer = Agent(role='写作者', llm=ChatOpenAI(model=models["execution"])) editor = Agent(role='编辑', llm=ChatOpenAI(model=models["review"])) 竞品对比 特性 CrewAI AutoGen LangGraph 学习曲线 低 中 高 角色定义 直观 灵活 程序化 工作流模式 顺序/层次 多种 自定义图 内置工具 丰富 基础 需集成 记忆系统 ✅ ✅ 需实现 人类介入 ✅ ✅ ✅ 适合场景 内容/分析 研究/开发 复杂流程 生产成熟度 中高 高 中 最佳实践 角色设计要明确：每个 Agent 的角色、目标和背景故事要有清晰区分 任务粒度适中：太粗导致质量差，太细导致开销大 合理使用上下文依赖：通过 context 参数传递任务间依赖 监控成本：多 Agent 会成倍增加 token 消耗 测试与迭代：先小规模测试，逐步扩展团队 结语 CrewAI 以其直观的角色-任务模型，成为构建 AI 团队最易上手的框架之一。它特别适合内容创作、市场分析、研究报告等"软"任务场景。对于需要精细控制执行流程的场景，可能需要结合其他框架使用。 ...

为什么需要 Prompt 工程化 在原型阶段，Prompt 通常是一个写在代码里的字符串常量。但当应用走向生产，问题开始浮现： 改了一个词，线上效果突然变差，却不知道回退到哪个版本 A/B 测试靠手动切换环境变量，数据散落在日志文件里 新来的同事改了 Prompt，破坏了之前精心设计的 Few-shot 格式 同一个功能有 5 个 Prompt 变体，没人知道哪个在跑 Prompt 工程化的核心目标：让 Prompt 成为可追踪、可测试、可回滚的一等公民。 Prompt 版本管理 目录结构设计 prompts/ ├── config.yaml # 全局配置 ├── chatbot/ # 功能模块 │ ├── meta.yaml # 模块元数据 │ ├── v1.0.0/ # 语义化版本 │ │ ├── system.txt # 系统提示 │ │ ├── few_shot.jsonl # Few-shot 示例 │ │ └── config.yaml # 模型参数 │ ├── v1.1.0/ │ │ ├── system.txt │ │ ├── few_shot.jsonl │ │ └── config.yaml │ └── v2.0.0/ # 大版本变更 │ ├── system.txt │ ├── few_shot.jsonl │ └── config.yaml └── classifier/ └── ... Prompt 注册中心实现 import yaml import json from pathlib import Path from dataclasses import dataclass, field from typing import Any @dataclass class PromptVersion: """单个 Prompt 版本""" name: str version: str system: str few_shot: list[dict] = field(default_factory=list) model: str = "gpt-4o-mini" temperature: float = 0.7 max_tokens: int = 1024 status: str = "active" # draft | testing | active | archived def render(self, user_input: str) -> list[dict]: """渲染为 API 消息格式""" messages = [{"role": "system", "content": self.system}] for example in self.few_shot: messages.append({"role": example["role"], "content": example["content"]}) messages.append({"role": "user", "content": user_input}) return messages class PromptRegistry: """Prompt 版本注册中心""" def __init__(self, base_dir: str = "prompts"): self.base_dir = Path(base_dir) self._cache: dict[str, PromptVersion] = {} def load(self, name: str, version: str = "latest") -> PromptVersion: """加载指定版本的 Prompt""" if version == "latest": version = self._get_latest_version(name) cache_key = f"{name}@{version}" if cache_key in self._cache: return self._cache[cache_key] version_dir = self.base_dir / name / f"v{version}" # 加载系统提示 system = (version_dir / "system.txt").read_text(encoding="utf-8") # 加载 Few-shot few_shot = [] few_shot_path = version_dir / "few_shot.jsonl" if few_shot_path.exists(): for line in few_shot_path.read_text(encoding="utf-8").strip().split("\n"): few_shot.append(json.loads(line)) # 加载配置 config_path = version_dir / "config.yaml" config = yaml.safe_load(config_path.read_text(encoding="utf-8")) pv = PromptVersion( name=name, version=version, system=system, few_shot=few_shot, model=config.get("model", "gpt-4o-mini"), temperature=config.get("temperature", 0.7), max_tokens=config.get("max_tokens", 1024), status=config.get("status", "active"), ) self._cache[cache_key] = pv return pv def _get_latest_version(self, name: str) -> str: """获取最新 active 版本""" module_dir = self.base_dir / name versions = [] for d in module_dir.iterdir(): if d.is_dir() and d.name.startswith("v"): config = yaml.safe_load((d / "config.yaml").read_text(encoding="utf-8")) if config.get("status") == "active": versions.append(d.name[1:]) # 去掉 'v' 前缀 versions.sort(key=lambda v: [int(x) for x in v.split(".")]) return versions[-1] if versions else "1.0.0" def diff(self, name: str, v1: str, v2: str) -> dict: """对比两个版本的差异""" p1 = self.load(name, v1) p2 = self.load(name, v2) return { "system_changed": p1.system != p2.system, "few_shot_changed": p1.few_shot != p2.few_shot, "model_changed": p1.model != p2.model, "temperature_changed": p1.temperature != p2.temperature, } # 使用示例 registry = PromptRegistry("prompts") prompt = registry.load("chatbot", "latest") messages = prompt.render("你好，帮我查一下订单") 版本管理规范 版本类型 变更内容 示例 Major (x.0.0) Prompt 结构重构、角色定义变更 从单轮改为多轮对话 Minor (1.x.0) Few-shot 增删、指令逻辑调整 新增 2 个示例 Patch (1.0.x) 文案微调、错别字修正 “请” → “请帮我” A/B 测试框架 import random import hashlib from dataclasses import dataclass, field from collections import defaultdict import time @dataclass class ABTestConfig: """A/B 测试配置""" test_name: str variants: dict[str, PromptVersion] # variant_name -> Prompt traffic_split: dict[str, float] # variant_name -> 流量比例 metrics: list[str] = field(default_factory=lambda: [ "user_satisfaction", "response_length", "latency_ms", "cost" ]) min_sample_size: int = 100 def assign(self, user_id: str) -> str: """基于用户 ID 确定性分配变体（同一用户始终进入同一组）""" hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) ratio = (hash_val % 10000) / 10000.0 cumulative = 0.0 for variant, weight in self.traffic_split.items(): cumulative += weight if ratio < cumulative: return variant return list(self.traffic_split.keys())[-1] @dataclass class ExperimentResult: variant: str user_id: str metric: str value: float timestamp: float = field(default_factory=time.time) class ABTestRunner: """A/B 测试运行器""" def __init__(self): self.results: list[ExperimentResult] = [] def run( self, test_config: ABTestConfig, user_id: str, user_input: str, execute_fn, # callable: (PromptVersion, str) -> dict ) -> dict: """执行一次 A/B 测试请求""" variant_name = test_config.assign(user_id) prompt = test_config.variants[variant_name] # 执行并收集指标 result = execute_fn(prompt, user_input) # 记录指标 for metric in test_config.metrics: if metric in result: self.results.append(ExperimentResult( variant=variant_name, user_id=user_id, metric=metric, value=result[metric], )) return {"variant": variant_name, "result": result} def analyze(self, test_name: str) -> dict: """分析实验结果""" stats = defaultdict(lambda: defaultdict(list)) for r in self.results: stats[r.variant][r.metric].append(r.value) report = {} for variant, metrics in stats.items(): report[variant] = {} for metric, values in metrics.items(): vals = sorted(values) report[variant][metric] = { "count": len(vals), "mean": sum(vals) / len(vals), "median": vals[len(vals) // 2], "p95": vals[int(len(vals) * 0.95)] if len(vals) > 20 else None, } return report def is_significant(self, test_name: str, metric: str, alpha: float = 0.05) -> bool: """简单的统计显著性检验（Z 检验）""" import math variants = [r.variant for r in self.results if r.metric == metric] if len(set(variants)) < 2: return False # 按 variant 分组 groups = defaultdict(list) for r in self.results: if r.metric == metric: groups[r.variant].append(r.value) if len(groups) < 2: return False v1, v2 = list(groups.keys())[:2] s1, s2 = groups[v1], groups[v2] if len(s1) < 30 or len(s2) < 30: return False # 样本不足 m1, m2 = sum(s1) / len(s1), sum(s2) / len(s2) var1 = sum((x - m1) ** 2 for x in s1) / len(s1) var2 = sum((x - m2) ** 2 for x in s2) / len(s2) se = math.sqrt(var1 / len(s1) + var2 / len(s2)) if se == 0: return False z = abs(m1 - m2) / se return z > 1.96 # 95% 置信度 # 使用示例 runner = ABTestRunner() test_config = ABTestConfig( test_name="chatbot_tone_v2", variants={ "control": registry.load("chatbot", "1.0.0"), "treatment": registry.load("chatbot", "1.1.0"), }, traffic_split={"control": 0.5, "treatment": 0.5}, ) def execute_fn(prompt: PromptVersion, user_input: str) -> dict: # 实际调用 LLM return { "user_satisfaction": 4.5, # 用户评分 "response_length": 320, "latency_ms": 850, "cost": 0.003, } # 模拟 200 次请求 for i in range(200): runner.run(test_config, f"user-{i}", "帮我查订单", execute_fn) # 分析结果 report = runner.analyze("chatbot_tone_v2") for variant, metrics in report.items(): print(f"\n=== {variant} ===") for metric, stats in metrics.items(): print(f" {metric}: mean={stats['mean']:.2f}, p95={stats['p95']}") print(f"\n统计显著: {runner.is_significant('chatbot_tone_v2', 'user_satisfaction')}") 回归评测体系 每次 Prompt 变更前，必须通过回归测试集的验证。 ...

引言 RAG（Retrieval-Augmented Generation）是当前最流行的 LLM 应用架构之一。然而，从 Demo 到生产之间横亘着巨大的鸿沟。本文基于多个 RAG 生产项目的实战经验，总结 12 个最常见、最致命的坑，并给出经过验证的解决方案。 坑 1：文档分块策略不当 问题 天真地按固定长度分块（如每 512 字符），导致： 语义被截断（一个完整的段落从中间切断） 关键信息分散在多个块中，检索时只命中一部分 表格和列表被拆碎，失去结构信息 解决方案：语义分块 + 重叠窗口 from dataclasses import dataclass from typing import List import re @dataclass class Chunk: text: str metadata: dict token_count: int = 0 class SemanticChunker: """基于语义边界的智能分块器""" def __init__( self, target_size: int = 400, # 目标块大小（tokens） min_size: int = 100, # 最小块大小 max_size: int = 600, # 最大块大小 overlap: int = 50, # 重叠区间 ): self.target_size = target_size self.min_size = min_size self.max_size = max_size self.overlap = overlap def chunk_document(self, text: str, source: str = "") -> List[Chunk]: """分块主流程""" # Step 1: 按结构边界切分 sections = self._split_by_structure(text) # Step 2: 对每个 section 按 paragraph 切分 paragraphs = [] for section in sections: paragraphs.extend(self._split_by_paragraph(section)) # Step 3: 合并过小的段落，拆分过大的段落 chunks = self._merge_and_split(paragraphs) # Step 4: 添加重叠 chunks = self._add_overlap(chunks) # Step 5: 附加元数据 return [ Chunk( text=c, metadata={"source": source, "chunk_index": i, "total_chunks": len(chunks)}, token_count=len(c) // 2, # 粗略估算 ) for i, c in enumerate(chunks) ] def _split_by_structure(self, text: str) -> List[str]: """按标题、分隔符等结构边界切分""" # 按 Markdown 标题切分 pattern = r'(?=^#{1,6}\s)' sections = re.split(pattern, text, flags=re.MULTILINE) return [s.strip() for s in sections if s.strip()] def _split_by_paragraph(self, text: str) -> List[str]: """按段落（双换行）切分""" paras = text.split("\n\n") return [p.strip() for p in paras if p.strip()] def _merge_and_split(self, paragraphs: List[str]) -> List[str]: """合并过小段落，拆分过大段落""" chunks = [] buffer = [] buffer_size = 0 for para in paragraphs: para_size = len(para) // 2 # 粗略 token 估算 if buffer_size + para_size > self.max_size and buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if para_size > self.max_size: # 单段落过长，按句子切分 if buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 sentences = re.split(r'(?<=[。！？.!?])\s+', para) sent_buffer = [] sent_size = 0 for sent in sentences: if sent_size + len(sent) // 2 > self.max_size and sent_buffer: chunks.append(" ".join(sent_buffer)) sent_buffer = [] sent_size = 0 sent_buffer.append(sent) sent_size += len(sent) // 2 if sent_buffer: chunks.append(" ".join(sent_buffer)) else: buffer.append(para) buffer_size += para_size if buffer_size >= self.target_size: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if buffer: chunks.append("\n\n".join(buffer)) return chunks def _add_overlap(self, chunks: List[str]) -> List[str]: """为相邻块添加重叠""" if self.overlap <= 0 or len(chunks) <= 1: return chunks result = [chunks[0]] for i in range(1, len(chunks)): prev_text = chunks[i - 1] overlap_text = prev_text[-self.overlap * 2:] # 粗略取后半段 result.append(overlap_text + " " + chunks[i]) return result # 使用示例 chunker = SemanticChunker(target_size=400, overlap=50) document = open("knowledge_base/product_manual.md").read() chunks = chunker.chunk_document(document, source="product_manual.md") print(f"分块完成: {len(chunks)} 个块") 分块策略对比 策略 优点 缺点 适用场景 固定长度 实现简单 语义截断 不推荐 按段落 保持语义 块大小不均 短文档 语义分块 语义完整 实现复杂 通用推荐 按文档结构 保持层级 需要结构化输入 Markdown/HTML 递归分块 灵活适配 可控性差 混合内容 坑 2：Embedding 模型与 LLM 不匹配 问题 用 OpenAI 的 text-embedding-3-large 做向量，但生成用的是 Claude 模型。两者对语义的理解不同，可能导致检索到的内容并非生成模型"认为"最相关的。 ...

Agent 模式全景 复杂度 → 单Agent ──→ Router ──→ Fan-out/Fan-in ──→ Hierarchical ──→ Multi-Agent │ │ │ │ │ │ 简单任务 │ 按需分发 │ 并行处理 │ 管理者+执行者 │ 自主协作 │ │ │ │ │ └────────────┴──────────────┴──────────────────┴────────────────┘ 模式 适用场景 复杂度 延迟 错误恢复 单 Agent 简单问答、单一任务 低 低 简单重试 Router 多领域问答、任务分类 中 低+路由 单分支重试 Fan-out/Fan-in 并行研究、批量处理 中 高(并行) 部分失败容忍 Hierarchical 复杂项目、多步骤流程 高 高 子任务级恢复 Multi-Agent 开放式协作、辩论 极高 最高 最复杂 单 Agent 架构 最基础的形态：一个 Agent + 一组工具。 ...

为什么需要实战案例 了解 Codex 的功能很重要，但更重要的是知道在真实生产环境中如何使用它。本文收录了 10 个经过验证的生产场景，每个场景包含：业务背景、Codex 配置、执行流程、代码示例和效果数据。 案例 1：自动化运维巡检 业务背景 某互联网公司有 20 台服务器，需要每天检查 CPU/内存/磁盘/服务状态，过去由运维工程师手动执行，每天耗时 1.5 小时。 Codex 配置 { "task": "运维巡检", "schedule": "0 9 * * *", "model": "deepseek-chat", "skills": ["ssh-tools", "report-generator"] } 执行流程 # Codex 执行的巡检脚本 import subprocess import json from datetime import datetime servers = json.load(open("servers.json")) report = [] for server in servers: # SSH 执行远程命令 cpu = subprocess.check_output( f"ssh {server['user']}@{server['host']} 'top -bn1 | grep Cpu'", shell=True ).decode() disk = subprocess.check_output( f"ssh {server['user']}@{server['host']} 'df -h /'", shell=True ).decode() services = subprocess.check_output( f"ssh {server['user']}@{server['host']} 'systemctl is-active nginx postgresql redis'", shell=True ).decode() report.append({ "server": server['name'], "cpu": parse_cpu(cpu), "disk": parse_disk(disk), "services": services.strip().split('\n'), "timestamp": datetime.now().isoformat() }) # 生成报告 generate_report(report, format="markdown", output="daily_check.md") # 发送邮件 send_email("ops@company.com", "每日巡检报告", "daily_check.md") 效果 指标 优化前 优化后 耗时 1.5小时/天 3分钟/天 漏检率 8% 0% 报告格式 不统一 标准化 人力成本 1.5人时/天 0 案例 2：数据分析报告自动化 业务背景 市场团队每周需要从 3 个数据源（Google Analytics、内部数据库、CRM API）拉取数据，生成周报。 ...

Demo 与生产的差距 Function Calling 的 demo 很简单：定义一个函数，LLM 返回参数，调用函数，返回结果。但在生产中，你需要处理： LLM 返回了不存在的函数 参数类型不对、必填字段缺失 函数调用超时 并行调用之间的依赖冲突 用户通过参数注入恶意命令 调用链过深导致上下文爆炸 Schema 设计：函数定义是契约 好的 Schema 长什么样 # ❌ 糟糕的 Schema: 模糊、无约束 BAD_SCHEMA = { "name": "search", "description": "搜索东西", "parameters": { "type": "object", "properties": { "query": {"type": "string"} } } } # ✅ 好的 Schema: 精确、有约束、有枚举 GOOD_SCHEMA = { "name": "search_product", "description": "在商品库中搜索商品。当用户询问有没有某种商品、价格、库存时使用此函数。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，从用户意图中提取，去除无意义词" }, "category": { "type": "string", "enum": ["electronics", "clothing", "food", "books", "other"], "description": "商品类别" }, "max_price": { "type": "number", "minimum": 0, "description": "用户能接受的最高价格(元)" }, "sort_by": { "type": "string", "enum": ["price_asc", "price_desc", "relevance", "sales"], "default": "relevance" } }, "required": ["query"], "additionalProperties": False } } Schema 设计原则 原则 说明 示例 名字即文档 函数名要自解释 search_product > search description 要写触发条件 不是描述函数做什么，而是何时用 “当用户询问商品价格时使用” 枚举优于自由文本 能枚举就不要用 string category 用 enum 设默认值 减少LLM猜测 sort_by 默认 relevance 禁止额外属性 additionalProperties: false 防止 LLM 编造参数 范围约束 用 min/max 限制数值 max_price minimum: 0 参数验证：不要信任 LLM 的输出 import json from pydantic import BaseModel, Field, ValidationError from typing import Optional, Literal # 用 Pydantic 做二次验证 class SearchProductParams(BaseModel): query: str = Field(..., min_length=1, max_length=100) category: Optional[Literal["electronics", "clothing", "food", "books", "other"]] = None max_price: Optional[float] = Field(None, ge=0, le=1000000) sort_by: Literal["price_asc", "price_desc", "relevance", "sales"] = "relevance" class FunctionCallValidator: def __init__(self): self.schemas = { "search_product": SearchProductParams, } def validate(self, function_name: str, arguments: dict) -> tuple[bool, any]: schema_cls = self.schemas.get(function_name) if not schema_cls: return False, {"error": f"Unknown function: {function_name}"} try: validated = schema_cls(**arguments) return True, validated except ValidationError as e: return False, {"error": e.errors()} 处理 LLM 返回的常见问题 class RobustFunctionExecutor: async def execute(self, llm_response: dict) -> dict: tool_calls = llm_response.get("tool_calls", []) results = [] for call in tool_calls: name = call["function"]["name"] # 1. 函数是否存在? if name not in self.registry: results.append(self._error_result(call, f"未知函数: {name}")) continue # 2. 参数解析 try: args = json.loads(call["function"]["arguments"]) except json.JSONDecodeError: # LLM 返回了非法 JSON，尝试修复 args = self._repair_json(call["function"]["arguments"]) if args is None: results.append(self._error_result(call, "参数 JSON 解析失败")) continue # 3. 参数验证 ok, validated = self.validator.validate(name, args) if not ok: results.append(self._error_result(call, str(validated))) continue # 4. 执行(带超时) try: result = await asyncio.wait_for( self.registry[name](**validated.dict()), timeout=10 ) results.append({"tool_call_id": call["id"], "result": result}) except asyncio.TimeoutError: results.append(self._error_result(call, "函数执行超时")) except Exception as e: results.append(self._error_result(call, f"执行错误: {str(e)}")) return results def _repair_json(self, broken: str) -> dict | None: """尝试修复 LLM 输出的破损 JSON""" # 去除尾部逗号 fixed = broken.rstrip().rstrip(",") # 补全括号 open_braces = fixed.count("{") - fixed.count("}") open_brackets = fixed.count("[") - fixed.count("]") fixed += "}" * open_braces + "]" * open_brackets try: return json.loads(fixed) except: return None 错误恢复：让 LLM 从错误中学习 async def function_call_loop(messages: list, max_rounds: int = 5): """多轮函数调用循环，带错误恢复""" for round_num in range(max_rounds): response = await client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=tool_definitions, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content # LLM 认为完成了 # 执行所有工具调用 for tool_call in msg.tool_calls: success, result = await executor.execute_single(tool_call) # 把结果（包括错误）反馈给 LLM messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) if not success: # 在错误信息中给出修复建议 messages.append({ "role": "system", "content": f"函数 {tool_call.function.name} 调用失败: {result['error']}。请修正参数后重试，或换一种方式回答。" }) return "达到最大调用轮数，请简化请求。" 并行调用与依赖管理 import asyncio from typing import Dict, List, Set class ParallelCallOrchestrator: def __init__(self): # 声明函数间的依赖关系 self.dependencies = { "get_order_detail": [], # 无依赖 "check_inventory": [], # 无依赖 "calculate_shipping": ["check_inventory"], # 依赖库存检查 "create_order": ["get_order_detail", "check_inventory"], } async def execute_parallel(self, tool_calls: list) -> list: # 构建依赖图 independent = [] dependent = {} for call in tool_calls: name = call["function"]["name"] deps = self.dependencies.get(name, []) if not deps: independent.append(call) else: dependent[call["id"]] = {"call": call, "deps": set(deps)} # 先执行无依赖的 results = {} tasks = [self._execute_and_store(call, results) for call in independent] await asyncio.gather(*tasks, return_exceptions=True) # 再执行有依赖的（拓扑排序） while dependent: ready = [ cid for cid, info in dependent.items() if info["deps"].issubset(set(results.keys())) ] if not ready: # 有循环依赖，强制执行剩余的 ready = list(dependent.keys()) tasks = [] for cid in ready: tasks.append(self._execute_and_store(dependent[cid]["call"], results)) del dependent[cid] await asyncio.gather(*tasks, return_exceptions=True) return list(results.values()) 安全沙箱：永远不要直接执行 LLM 生成的代码 import subprocess import tempfile import os class CodeExecutionSandbox: """安全执行 LLM 生成的代码""" def __init__(self): self.allowed_modules = {"math", "statistics", "json", "re"} self.timeout = 5 self.memory_limit = "256m" async def execute_python(self, code: str) -> dict: # 1. 静态检查 violations = self._static_check(code) if violations: return {"error": "安全检查失败", "violations": violations} # 2. 在 Docker 容器中执行 with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(code) f.flush() try: result = subprocess.run( ["docker", "run", "--rm", "--memory", self.memory_limit, "--cpus", "0.5", "--network", "none", # 无网络 "--read-only", # 只读文件系统 "--tmpfs", "/tmp:size=10m", # 临时目录 "-v", f"{f.name}:/app/code.py:ro", "python:3.12-slim", "python", "/app/code.py"], capture_output=True, timeout=self.timeout, text=True ) return { "stdout": result.stdout[:5000], # 截断 "stderr": result.stderr[:2000], "exit_code": result.returncode } except subprocess.TimeoutExpired: return {"error": "执行超时"} finally: os.unlink(f.name) def _static_check(self, code: str) -> list: violations = [] # 检查 import for line in code.split("\n"): if "import" in line: module = line.split("import")[-1].strip().split(".")[0] if module not in self.allowed_modules: violations.append(f"禁止导入模块: {module}") # 检查危险函数 dangerous = ["open(", "exec(", "eval(", "os.system", "subprocess", "__import__"] for d in dangerous: if d in code: violations.append(f"禁止使用: {d}") return violations 审计日志：记录每次调用 import logging from datetime import datetime logger = logging.getLogger("function_caller") class AuditLogger: def log_call(self, function_name: str, arguments: dict, result: dict, duration_ms: float, success: bool): logger.info(json.dumps({ "timestamp": datetime.utcnow().isoformat(), "function": function_name, "arguments": self._sanitize(arguments), # 脱敏 "result_size": len(str(result)), "duration_ms": duration_ms, "success": success, "trace_id": self._get_trace_id(), }, ensure_ascii=False)) def _sanitize(self, args: dict) -> dict: """脱敏处理""" sensitive_keys = {"password", "token", "api_key", "credit_card"} sanitized = {} for k, v in args.items(): if k.lower() in sensitive_keys: sanitized[k] = "***REDACTED***" else: sanitized[k] = v return sanitized 性能优化 函数定义缓存 # 工具定义不要每次请求都重新构建 from functools import lru_cache @lru_cache(maxsize=1) def get_tool_definitions(): return [ {"type": "function", "function": schema} for schema in load_all_schemas() ] 函数调用结果缓存 import hashlib from datetime import timedelta class ResultCache: def __init__(self, redis_client): self.redis = redis_client self.ttl = timedelta(minutes=10) async def get_or_execute(self, func_name: str, args: dict, executor): # 对幂等函数做缓存 cache_key = self._make_key(func_name, args) cached = await self.redis.get(cache_key) if cached: return json.loads(cached) result = await executor(func_name, args) await self.redis.setex( cache_key, int(self.ttl.total_seconds()), json.dumps(result, ensure_ascii=False) ) return result def _make_key(self, name: str, args: dict) -> str: arg_hash = hashlib.md5(json.dumps(args, sort_keys=True).encode()).hexdigest() return f"fc:{name}:{arg_hash}" 总结 Function Calling 在生产中可靠运行的关键：严格的 Schema 设计是契约，二次验证是防线，错误恢复是韧性，并行管理是效率，安全沙箱是底线，审计日志是追溯。把每个 LLM 返回的函数调用都当作不可信输入来处理，就能避免大部分生产事故。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么需要安全检查清单 LLM 应用引入了全新的攻击面：Prompt 注入、训练数据泄露、模型越狱、有害内容生成。传统 Web 安全检查清单覆盖不了这些。本文提供 50 项检查项，分为 5 大类，每一项都必须在上线前确认。 一、输入安全（12 项） Prompt 注入防护 # 检查项 风险等级 验证方式 1 用户输入与系统指令分离（使用 system role） 高 审查 prompt 结构 2 用户输入被包裹在分隔符中（如 <user_input>） 高 审查 prompt 模板 3 系统指令中包含"忽略以上指令"的防御声明 中 审查 system prompt 4 对用户输入做长度限制（建议 < 10K 字符） 中 压测验证 5 过滤已知的 Prompt 注入模式 高 红队测试 # Prompt 注入检测器 class PromptInjectionGuard: INJECTION_PATTERNS = [ r"ignore\s+(all\s+)?(previous|above|prior)\s+instructions", r"you\s+are\s+now\s+(a|an)\s+\w+", r"system\s*:\s*", r"<\|im_start\|>", r"forgot\s+your\s+rules", r"reveal\s+your\s+(system\s+)?prompt", ] def check(self, user_input: str) -> tuple[bool, list]: import re violations = [] for pattern in self.INJECTION_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): violations.append(pattern) return len(violations) == 0, violations PII 保护 # 检查项 风险等级 验证方式 6 输入中的 PII 被检测并脱敏 高 传入含 PII 的测试用例 7 PII 脱敏日志开启（不记录原始 PII） 高 审查日志配置 8 不会将用户输入原样发送给第三方服务 高 审查 API 调用链 import re class PIIScrubber: PATTERNS = { "phone": (r"\b1[3-9]\d{9}\b", "[PHONE]"), "email": (r"\b[\w.-]+@[\w.-]+\.\w+\b", "[EMAIL]"), "id_card": (r"\b\d{17}[\dXx]\b", "[ID_CARD]"), "bank_card": (r"\b\d{16,19}\b", "[BANK_CARD]"), } def scrub(self, text: str) -> str: for pii_type, (pattern, replacement) in self.PATTERNS.items(): text = re.sub(pattern, replacement, text) return text 内容安全 # 检查项 风险等级 验证方式 9 输入内容分类（是否包含暴力/违法内容） 高 红队测试 10 对敏感话题有预设的拒绝策略 中 审查 system prompt 11 支持多语言输入的过滤 中 多语言测试 12 对图片输入有内容审核（多模态场景） 高 上传违规模拟图 二、输出安全（10 项） 有害内容过滤 # 检查项 风险等级 验证方式 13 输出经过毒性检测模型 高 触发性测试用例 14 输出经过偏见/歧视检测 高 公平性测试集 15 对特定领域（医疗/法律/金融）有免责声明 中 审查输出模板 16 不输出可执行代码的直接运行结果 中 代码注入测试 class OutputSafetyFilter: def __init__(self, toxicity_model, threshold=0.7): self.model = toxicity_model self.threshold = threshold async def filter(self, response: str) -> tuple[str, bool]: # 毒性检测 toxicity = await self.model.predict(response) if toxicity > self.threshold: return self._safe_fallback(), False # 敏感信息泄露检测 if self._detect_leaked_info(response): return self._safe_fallback(), False return response, True def _detect_leaked_info(self, text): """检测输出中是否泄露了系统信息""" sensitive_patterns = [ r"api[_-]?key", r"sk-[a-zA-Z0-9]{20,}", r"password", r"secret", r"token", r"/[a-z]:\\users\\", # Windows 路径 ] return any( re.search(p, text, re.IGNORECASE) for p in sensitive_patterns ) 输出完整性 # 检查项 风险等级 验证方式 17 输出有长度上限（防止 Token 爆炸） 中 触发长输出测试 18 流式输出有超时保护 中 模拟慢速输出 19 输出格式校验（JSON/XML 是否合法） 中 格式错误注入测试 20 输出不包含训练数据原文（版权风险） 中 已知文本检索 21 输出不包含 Prompt 模板内容（提示泄露） 高 “重复你的指令"测试 22 输出经过 PII 二次过滤 高 PII 回显测试 三、模型安全（10 项） 越狱防护 # 检查项 风险等级 验证方式 23 通过越狱测试集（DAN/Roleplay 编码等） 高 自动化越狱测试 24 对多轮对话有累积风险检测 高 多轮越狱攻击测试 25 对编码/混淆输入有解码检测 中 Base64/Unicode 混淆测试 class JailbreakDetector: ENCODING_PATTERNS = [ (r"base64:", self._decode_base64), (r"\\u[0-9a-fA-F]{4}", self._decode_unicode), (r"\\x[0-9a-fA-F]{2}", self._decode_hex), ] async def check(self, user_input: str): # 1. 检查编码内容 decoded = self._try_decode(user_input) if decoded != user_input: # 对解码后的内容也做注入检测 safe, _ = PromptInjectionGuard().check(decoded) if not safe: return False, "Encoded injection detected" # 2. 检查角色扮演越狱 roleplay_patterns = [ r"pretend you are", r"act as (if you are )?DAN", r"you are (in )?developer mode", r"jailbreak", ] for pattern in roleplay_patterns: if re.search(pattern, user_input, re.IGNORECASE): return False, f"Roleplay jailbreak: {pattern}" return True, None 模型隔离 # 检查项 风险等级 验证方式 26 不同租户的会话隔离 高 跨租户数据泄露测试 27 对话历史有长度限制（防止上下文污染） 中 长对话测试 28 Function Calling 参数有白名单校验 高 构造恶意参数测试 29 模型输出不直接执行（需人工/代码确认） 高 审查执行链路 30 有模型使用量配额（防止滥用） 中 超额测试 31 模型版本变更经过安全评估 中 审查变更流程 32 对抗样本检测（异常输入模式） 中 对抗测试集 四、基础设施安全（10 项） # 检查项 风险等级 验证方式 33 API Key 存储在密钥管理服务（非代码/配置文件） 高 审查部署配置 34 API 通信全程 HTTPS/TLS 1.2+ 高 SSL 扫描 35 对 LLM API 调用有网络白名单限制 中 审查网络策略 36 向量数据库有访问控制 高 审查 DB ACL 37 Embedding 服务有认证 中 未认证调用测试 38 日志中不包含 API Key / 完整 Prompt 高 审查日志输出 39 监控系统有异常调用检测（频率/内容） 中 审查告警规则 40 容器/进程以最小权限运行 中 审查 K8s manifest 41 模型文件有完整性校验 中 校验和验证 42 有 DDoS 防护（CDN/WAF） 中 审查网络架构 五、合规审计（8 项） # 检查项 风险等级 验证方式 43 所有 LLM 调用有审计日志（who/when/what/model） 高 审查日志格式 44 用户知情同意（明确告知使用 AI） 高 审查 UI/ToS 45 数据保留策略明确（日志/对话历史保留期限） 高 审查数据策略 46 支持用户数据删除请求（GDPR/PIPL） 高 删除流程测试 47 模型训练数据来源可追溯 中 审查文档 48 有 AI 生成内容标识（水印/声明） 中 审查输出格式 49 定期安全评估（至少每季度） 中 审查评估记录 50 有应急响应预案（模型输出有害内容时） 高 审查 IR 计划 自动化检查脚本 class SecurityChecklist: """可自动化的安全检查项""" CHECKS = { "input_length_limit": {"auto": True, "critical": False}, "prompt_injection_filter": {"auto": True, "critical": True}, "pii_scrubbing": {"auto": True, "critical": True}, "output_toxicity_check": {"auto": True, "critical": True}, "output_length_limit": {"auto": True, "critical": False}, "jailbreak_resistance": {"auto": True, "critical": True}, "prompt_leak_prevention": {"auto": True, "critical": True}, "tls_check": {"auto": True, "critical": True}, "log_pii_check": {"auto": True, "critical": True}, "rate_limiting": {"auto": True, "critical": False}, } async def run_all(self, target_url, api_key): results = {} for check_name, config in self.CHECKS.items(): if not config["auto"]: continue method = getattr(self, f"check_{check_name}") try: passed, detail = await method(target_url, api_key) results[check_name] = { "passed": passed, "detail": detail, "critical": config["critical"], } except Exception as e: results[check_name] = { "passed": False, "detail": f"Check error: {e}", "critical": config["critical"], } # 汇总报告 total = len(results) passed = sum(1 for r in results.values() if r["passed"]) critical_fail = [ k for k, v in results.items() if not v["passed"] and v["critical"] ] return { "total": total, "passed": passed, "failed": total - passed, "critical_failures": critical_fail, "ready_for_production": len(critical_fail) == 0, "details": results, } 上线审批流程 开发完成 → 自动化检查（38 项可自动化） │ 全部通过？ ├─ 是 → 人工审查（12 项需人工） │ ├─ 全部通过 → 安全团队签字 → 上线 │ └─ 有问题 → 整改 → 重新审查 └─ 否 → 修复 → 重新自动检查 总结 LLM 安全不是可选项。50 项检查中，标记为"高风险"的约 25 项——这些是上线阻断项，不通过不上线。关键检查包括：Prompt 注入防护、PII 脱敏、越狱抵抗、输出过滤、密钥管理。自动化检查覆盖约 76% 的项目，剩余需人工审查。建议将检查清单集成到 CI/CD 流水线中，每次部署前自动运行。安全是一场持续的攻防战，检查清单只是起点，不是终点。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

原型 vs 生产：到底差在哪 在 ChatGPT 界面里跑通一个 demo，和生产环境服务千万用户，中间隔着一条鸿沟。原型阶段你不需要关心的事情，在生产中每一个都会变成事故： 维度 原型阶段 生产阶段 延迟 手动等待，几秒可接受 P99 < 3s，超时降级 可用性 失败就重试 99.9% SLA，多活容灾 成本 不关心 每月 API 账单可预测 安全 Prompt 直接硬编码 密钥管理、输入消毒、输出过滤 可观测性 print 大法 结构化日志、指标、链路追踪 一致性 每次回答不同 温度控制、缓存、确定性输出 核心认知：原型验证的是"能不能做"，生产回答的是"能不能稳定做、能不能赚钱做、能不能安全做"。 API 选型：不只是选个模型 模型选择决策树 def choose_model(task: str, latency_req: float, budget: float) -> str: if task in ["简单分类", "抽取"] and latency_req < 0.5: return "gpt-4o-mini" # 快且便宜 if task in ["复杂推理", "代码生成"] and budget > 0.05: return "gpt-4o" # 能力优先 if task in ["长文档总结"] and budget < 0.02: return "gpt-4o-mini" # 上下文够长，成本可控 return "gpt-4o-mini" # 默认保守 成本估算公式 单次调用成本 = (input_tokens × input_price) + (output_tokens × output_price) 月度预算 = 日均调用量 × 平均(input_tokens + output_tokens) × 30 × 单价 # GPT-4o (2026 pricing) # Input: $2.50 / 1M tokens # Output: $10.00 / 1M tokens # 一个典型客服对话: ~500 input + ~200 output # 单次成本 ≈ $0.00325 # 日均 10000 次 → 月成本 ≈ $975 多模型路由策略 生产环境不要只用一个模型。根据请求复杂度路由到不同模型： ...