Prompt工程

为什么Prompt需要A/B测试 “我觉得这个Prompt更好”——这是Prompt工程中最常见也最危险的句子。直觉在Prompt优化中往往不可靠：一个看起来更精巧的Prompt可能在生产环境中表现更差。 真实案例： 某电商团队将客服Prompt从"详细回复"版本切换到"简洁回复"版本，直觉上认为用户更喜欢简洁。A/B测试结果：简洁版本的客户满意度下降了12%，因为用户需要多次追问才能解决问题。 数据驱动的Prompt优化不是可选项——它是生产系统的必需品。 Prompt版本控制 语义化版本控制 from dataclasses import dataclass from typing import Optional import hashlib @dataclass class PromptVersion: """Prompt版本""" major: int # 不兼容的修改（如输出格式变更） minor: int # 向后兼容的功能新增 patch: int # Bug修复和微调 hash: str # 内容哈希 @property def version_string(self) -> str: return f"v{self.major}.{self.minor}.{self.patch}" @staticmethod def compute_hash(content: str) -> str: return hashlib.sha256(content.encode()).hexdigest()[:12] class VersionedPrompt: """带版本控制的Prompt""" def __init__(self): self.versions: list[dict] = [] self.active_version: Optional[str] = None def commit(self, prompt_content: str, change_type: str = "patch", changelog: str = "") -> str: """ 提交新版本 change_type: major / minor / patch """ # 确定版本号 if not self.versions: version = PromptVersion(1, 0, 0, "") else: latest = self.versions[-1]["version"] if change_type == "major": version = PromptVersion(latest.major + 1, 0, 0, "") elif change_type == "minor": version = PromptVersion(latest.major, latest.minor + 1, 0, "") else: version = PromptVersion(latest.major, latest.minor, latest.patch + 1, "") version.hash = PromptVersion.compute_hash(prompt_content) version_id = f"{version.version_string}-{version.hash}" self.versions.append({ "version_id": version_id, "version": version, "content": prompt_content, "changelog": changelog, "timestamp": datetime.now(), "is_active": False }) return version_id def rollback(self, version_id: str): """回滚到指定版本""" for v in self.versions: v["is_active"] = (v["version_id"] == version_id) self.active_version = version_id def diff(self, version_a: str, version_b: str) -> str: """比较两个版本的差异""" import difflib content_a = next(v["content"] for v in self.versions if v["version_id"] == version_a) content_b = next(v["content"] for v in self.versions if v["version_id"] == version_b) diff = difflib.unified_diff( content_a.splitlines(keepends=True), content_b.splitlines(keepends=True), fromfile=version_a, tofile=version_b ) return ''.join(diff) A/B测试框架 测试设计 from dataclasses import dataclass from enum import Enum import numpy as np from scipy import stats class MetricType(Enum): ACCURACY = "accuracy" LATENCY = "latency" COST = "cost" USER_SATISFACTION = "satisfaction" TASK_COMPLETION = "completion" @dataclass class ABTestConfig: """A/B测试配置""" name: str description: str # 变体 control_version: str # 基线版本（A） treatment_version: str # 实验版本（B） # 流量分配 traffic_split: float # treatment流量比例 (0-1) # 指标 primary_metric: MetricType # 主要指标 secondary_metrics: list[MetricType] # 统计参数 significance_level: float = 0.05 # α statistical_power: float = 0.8 # 1-β minimum_detectable_effect: float = 0.05 # MDE # 持续时间 min_samples_per_variant: int = 1000 max_duration_days: int = 14 class PromptABTest: """Prompt A/B测试执行器""" def __init__(self, config: ABTestConfig, prompt_registry): self.config = config self.registry = prompt_registry self.results: dict[str, list] = { "control": [], "treatment": [] } def assign_variant(self, user_id: str) -> str: """分配用户到变体""" # 使用用户ID哈希确保同一用户始终在同一组 hash_value = hash(user_id) % 100 / 100 if hash_value < self.config.traffic_split: return "treatment" else: return "control" def get_prompt(self, variant: str) -> str: """获取对应变体的Prompt""" version = (self.config.control_version if variant == "control" else self.config.treatment_version) return self.registry.get(version) def record_result(self, variant: str, result: dict): """记录实验结果""" self.results[variant].append(result) def analyze(self) -> dict: """分析实验结果""" control_data = [r[self.config.primary_metric.value] for r in self.results["control"]] treatment_data = [r[self.config.primary_metric.value] for r in self.results["treatment"]] # 描述性统计 control_mean = np.mean(control_data) treatment_mean = np.mean(treatment_data) # 统计检验 if self._is_continuous(self.config.primary_metric): # 连续指标：t检验 t_stat, p_value = stats.ttest_ind( treatment_data, control_data ) effect_size = (treatment_mean - control_mean) / np.std(control_data) else: # 二分类指标：卡方检验 control_success = sum(control_data) treatment_success = sum(treatment_data) chi2, p_value = stats.chi2_contingency([ [control_success, len(control_data) - control_success], [treatment_success, len(treatment_data) - treatment_success] ])[:2] effect_size = (treatment_mean - control_mean) / control_mean # 置信区间 ci = self._confidence_interval( treatment_data, control_data, self.config.significance_level ) # 结论 significant = p_value < self.config.significance_level winner = "treatment" if significant and treatment_mean > control_mean else \ "control" if significant else "inconclusive" return { "control_mean": control_mean, "treatment_mean": treatment_mean, "relative_improvement": (treatment_mean - control_mean) / control_mean, "p_value": p_value, "effect_size": effect_size, "confidence_interval": ci, "significant": significant, "winner": winner, "sample_sizes": { "control": len(control_data), "treatment": len(treatment_data) } } def _confidence_interval(self, treatment, control, alpha): """计算置信区间""" diff = np.mean(treatment) - np.mean(control) se = np.sqrt(np.var(treatment)/len(treatment) + np.var(control)/len(control)) z = stats.norm.ppf(1 - alpha/2) return (diff - z * se, diff + z * se) 样本量计算 class SampleSizeCalculator: """计算所需样本量""" @staticmethod def for_proportion(baseline_rate: float, mde: float, alpha: float = 0.05, power: float = 0.8) -> int: """ 比率指标（如准确率）的样本量计算 baseline_rate: 基线比率 mde: 最小可检测效应 alpha: 显著性水平 power: 统计功效 """ from scipy.stats import norm z_alpha = norm.ppf(1 - alpha/2) z_beta = norm.ppf(power) p1 = baseline_rate p2 = baseline_rate + mde p_avg = (p1 + p2) / 2 n = ((z_alpha * np.sqrt(2 * p_avg * (1 - p_avg)) + z_beta * np.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / (p1 - p2) ** 2 return int(np.ceil(n)) @staticmethod def for_continuous(baseline_std: float, mde: float, alpha: float = 0.05, power: float = 0.8) -> int: """ 连续指标（如延迟）的样本量计算 """ from scipy.stats import norm z_alpha = norm.ppf(1 - alpha/2) z_beta = norm.ppf(power) n = 2 * ((z_alpha + z_beta) * baseline_std / mde) ** 2 return int(np.ceil(n)) 多变量测试（Multivariate Testing） class MultivariatePromptTest: """ 多变量Prompt测试 同时测试多个Prompt组件的变化 """ def __init__(self): self.factors = {} # 因子及其变体 def add_factor(self, name: str, variants: list[str]): """添加测试因子""" self.factors[name] = variants def generate_combinations(self) -> list[dict]: """生成所有组合""" from itertools import product factor_names = list(self.factors.keys()) factor_values = list(self.factors.values()) combinations = [] for values in product(*factor_values): combinations.append(dict(zip(factor_names, values))) return combinations def design(self) -> dict: """实验设计""" combos = self.generate_combinations() return { "total_combinations": len(combos), "combinations": combos, "traffic_per_combo": 1.0 / len(combos), "min_samples_per_combo": self._calc_min_samples(), "estimated_duration_days": self._estimate_duration(), } 实际案例：客服Prompt优化 # 案例：优化电商客服Prompt # 基线Prompt (v1.0.0) control_prompt = """ 你是一个电商客服助手。请回答用户的问题。 """ # 实验Prompt (v1.1.0) - 添加了情绪感知和解决方案导向 treatment_prompt = """ 你是一个专业的电商客服助手。 回答原则： 1. 先理解用户的情绪和核心诉求 2. 提供具体的解决方案，而非泛泛而谈 3. 如果需要转人工，明确说明原因 4. 保持友好但专业的语调 回答结构： - 确认问题：简要复述用户的问题 - 解决方案：给出具体步骤 - 后续支持：提供额外帮助选项 """ # 配置A/B测试 config = ABTestConfig( name="客服Prompt优化-情绪感知版", description="测试添加情绪感知和结构化回答是否提升客户满意度", control_version="v1.0.0", treatment_version="v1.1.0", traffic_split=0.5, primary_metric=MetricType.USER_SATISFACTION, secondary_metrics=[MetricType.TASK_COMPLETION, MetricType.LATENCY], minimum_detectable_effect=0.03, # 3%提升 min_samples_per_variant=2000, ) # 运行测试 ab_test = PromptABTest(config, prompt_registry) # 分析结果 results = ab_test.analyze() """ 预期输出： { "control_mean": 0.78, # 基线满意度 78% "treatment_mean": 0.83, # 实验组满意度 83% "relative_improvement": 0.064, # 6.4%提升 "p_value": 0.002, # p < 0.05，显著 "effect_size": 0.12, # 小到中等效应 "confidence_interval": (0.02, 0.08), "significant": True, "winner": "treatment" } """ 渐进式发布 class ProgressiveRollout: """ 渐进式发布 获胜的变体逐步增加流量 """ ROLLOUT_STAGES = [ {"traffic": 0.05, "duration_hours": 24, "min_success_rate": 0.85}, {"traffic": 0.20, "duration_hours": 48, "min_success_rate": 0.87}, {"traffic": 0.50, "duration_hours": 72, "min_success_rate": 0.88}, {"traffic": 1.00, "duration_hours": None, "min_success_rate": 0.88}, ] async def rollout(self, prompt_version: str): """渐进式发布""" for stage in self.ROLLOUT_STAGES: # 设置流量 await self.traffic_manager.set_traffic( prompt_version, stage["traffic"] ) # 等待观察期 if stage["duration_hours"]: await asyncio.sleep(stage["duration_hours"] * 3600) # 检查指标 success_rate = await self.metrics.get_success_rate( prompt_version, window_hours=stage["duration_hours"] ) if success_rate < stage["min_success_rate"]: # 回滚 await self.traffic_manager.rollback(prompt_version) await self.alerting.notify( f"发布失败：成功率 {success_rate:.1%} < " f"阈值 {stage['min_success_rate']:.1%}" ) return False return True 结语 Prompt优化不应该是基于直觉的猜测游戏。版本控制提供了可追溯的历史，A/B测试提供了统计严谨的决策依据。2026年的Prompt工程实践应该是： ...

从"散装Prompt"到"Prompt工程体系" 2026年，大型企业平均拥有超过5000个生产环境Prompt。这些Prompt分散在不同团队、不同项目中，由不同开发者编写，使用不同模型，服务于不同场景。如果没有系统化的管理方案，Prompt的维护成本将急剧攀升。 典型问题： 同一业务的Prompt在10个项目中各自维护，修改需要同步10处 离职员工的Prompt无人理解，不敢修改 模型升级后30%的Prompt性能下降，但无人知晓 没有统一的Prompt质量标准，质量参差不齐 本文分享我们在过去两年中构建企业级Prompt管理系统的实践经验。 Prompt模板架构 模板结构设计 from dataclasses import dataclass, field from typing import Optional from enum import Enum class PromptCategory(Enum): SYSTEM = "system" # 系统级Prompt TASK = "task" # 任务级Prompt GUARDRAIL = "guardrail" # 安全护栏 UTILITY = "utility" # 工具函数 EVALUATION = "evaluation" # 评估用 class PromptStatus(Enum): DRAFT = "draft" REVIEW = "review" TESTING = "testing" STAGING = "staging" PRODUCTION = "production" DEPRECATED = "deprecated" @dataclass class PromptTemplate: """Prompt模板定义""" id: str # 唯一标识 name: str # 模板名称 category: PromptCategory # 分类 status: PromptStatus # 状态 # 模板内容 system_prompt: str # 系统提示 user_prompt_template: str # 用户提示模板（含变量） variables: list[dict] # 变量定义 # 元数据 description: str # 描述 author: str # 作者 version: str # 版本号 tags: list[str] # 标签 # 配置 model_config: dict # 模型配置 expected_output: Optional[dict] # 期望输出格式 # 质量指标 quality_score: Optional[float] # 质量评分 latency_p95: Optional[float] # P95延迟 success_rate: Optional[float] # 成功率 # 关联 dependencies: list[str] = field(default_factory=list) # 依赖的其他模板 parent_id: Optional[str] = None # 父模板（继承关系） 模板语法 class PromptTemplateEngine: """ Prompt模板引擎 支持变量插值、条件逻辑、循环和继承 """ # 模板语法示例 TEMPLATE_EXAMPLE = """ {{#system}} 你是{{role}}，专注于{{domain}}领域。 核心规则： {{#rules}} - {{.}} {{/rules}} {{#if strict_mode}} ⚠️ 严格遵守以上规则，不允许偏离。 {{/if}} {{/system}} {{#user}} {{user_input}} {{#if context}} 相关上下文： {{#context}} --- {{.}} --- {{/context}} {{/if}} {{#if examples}} 参考示例： {{#examples}} 输入：{{input}} 输出：{{output}} {{/examples}} {{/if}} {{/user}} """ def __init__(self): self.templates: dict[str, PromptTemplate] = {} self.cache = {} def render(self, template_id: str, variables: dict) -> dict: """渲染模板""" template = self.templates.get(template_id) if not template: raise ValueError(f"模板 {template_id} 不存在") # 合并默认变量 merged_vars = self._merge_defaults(template, variables) # 验证必填变量 self._validate_variables(template, merged_vars) # 渲染 system = self._render_text(template.system_prompt, merged_vars) user = self._render_text(template.user_prompt_template, merged_vars) return { "system": system, "user": user, "model_config": template.model_config, "template_id": template_id, "version": template.version } def _render_text(self, template_text: str, variables: dict) -> str: """渲染模板文本""" # 使用Jinja2或自定义模板引擎 from jinja2 import Template tpl = Template(template_text) return tpl.render(**variables) Prompt注册中心 集中化存储 class PromptRegistry: """ Prompt注册中心 所有Prompt模板的单一可信来源（Single Source of Truth） """ def __init__(self, storage_backend="postgresql"): self.storage = self._init_storage(storage_backend) async def register(self, template: PromptTemplate) -> str: """注册新模板""" # 验证 self._validate_template(template) # 检查命名冲突 if await self._exists(template.name, template.version): raise ValueError(f"模板 {template.name} v{template.version} 已存在") # 存储 template_id = await self.storage.save(template) # 建立索引 await self._update_index(template) return template_id async def get(self, template_id: str) -> PromptTemplate: """获取模板""" return await self.storage.get(template_id) async def search(self, query: dict) -> list[PromptTemplate]: """搜索模板""" # 支持按名称、标签、分类、状态搜索 return await self.storage.search(query) async def update(self, template_id: str, updates: dict) -> PromptTemplate: """更新模板（创建新版本）""" current = await self.get(template_id) # 创建新版本 new_version = self._increment_version(current.version) updated = PromptTemplate( **{**current.__dict__, **updates, "version": new_version, "parent_id": template_id} ) # 注册新版本 new_id = await self.register(updated) # 标记旧版本 await self.storage.update( template_id, {"status": PromptStatus.DEPRECATED} ) return updated 权限管理 class PromptAccessControl: """ Prompt权限管理 """ PERMISSIONS = { "read": "查看模板", "write": "创建/修改模板", "deploy": "部署到生产", "delete": "删除模板", "export": "导出模板", } ROLES = { "viewer": ["read"], "developer": ["read", "write"], "reviewer": ["read", "write"], "admin": ["read", "write", "deploy", "delete", "export"], } def check_permission(self, user_id: str, template_id: str, permission: str) -> bool: """检查用户权限""" user_role = self._get_user_role(user_id) allowed = self.ROLES.get(user_role, []) if permission not in allowed: return False # 项目级权限检查 template = self.registry.get(template_id) if not self._has_project_access(user_id, template.project): return False return True Prompt流水线 CI/CD for Prompts class PromptPipeline: """ Prompt CI/CD 流水线 从开发到部署的完整流程 """ async def run_pipeline(self, template: PromptTemplate): """执行完整流水线""" results = {} # 阶段1: 静态检查 results["lint"] = await self._lint(template) if not results["lint"]["passed"]: return results # 阶段2: 单元测试 results["unit_test"] = await self._unit_test(template) if not results["unit_test"]["passed"]: return results # 阶段3: 安全检查 results["security"] = await self._security_scan(template) if not results["security"]["passed"]: return results # 阶段4: 性能测试 results["performance"] = await self._performance_test(template) # 阶段5: A/B测试准备 results["ab_setup"] = await self._setup_ab_test(template) # 阶段6: 部署 if all(r.get("passed", True) for r in results.values()): results["deploy"] = await self._deploy(template) return results async def _lint(self, template: PromptTemplate) -> dict: """静态检查""" issues = [] # 检查变量完整性 used_vars = self._extract_variables(template.user_prompt_template) defined_vars = [v["name"] for v in template.variables] for var in used_vars: if var not in defined_vars: issues.append(f"未定义的变量: {var}") # 检查长度 if len(template.system_prompt) > 2000: issues.append("System Prompt过长（>2000字符），可能影响性能") # 检查安全 dangerous_patterns = ["ignore previous", "you are now", "system prompt"] for pattern in dangerous_patterns: if pattern in template.user_prompt_template.lower(): issues.append(f"潜在安全风险: 包含 '{pattern}'") return { "passed": len(issues) == 0, "issues": issues } async def _unit_test(self, template: PromptTemplate) -> dict: """单元测试""" test_cases = template.variables.get("test_cases", []) results = [] for case in test_cases: rendered = self.engine.render(template.id, case["input"]) response = await self.llm.generate(rendered) passed = self._evaluate_response( response, case["expected"] ) results.append({ "case_name": case.get("name", "unnamed"), "passed": passed, "response": response[:200] }) pass_rate = sum(r["passed"] for r in results) / len(results) return { "passed": pass_rate >= 0.9, "pass_rate": pass_rate, "results": results } async def _performance_test(self, template: PromptTemplate) -> dict: """性能测试""" import time latencies = [] for _ in range(50): start = time.time() rendered = self.engine.render(template.id, {}) response = await self.llm.generate(rendered) latencies.append(time.time() - start) import numpy as np return { "passed": np.percentile(latencies, 95) < 5.0, # P95 < 5秒 "p50": np.median(latencies), "p95": np.percentile(latencies, 95), "p99": np.percentile(latencies, 99), } Prompt监控系统 实时监控 class PromptMonitor: """ Prompt生产环境监控 """ def __init__(self): self.metrics_store = MetricsStore() self.alerting = AlertingSystem() async def record_invocation(self, template_id: str, version: str, invocation_data: dict): """记录每次Prompt调用""" await self.metrics_store.record({ "template_id": template_id, "version": version, "timestamp": datetime.now(), "input": invocation_data["input"], "output": invocation_data["output"], "latency_ms": invocation_data["latency_ms"], "tokens_used": invocation_data["tokens_used"], "cost": invocation_data["cost"], "success": invocation_data["success"], "user_feedback": invocation_data.get("user_feedback"), }) # 实时检查 await self._check_anomalies(template_id, invocation_data) async def _check_anomalies(self, template_id: str, data: dict): """异常检测""" # 延迟异常 baseline_latency = await self.metrics_store.get_baseline_latency(template_id) if data["latency_ms"] > baseline_latency * 3: await self.alerting.send_alert( level="warning", template_id=template_id, message=f"延迟异常: {data['latency_ms']}ms (基线: {baseline_latency}ms)" ) # 成功率下降 recent_success_rate = await self.metrics_store.get_recent_success_rate( template_id, window_minutes=30 ) if recent_success_rate < 0.85: await self.alerting.send_alert( level="critical", template_id=template_id, message=f"成功率下降: {recent_success_rate:.1%}" ) # 成本异常 daily_cost = await self.metrics_store.get_daily_cost(template_id) if daily_cost > 100: # 日成本超过100元 await self.alerting.send_alert( level="warning", template_id=template_id, message=f"日成本异常: ¥{daily_cost}" ) 仪表盘 class PromptDashboard: """Prompt管理仪表盘数据生成""" def generate_report(self, date_range: tuple) -> dict: return { "overview": { "total_templates": self._count_templates(), "active_templates": self._count_active_templates(), "total_invocations": self._count_invocations(date_range), "total_cost": self._sum_cost(date_range), "avg_success_rate": self._avg_success_rate(date_range), "avg_latency_p95": self._avg_latency(date_range), }, "top_templates": self._top_templates(date_range, n=10), "quality_issues": self._identify_quality_issues(date_range), "cost_breakdown": self._cost_breakdown(date_range), "performance_trends": self._performance_trends(date_range), "recommendations": self._generate_recommendations(date_range), } 模板继承与组合 class PromptInheritance: """ Prompt模板继承系统 支持模板之间的继承和组合 """ def resolve(self, template_id: str) -> PromptTemplate: """ 解析模板继承链，生成最终Prompt """ template = self.registry.get(template_id) if template.parent_id: # 递归解析父模板 parent = self.resolve(template.parent_id) # 合并：子模板覆盖父模板 return self._merge(parent, template) return template def _merge(self, parent: PromptTemplate, child: PromptTemplate) -> PromptTemplate: """合并父子模板""" return PromptTemplate( id=child.id, name=child.name, system_prompt=child.system_prompt or parent.system_prompt, user_prompt_template=child.user_prompt_template or parent.user_prompt_template, variables=self._merge_variables(parent.variables, child.variables), # ... 其他字段 ) 最佳实践总结 Prompt模板管理清单 维度 实践 优先级 存储 集中化注册中心 P0 版本 语义化版本控制 P0 权限 基于角色的访问控制 P1 测试 自动化单元测试 P0 安全 注入扫描+内容审查 P0 监控 延迟/成功率/成本 P0 文档 每个模板附带说明 P1 复用 模板继承与组合 P1 优化 A/B测试框架 P2 治理 定期审查与清理 P1 结语 Prompt模板管理是AI工程化的基础设施。2026年的经验表明：将Prompt视为代码（Prompt as Code）是正确的方向。 版本控制、CI/CD、测试、监控——这些软件工程的成熟实践同样适用于Prompt管理。 ...

Few-shot Prompting的2026年新认知 Few-shot Prompting——通过在Prompt中提供少量示例来引导模型行为——是最古老也最有效的Prompt工程技巧之一。2026年，随着上下文窗口从8K扩展到1M+ tokens，Few-shot的可能性大幅扩展，但"更多不等于更好"。 2026年核心发现： 示例质量比数量更重要（5个精选示例 > 50个随机示例） 示例顺序对结果影响可达15-20% 示例与查询的语义相似度是选择的关键指标 负面示例（错误案例+修正）比单纯正面示例更有效 示例选择算法 1. 随机选择（基线） import random def random_selection(examples: list[dict], k: int = 4) -> list[dict]: """随机选择k个示例""" return random.sample(examples, k) 2. KNN-based选择（2026年主流） import numpy as np from sklearn.neighbors import NearestNeighbors class KNNExampleSelector: """ 基于KNN的示例选择 选择与当前查询语义最相似的示例 """ def __init__(self, examples: list[dict], embed_model): self.examples = examples self.embed_model = embed_model # 预计算示例的embedding self.example_embeddings = np.array([ embed_model.encode(ex["input"]) for ex in examples ]) # 构建KNN索引 self.knn = NearestNeighbors( n_neighbors=min(20, len(examples)), metric='cosine' ) self.knn.fit(self.example_embeddings) def select(self, query: str, k: int = 4) -> list[dict]: """选择与query最相似的k个示例""" query_embed = self.embed_model.encode(query).reshape(1, -1) # KNN搜索 distances, indices = self.knn.kneighbors(query_embed) # 取top-k selected = [self.examples[i] for i in indices[0][:k]] return selected 3. 多样性感知选择 class DiversityAwareSelector: """ 多样性感知的示例选择 平衡相似性和多样性 """ def __init__(self, examples, embed_model): self.examples = examples self.embed_model = embed_model self.embeddings = np.array([ embed_model.encode(ex["input"]) for ex in examples ]) def select(self, query: str, k: int = 4, alpha: float = 0.5) -> list[dict]: """ alpha: 相似性权重 (0-1) 1-alpha: 多样性权重 """ query_embed = self.embed_model.encode(query) # 计算与查询的相似度 similarities = cosine_similarity( query_embed.reshape(1, -1), self.embeddings )[0] selected = [] selected_indices = [] for _ in range(k): scores = [] for i in range(len(self.examples)): if i in selected_indices: scores.append(-float('inf')) continue # 相似性分数 sim_score = similarities[i] # 多样性分数（与已选示例的最大距离） if selected_indices: max_sim_to_selected = max( cosine_similarity( self.embeddings[i].reshape(1, -1), self.embeddings[j].reshape(1, -1) )[0][0] for j in selected_indices ) div_score = 1 - max_sim_to_selected else: div_score = 1.0 # 综合分数 combined = alpha * sim_score + (1 - alpha) * div_score scores.append(combined) best_idx = np.argmax(scores) selected.append(self.examples[best_idx]) selected_indices.append(best_idx) return selected 4. 基于强化学习的选择 class RLExampleSelector: """ 基于强化学习的示例选择 通过历史反馈学习最优选择策略 """ def __init__(self, examples, embed_model): self.examples = examples self.embed_model = embed_model self.q_table = {} # state -> action values self.learning_rate = 0.1 self.epsilon = 0.1 def select(self, query: str, k: int = 4) -> list[dict]: query_embed = self.embed_model.encode(query) state = self._discretize_state(query_embed) if random.random() < self.epsilon: # 探索：随机选择 return random.sample(self.examples, k) else: # 利用：选择Q值最高的示例 selected = [] remaining = list(range(len(self.examples))) for _ in range(k): # 选择Q值最高的 q_values = [ self.q_table.get((state, i), 0.0) for i in remaining ] best = remaining[np.argmax(q_values)] selected.append(self.examples[best]) remaining.remove(best) return selected def update(self, query: str, selected_indices: list[int], reward: float): """根据反馈更新Q值""" query_embed = self.embed_model.encode(query) state = self._discretize_state(query_embed) for idx in selected_indices: key = (state, idx) old_q = self.q_table.get(key, 0.0) self.q_table[key] = old_q + self.learning_rate * ( reward - old_q ) 示例排列优化 排列效应分析 class ExampleOrderOptimizer: """ 示例排列优化器 研究：相同示例不同排列，准确率差异可达15-20% """ def __init__(self, model): self.model = model def evaluate_ordering(self, examples: list[dict], eval_set: list[dict]) -> float: """评估特定排列的准确率""" correct = 0 for eval_item in eval_set: prompt = self._build_prompt(examples, eval_item["input"]) response = self.model.generate(prompt) if self._check_answer(response, eval_item["output"]): correct += 1 return correct / len(eval_set) def find_optimal_order(self, examples: list[dict], eval_set: list[dict]) -> list[dict]: """寻找最优排列（贪心搜索）""" from itertools import permutations best_acc = 0 best_order = examples # 对于少量示例，可以穷举 if len(examples) <= 5: for perm in permutations(examples): acc = self.evaluate_ordering(list(perm), eval_set) if acc > best_acc: best_acc = acc best_order = list(perm) else: # 贪心搜索 best_order = self._greedy_search(examples, eval_set) return best_order 2026年排列最佳实践 ORDERING_GUIDELINES = """ === Few-shot 示例排列最佳实践 === 1. 最近效应（Recency Effect） - 模型更容易受最后一个示例的影响 - 将最相关的示例放在最后 2. 难度递进 - 从简单到复杂排列 - 帮助模型逐步理解任务 3. 正负交替 - 正例-反例-正例-反例 - 比连续正例更有效 4. 避免偏见 - 不要将所有同一类别的示例放在一起 - 打乱类别顺序减少偏见 5. 答案分布平衡 - 如果是分类任务，确保各类别示例数量均衡 - 避免模型偏向多数类 """ 负面示例技术 class NegativeExamplePrompting: """ 负面示例Prompting 展示错误案例及其修正，比纯正面示例更有效 """ @staticmethod def build_prompt(positive_examples: list[dict], negative_examples: list[dict], query: str) -> str: """ 构建包含正负示例的Prompt """ prompt = "请根据以下示例完成任务。\n\n" # 正面示例 prompt += "✅ 正确示例：\n" for ex in positive_examples: prompt += f"输入：{ex['input']}\n" prompt += f"输出：{ex['output']}\n" prompt += f"说明：{ex.get('explanation', '')}\n\n" # 负面示例 if negative_examples: prompt += "❌ 错误示例（请避免以下错误）：\n" for ex in negative_examples: prompt += f"输入：{ex['input']}\n" prompt += f"❌ 错误输出：{ex['wrong_output']}\n" prompt += f"✅ 正确输出：{ex['correct_output']}\n" prompt += f"错误原因：{ex['error_reason']}\n\n" # 查询 prompt += f"现在请处理：\n输入：{query}\n输出：" return prompt 效果对比 方法 准确率 错误减少 适用场景 仅正面示例 82% - 简单任务 仅负面示例 75% - 错误模式明确 正面+负面 89% -39% 复杂任务 正面+负面+解释 93% -61% 高精度需求 跨语言Few-shot class CrossLingualFewShot: """ 跨语言Few-shot Prompting 用英语示例指导中文任务（或反向） """ def __init__(self, model): self.model = model def cross_lingual_prompt(self, source_examples: list[dict], target_query: str, source_lang: str = "en", target_lang: str = "zh") -> str: """ 构建跨语言Few-shot Prompt """ prompt = f"""以下是用{source_lang}语言展示的任务示例。 请理解示例中的任务模式，并用{target_lang}语言完成下面的查询。 示例： """ for ex in source_examples: prompt += f"Input: {ex['input']}\n" prompt += f"Output: {ex['output']}\n\n" prompt += f"现在请用{target_lang}回答：\n" prompt += f"Input: {target_query}\n" prompt += f"Output: " return prompt def translated_examples_prompt(self, examples: list[dict], query: str, target_lang: str = "zh") -> str: """ 翻译示例到目标语言后再使用 """ translated = [] for ex in examples: translated_input = self.model.translate( ex['input'], target_lang=target_lang ) translated_output = self.model.translate( ex['output'], target_lang=target_lang ) translated.append({ 'input': translated_input, 'output': translated_output }) return self._build_standard_prompt(translated, query) 动态Few-shot class DynamicFewShotSystem: """ 动态Few-shot系统 每次查询动态选择最相关的示例 """ def __init__(self, example_pool: list[dict], embed_model, llm): self.example_pool = example_pool self.embed_model = embed_model self.llm = llm self.selector = DiversityAwareSelector(example_pool, embed_model) self.feedback_store = [] async def answer(self, query: str, k: int = 4) -> dict: """动态选择示例并回答""" # 1. 选择示例 examples = self.selector.select(query, k=k) # 2. 构建Prompt prompt = self._build_prompt(examples, query) # 3. 生成回答 response = await self.llm.generate(prompt) # 4. 记录用于后续优化 self.feedback_store.append({ "query": query, "selected_examples": examples, "response": response, "timestamp": datetime.now() }) return { "answer": response, "examples_used": examples, "prompt": prompt } def optimize_pool(self): """基于历史反馈优化示例池""" # 分析哪些示例被高频选中且效果好 example_stats = {} for record in self.feedback_store: for ex in record["selected_examples"]: ex_id = ex["id"] if ex_id not in example_stats: example_stats[ex_id] = { "count": 0, "success": 0 } example_stats[ex_id]["count"] += 1 # 保留高频且高效的示例，淘汰低效的 # ... 评估与调试 class FewShotEvaluator: """Few-shot Prompting评估工具""" def __init__(self, model, eval_dataset): self.model = model self.eval_set = eval_dataset async def evaluate_configuration(self, selector_class, k: int, ordering: str = "similarity_desc", use_negative: bool = False) -> dict: """评估特定Few-shot配置""" selector = selector_class(self.eval_set, self.model) results = [] for item in self.eval_set: # 选择示例 examples = selector.select(item["input"], k=k) # 排列 if ordering == "similarity_desc": examples = sorted(examples, key=lambda x: x["similarity"]) elif ordering == "difficulty_asc": examples = sorted(examples, key=lambda x: x["difficulty"]) # 构建Prompt prompt = self._build_prompt(examples, item["input"]) # 生成 response = await self.model.generate(prompt) # 评估 correct = self._check(response, item["output"]) results.append({ "correct": correct, "response": response, "expected": item["output"] }) accuracy = sum(r["correct"] for r in results) / len(results) return { "accuracy": accuracy, "k": k, "selector": selector_class.__name__, "ordering": ordering, "use_negative": use_negative, "detailed_results": results } 2026年黄金法则 FEW_SHOT_GOLDEN_RULES_2026 = """ === Few-shot Prompting 黄金法则 === 1. 质量 > 数量 - 3-5个精选示例优于20个随机示例 - 每个示例都应展示不同的模式 2. 相似性选择 - 使用KNN或语义搜索选择与查询最相关的示例 - 但保持一定多样性 3. 排列有讲究 - 最相关的示例放最后（近因效应） - 简单到复杂排列帮助理解 4. 包含负面示例 - 展示"不该怎么做"比只展示"该怎么做"更有效 - 附带错误原因说明 5. 答案平衡 - 分类任务中各类别示例数量均等 - 避免模型产生频率偏见 6. 动态选择 - 不同查询用不同示例 - 建立示例池，按需选择 7. 持续优化 - 记录每次查询的示例选择和效果 - 定期评估和更新示例池 """ 结语 Few-shot Prompting看似简单——“给几个例子嘛”——但做到极致需要深入理解模型行为和任务特性。2026年的核心认知是：Few-shot不是静态的模板填充，而是一个动态的、数据驱动的系统。 ...

为什么结构化输出如此重要 在生产环境中，LLM的输出需要被程序解析和处理。非结构化的自然语言输出虽然灵活，但带来三个严重问题： 解析不可靠：正则提取容易遗漏边界情况 集成困难：下游系统需要稳定的接口契约 验证缺失：无法保证输出满足业务约束 2026年，结构化输出已从"nice to have"变为"must have"。所有主流模型都提供了原生结构化输出能力。 技术方案全景 方案对比 方案 原理 可靠性 性能 灵活性 适用场景 JSON Mode 模型内置JSON生成 95% 高 中 简单结构 Function Calling 函数签名约束 97% 高 高 API调用 Constrained Decoding 解码时约束 99% 中 最高 严格格式 Pydantic + LLM Schema验证+重试 90% 低 高 复杂校验 XML标签 标签结构化 85% 高 低 简单提取 JSON Mode 基本使用 import json from openai import OpenAI client = OpenAI() # OpenAI JSON Mode response = client.chat.completions.create( model="gpt-4o-2026", response_format={"type": "json_object"}, messages=[ { "role": "system", "content": """你是一个信息提取助手。 请将用户输入提取为JSON格式，包含以下字段： - name: 姓名 - age: 年龄（整数） - skills: 技能列表（字符串数组） - experience: 工作经验（整数，单位年） """ }, { "role": "user", "content": "张三，28岁，精通Python和JavaScript，有5年开发经验" } ] ) result = json.loads(response.choices[0].message.content) print(result) # 输出: {"name": "张三", "age": 28, "skills": ["Python", "JavaScript"], "experience": 5} JSON Schema约束 # 2026年最新：JSON Schema强化约束 response = client.chat.completions.create( model="gpt-4o-2026", response_format={ "type": "json_schema", "json_schema": { "name": "employee_info", "strict": True, "schema": { "type": "object", "properties": { "name": {"type": "string", "minLength": 1, "maxLength": 50}, "age": {"type": "integer", "minimum": 18, "maximum": 65}, "skills": { "type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 20 }, "experience": {"type": "integer", "minimum": 0, "maximum": 50}, "level": { "type": "string", "enum": ["junior", "mid", "senior", "expert"] } }, "required": ["name", "age", "skills", "experience", "level"], "additionalProperties": False } } }, messages=[...] ) 各家模型JSON Mode对比 模型 JSON可靠性 Schema支持 性能影响 特殊限制 GPT-4o 95% 完整 <5% 需提示JSON关键词 Claude 4 93% XML标签 <3% 推荐XML格式 Gemini 2 94% 部分支持 <5% Qwen 3 92% 部分 <5% Llama 4 88% 不支持 <8% 需Few-shot Function Calling 基本架构 from dataclasses import dataclass from typing import Callable import inspect @dataclass class ToolDefinition: name: str description: str parameters: dict # JSON Schema class FunctionCallingSystem: """ 2026年Function Calling最佳实践 """ def __init__(self, model_client): self.model = model_client self.tools: dict[str, ToolDefinition] = {} self.handlers: dict[str, Callable] = {} def register_function(self, func: Callable, description: str): """注册可调用函数""" # 自动从函数签名生成Schema sig = inspect.signature(func) params = {} required = [] for name, param in sig.parameters.items(): param_type = param.annotation json_type = self._python_type_to_json(param_type) params[name] = { "type": json_type, "description": self._extract_param_doc(func, name) } if param.default == inspect.Parameter.empty: required.append(name) tool = ToolDefinition( name=func.__name__, description=description or func.__doc__, parameters={ "type": "object", "properties": params, "required": required } ) self.tools[func.__name__] = tool self.handlers[func.__name__] = func async def execute_with_functions(self, user_message: str) -> str: """带函数调用的对话""" messages = [{"role": "user", "content": user_message}] tools = [t.__dict__ for t in self.tools.values()] while True: response = await self.model.chat( messages=messages, tools=tools, tool_choice="auto" ) message = response.choices[0].message messages.append(message) if not message.tool_calls: # 模型没有调用工具，返回最终回复 return message.content # 执行函数调用 for tool_call in message.tool_calls: func_name = tool_call.function.name func_args = json.loads(tool_call.function.arguments) # 参数验证 validation = self._validate_arguments(func_name, func_args) if not validation["valid"]: result = f"参数错误: {validation['errors']}" else: # 执行函数 try: handler = self.handlers[func_name] result = await handler(**func_args) except Exception as e: result = f"执行错误: {str(e)}" # 将结果返回给模型 messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) 实际应用示例 # 定义工具函数 @dataclass class SearchResult: title: str url: str snippet: str async def search_web(query: str, max_results: int = 5) -> list[dict]: """搜索网络内容 Args: query: 搜索关键词 max_results: 最大返回结果数（默认5） """ # 实际搜索逻辑 results = await search_engine.search(query, limit=max_results) return [{"title": r.title, "url": r.url, "snippet": r.snippet} for r in results] async def get_weather(city: str, unit: str = "celsius") -> dict: """获取指定城市的天气信息 Args: city: 城市名称 unit: 温度单位（celsius或fahrenheit） """ weather = await weather_api.get(city, unit) return { "city": city, "temperature": weather.temp, "condition": weather.condition, "humidity": weather.humidity } async def send_email(to: str, subject: str, body: str) -> dict: """发送邮件 Args: to: 收件人邮箱 subject: 邮件主题 body: 邮件正文 """ await email_service.send(to, subject, body) return {"status": "sent", "to": to} # 注册并使用 system = FunctionCallingSystem(model_client) system.register_function(search_web, "搜索网络获取最新信息") system.register_function(get_weather, "查询天气信息") system.register_function(send_email, "发送邮件") # 执行 response = await system.execute_with_functions( "帮我查一下北京今天的天气，然后把结果发邮件给 zhangsan@example.com" ) Constrained Decoding 原理 Constrained Decoding（约束解码）在生成过程中实时约束token选择，确保输出符合预定义的语法规则。 ...

Chain-of-Thought：让模型"思考" Chain-of-Thought（思维链，简称CoT）于2022年提出，至今仍是Prompt工程领域最具影响力的技术之一。核心思想是：让模型显式输出推理过程，而非直接给出答案。 2026年，CoT已经从单一技术演化为一个完整的技术家族，包括CoT-SC、ToT、GoT、PoT等多种变体。本文将系统梳理这些技术，并提供实战代码。 CoT基础：显式推理 为什么CoT有效？ # 对比：标准Prompt vs CoT Prompt # 标准Prompt standard_prompt = """ 问：小明有5个苹果，小红给了他3个， 然后小明吃掉了2个。小明现在有多少苹果？ 答： """ # CoT Prompt cot_prompt = """ 问：小明有5个苹果，小红给了他3个， 然后小明吃掉了2个。小明现在有多少苹果？ 让我们逐步推理： 1. 小明开始有5个苹果 2. 小红给了他3个，所以：5 + 3 = 8个 3. 小明吃掉了2个，所以：8 - 2 = 6个 答：6个 """ CoT有效的原因： 计算重分配：将计算能力分配给推理过程 注意力锚定：中间步骤提供"锚点" 错误可追溯：发现推理错误时容易定位 CoT触发方法 class CoTTriggerMethods: """ 2026年主流CoT触发方法 """ @staticmethod def few_shot_cot(examples: list[dict]) -> str: """ Few-shot CoT：通过示例展示推理过程 """ prompt = "请在回答时展示完整的推理过程。\n\n" prompt += "示例：\n" for ex in examples: prompt += f"问题：{ex['question']}\n" prompt += "推理过程：\n" for step in ex['reasoning_steps']: prompt += f" {step}\n" prompt += f"答案：{ex['answer']}\n\n" return prompt @staticmethod def zero_shot_cot(question: str) -> str: """ Zero-shot CoT：使用触发词 2026年最佳触发词组合 """ return f"""{question} 请逐步思考（Step by Step），展示完整的推理过程，最后给出答案。""" @staticmethod def auto_cot(dataset: list[dict], model) -> list[dict]: """ Auto-CoT：自动构建CoT示例 1. 使用聚类选择多样性问题 2. 使用模型生成推理过程 3. 验证生成的正确性 """ # 步骤1：问题聚类 embeddings = model.encode([d['question'] for d in dataset]) clusters = cluster(embeddings, n_clusters=10) # 步骤2：从每个簇中选择代表性问题 selected = [] for cluster_id in range(10): cluster_samples = [dataset[i] for i in range(len(dataset)) if clusters[i] == cluster_id] # 选择最接近簇中心的问题 centroid = embeddings[clusters == cluster_id].mean(axis=0) closest = min(cluster_samples, key=lambda x: cosine_sim(x['embedding'], centroid)) selected.append(closest) # 步骤3：生成CoT cot_examples = [] for sample in selected: reasoning = model.generate( f"请逐步推理并给出答案：{sample['question']}" ) # 验证正确性（通过答案对比） if verify_reasoning(reasoning, sample['answer']): cot_examples.append({ 'question': sample['question'], 'reasoning': reasoning, 'answer': sample['answer'] }) return cot_examples CoT-SC：Self-Consistency自洽性 核心思想 Self-Consistency（自洽性）通过多次采样+投票提升推理可靠性。 ...

Prompt工程在2026年已经从"玄学技巧"演变为一门系统化的工程学科。当LLM能力越来越强，Prompt的焦点从"让模型能做"转向了"让模型做得好、做得稳、做得可控"。本文将从基础到企业级，全面梳理2026年Prompt工程的最新实践。 一、Prompt工程的2026年现状 范式转变 时期 核心挑战 Prompt焦点 代表技术 2022-2023 模型能力有限 如何让模型"能做" Few-shot, CoT 2024-2025 能力提升但不可控 如何让模型"做好" 结构化Prompt, ReAct 2026 能力强但需要规模化 如何让模型"做稳" Prompt管理, A/B测试, 自动优化 2026年的核心认知 模型能力已不是瓶颈：GPT-5/Claude 5的基本能力足以应对大多数任务 Prompt质量决定输出质量：同样的模型，好Prompt和差Prompt的效果差距可达300% Prompt是资产：企业Prompt需要版本管理、测试、监控——和代码一样 自动化是趋势：自动Prompt优化（APO）开始替代人工调优 二、基础技巧回顾与升级 1. 角色设定（Role Prompting） 2026年的最佳实践不再是简单的"你是一个专家"，而是结构化角色定义： # 角色定义 你是一位资深的金融分析师，拥有CFA证书和15年A股市场研究经验。 ## 专业知识 - 精通财务报表分析和估值模型（DCF, DDM, PEG） - 熟悉A股市场的行业轮动和风格切换 - 擅长宏观经济分析和政策解读 ## 分析风格 - 数据驱动：每个结论必须有数据支撑 - 辩证思考：同时分析利多和利空因素 - 风险意识：始终提示潜在风险 ## 输出规范 - 使用专业但易懂的语言 - 关键数据标注来源 - 给出明确的投资建议（买入/持有/卖出）和理由 2. Few-Shot Learning的进化 2026年的Few-Shot不再只是"给几个例子"，而是动态示例选择： ...

上下文窗口的"房价"问题 2026 年，虽然主流模型的上下文窗口已达到 128K-1M tokens，但"窗口越大越不够用"——RAG 检索结果、工具调用返回、对话历史、知识库内容，每个环节都在争抢窗口空间。Prompt 压缩技术就像是在有限的土地上建造高层建筑，让每一个 token 都发挥最大价值。 一、Prompt 压缩的价值 1.1 成本与性能双优化 优化维度 压缩前 压缩后 改善 输入 Token 数 8000 4000 -50% API 成本（/千次） $24 $12 -50% 响应延迟 3.2s 1.8s -44% 上下文利用率 40% 75% +87.5% 信息保留率 100% 92-97% -3~8% 1.2 压缩策略分类 Prompt 压缩 ├── 无损压缩 │ ├── 符号化压缩（缩写、代码化） │ ├── 结构化压缩（JSON→紧凑格式） │ └── 去冗余压缩（删除重复信息） ├── 有损压缩 │ ├── 语义压缩（LLM 总结） │ ├── 选择性保留（截断低重要性内容） │ └── 信息蒸馏（提取关键信息） └── 混合压缩 └── 分层压缩策略 二、无损压缩技术 2.1 符号化压缩 class SymbolicCompressor: """符号化压缩——用短符号替代长文本""" SYMBOL_MAP = { # 常见指令缩写 "请分析以下内容并给出": "分析:", "请根据以上信息回答": "回答:", "以下是相关的背景信息": "背景:", "请注意以下重要事项": "注意:", # 角色缩写 "你是一个专业的": "角色:", "你的核心职责是": "职责:", # 格式缩写 "请用Markdown表格格式输出": "→MD表格", "请用JSON格式输出": "→JSON", "请用列表格式输出": "→列表", # 常见短语 "需要注意的是": "⚠", "重要提醒": "‼", "例如": "如", "也就是说": "即", } def compress(self, prompt: str) -> str: for full, symbol in self.SYMBOL_MAP.items(): prompt = prompt.replace(full, symbol) return prompt def decompress_guide(self) -> str: """生成符号说明（添加到System Prompt）""" guide = "符号说明: " for full, symbol in self.SYMBOL_MAP.items(): guide += f"{symbol}={full[:4]}.. " return guide 2.2 结构化压缩 class StructuralCompressor: """结构化压缩——压缩冗余的格式""" def compress_table(self, markdown_table: str) -> str: """压缩 Markdown 表格""" lines = markdown_table.strip().split('\n') if len(lines) < 3: return markdown_table # 提取表头和数据 headers = [h.strip() for h in lines[0].split('|')[1:-1]] data_rows = [] for line in lines[2:]: # 跳过分隔行 cells = [c.strip() for c in line.split('|')[1:-1]] data_rows.append(cells) # 紧凑格式：用 | 分隔，不用对齐 compact = '|'.join(headers) + '\n' for row in data_rows: compact += '|'.join(row) + '\n' return compact def compress_json(self, json_str: str) -> str: """压缩 JSON""" import json data = json.loads(json_str) return json.dumps(data, ensure_ascii=False, separators=(',', ':')) def compress_list(self, markdown_list: str) -> str: """压缩列表""" lines = markdown_list.strip().split('\n') items = [l.lstrip('- *').strip() for l in lines if l.strip()] return '; '.join(items) 2.3 去冗余压缩 class RedundancyRemover: """去冗余压缩""" def compress(self, prompt: str) -> str: # 1. 移除重复段落 prompt = self._remove_duplicate_paragraphs(prompt) # 2. 移除重复句子 prompt = self._remove_duplicate_sentences(prompt) # 3. 移除空白行 prompt = self._remove_blank_lines(prompt) # 4. 合并连续空格 import re prompt = re.sub(r' {2,}', ' ', prompt) return prompt def _remove_duplicate_paragraphs(self, text: str) -> str: paragraphs = text.split('\n\n') seen = set() unique = [] for p in paragraphs: normalized = p.strip().lower() if normalized and normalized not in seen: seen.add(normalized) unique.append(p) return '\n\n'.join(unique) def _remove_duplicate_sentences(self, text: str) -> str: import re sentences = re.split(r'(?<=[。.!?！？])\s+', text) seen = set() unique = [] for s in sentences: if s.strip() and s.strip() not in seen: seen.add(s.strip()) unique.append(s) return ' '.join(unique) 三、有损压缩技术 3.1 LLM 语义压缩 class SemanticCompressor: """使用 LLM 进行语义压缩""" COMPRESSION_PROMPT = """请压缩以下文本，要求： 1. 保留所有关键信息和数据 2. 保留逻辑结构和因果关系 3. 移除冗余描述和过渡语句 4. 用更简洁的表达替代冗长表达 5. 保持事实准确性 原始文本（{original_tokens} tokens）： {text} 输出压缩后的文本，目标：{target_tokens} tokens以内。""" def compress(self, text: str, target_ratio: float = 0.5, llm_client=None) -> str: original_tokens = self._estimate_tokens(text) target_tokens = int(original_tokens * target_ratio) prompt = self.COMPRESSION_PROMPT.format( original_tokens=original_tokens, text=text, target_tokens=target_tokens ) compressed = llm_client.generate(prompt) return compressed def _estimate_tokens(self, text: str) -> int: # 粗略估算 chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') other_chars = len(text) - chinese_chars return chinese_chars * 2 + other_chars // 4 3.2 选择性保留压缩 class SelectiveCompressor: """选择性保留——基于重要性的压缩""" def compress(self, text: str, target_ratio: float = 0.5) -> str: # 1. 分割为句子 sentences = self._split_sentences(text) # 2. 计算每个句子的重要性分数 scored = self._score_sentences(sentences) # 3. 保留高重要性句子 target_count = int(len(sentences) * target_ratio) top_sentences = sorted(scored, key=lambda x: -x[1])[:target_count] # 4. 按原顺序排列 top_sentences.sort(key=lambda x: x[2]) # 按原始位置排序 return ' '.join(s[0] for s in top_sentences) def _score_sentences(self, sentences: list) -> list: """使用 TextRank 思想计算句子重要性""" # 基于句子间相似度构建图 n = len(sentences) scores = [1.0] * n for iteration in range(10): # 迭代计算 new_scores = [] for i, sent in enumerate(sentences): score = 0.15 # 基础分 for j, other in enumerate(sentences): if i != j: sim = self._sentence_similarity(sent, other) score += 0.85 * sim * scores[j] / max( sum(self._sentence_similarity(other, s2) for k, s2 in enumerate(sentences) if k != j), 1e-8 ) new_scores.append(score) scores = new_scores return [(sentences[i], scores[i], i) for i in range(n)] def _sentence_similarity(self, s1: str, s2: str) -> float: """计算两个句子的相似度""" words1 = set(s1.split()) words2 = set(s2.split()) intersection = words1 & words2 union = words1 | words2 return len(intersection) / max(len(union), 1) 3.3 信息蒸馏 class InformationDistiller: """信息蒸馏——提取关键信息，丢弃细节""" DISTILL_PROMPT = """从以下文本中提取关键信息，使用紧凑格式输出。 输出格式： - 主题：[1-2句话] - 关键事实：[每条1行，最多5条] - 数据：[数值和单位] - 结论：[1句话] 文本： {text}""" def distill(self, text: str, llm_client) -> str: prompt = self.DISTILL_PROMPT.format(text=text) return llm_client.generate(prompt) 四、分层压缩策略 class LayeredCompressor: """分层压缩策略——不同内容用不同压缩方法""" def __init__(self, llm_client): self.llm = llm_client self.symbolic = SymbolicCompressor() self.structural = StructuralCompressor() self.redundancy = RedundancyRemover() self.semantic = SemanticCompressor() self.selective = SelectiveCompressor() self.distiller = InformationDistiller(llm_client) def compress(self, prompt: str, target_ratio: float = 0.5) -> str: """分层压缩""" original_tokens = self._estimate_tokens(prompt) target_tokens = int(original_tokens * target_ratio) # Layer 1: 无损压缩（总是执行） prompt = self.symbolic.compress(prompt) prompt = self.structural.compress_json(prompt) prompt = self.redundancy.compress(prompt) current_tokens = self._estimate_tokens(prompt) if current_tokens <= target_tokens: return prompt # 无损压缩已达标 # Layer 2: 内容分类 sections = self._classify_sections(prompt) # Layer 3: 按类别压缩 compressed_sections = [] for section_type, content in sections: if section_type == 'rules': # 规则类：仅做符号压缩 compressed_sections.append(self.symbolic.compress(content)) elif section_type == 'knowledge': # 知识类：语义压缩 compressed_sections.append( self.semantic.compress(content, 0.4, self.llm) ) elif section_type == 'examples': # 示例类：选择性保留 compressed_sections.append( self.selective.compress(content, 0.6) ) elif section_type == 'context': # 上下文类：信息蒸馏 compressed_sections.append( self.distiller.distill(content, self.llm) ) else: compressed_sections.append(content) result = '\n\n'.join(compressed_sections) # Layer 4: 如果仍超标，全局压缩 if self._estimate_tokens(result) > target_tokens: result = self.semantic.compress(result, target_tokens / self._estimate_tokens(result), self.llm) return result def _classify_sections(self, prompt: str) -> list: """将 Prompt 分为不同类型的段落""" sections = [] current_section = "" current_type = "other" for line in prompt.split('\n'): if line.startswith('规则') or line.startswith('约束'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'rules' elif line.startswith('知识') or line.startswith('背景'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'knowledge' elif line.startswith('示例') or line.startswith('例子'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'examples' elif line.startswith('上下文') or line.startswith('历史'): if current_section: sections.append((current_type, current_section)) current_section = line + '\n' current_type = 'context' else: current_section += line + '\n' if current_section: sections.append((current_type, current_section)) return sections 五、对话历史压缩 class ConversationHistoryCompressor: """对话历史压缩——长对话的上下文管理""" def __init__(self, llm_client, max_history_tokens: int = 4000): self.llm = llm_client self.max_tokens = max_history_tokens def compress_history(self, messages: list) -> list: """压缩对话历史""" total_tokens = sum(self._estimate_tokens(m['content']) for m in messages) if total_tokens <= self.max_tokens: return messages # 不需要压缩 # 策略：保留最近 N 轮 + 压缩早期对话 recent_count = min(6, len(messages)) # 保留最近3轮 recent = messages[-recent_count:] old = messages[:-recent_count] # 压缩早期对话 summary = self._summarize_conversation(old) # 构建压缩后的历史 compressed = [ {"role": "system", "content": f"对话摘要：{summary}"}, *recent ] return compressed def _summarize_conversation(self, messages: list) -> str: """总结早期对话""" conversation_text = '\n'.join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请总结以下对话的关键信息，保留： 1. 用户的核心需求和偏好 2. 已达成的结论和决定 3. 未解决的问题 4. 重要的事实和数据 对话内容： {conversation_text} 总结（不超过200字）：""" return self.llm.generate(prompt) 六、压缩效果评估 class CompressionEvaluator: """压缩效果评估器""" def evaluate(self, original: str, compressed: str, test_cases: list, llm_client) -> dict: results = { 'compression_ratio': len(compressed) / len(original), 'token_reduction': 1 - self._tokens(compressed) / self._tokens(original), } # 信息保留率评估 info_retention = self._evaluate_info_retention(original, compressed, llm_client) results['info_retention'] = info_retention # 任务效果评估 original_scores = [] compressed_scores = [] for case in test_cases: # 使用原始 Prompt original_response = llm_client.generate(original + case['input']) original_scores.append(self._score(original_response, case['expected'])) # 使用压缩 Prompt compressed_response = llm_client.generate(compressed + case['input']) compressed_scores.append(self._score(compressed_response, case['expected'])) results['original_accuracy'] = sum(original_scores) / len(original_scores) results['compressed_accuracy'] = sum(compressed_scores) / len(compressed_scores) results['accuracy_drop'] = results['original_accuracy'] - results['compressed_accuracy'] # 成本节省 results['cost_saving'] = results['token_reduction'] return results 压缩效果实测数据 压缩方法 压缩率 信息保留 准确率变化 延迟改善 符号压缩 15% 100% 0% +10% 去冗余 20% 100% 0% +15% 语义压缩 45% 94% -3% +40% 选择性保留 50% 90% -5% +45% 信息蒸馏 65% 85% -8% +55% 分层压缩 50% 96% -2% +44% 七、最佳实践 先无损后有损：先尝试无损压缩，不够再考虑有损 分层压缩最优：不同内容用不同策略，综合效果最好 规则不可压缩：System Prompt 中的规则和约束不应被有损压缩 压缩vs精简：很多时候重新设计比压缩更有效 监控压缩质量：定期评估压缩后的任务效果 缓存压缩结果：相同输入的压缩结果可以缓存 结语 Prompt 压缩是在信息密度和效果之间寻找最优平衡点的艺术。2026 年的工具链已经足够成熟，可以实现接近无损的 50% 压缩——这意味着同样的上下文窗口可以容纳两倍的信息，同样的预算可以处理两倍的请求。 ...

Few-Shot 学习的示例选择困境 Few-Shot Prompt（少样本提示）是大模型 In-Context Learning（上下文学习）的核心技术。但"选哪些示例"一直是效果差异的关键——同样的 Few-Shot 模板，选对示例效果可达 95%，选错可能只有 60%。2026 年，示例选择已经从"人工挑选"进化为算法化、自适应的选择系统。 一、示例选择为什么重要 1.1 示例质量对效果的影响 选择策略 准确率 说明 随机选择 62% 从示例池随机取 人工选择 78% 领域专家挑选 相似度检索 87% 基于语义相似度 多样性采样 84% 保证示例多样性 算法优化选择 93% 多维度综合优化 自适应选择 95% 根据输入动态选择 1.2 示例选择的三要素 ┌──────────────────────────────────┐ │ 示例选择三要素 │ ├──────────────────────────────────┤ │ 相关性：示例与当前输入的关联度 │ │ 多样性：示例集覆盖不同情况 │ │ 一致性：示例间的标注风格统一 │ └──────────────────────────────────┘ 二、示例选择算法 2.1 基于相似度的选择（kNN） import numpy as np from sklearn.metrics.pairwise import cosine_similarity class SimilarityBasedSelector: """基于语义相似度的示例选择""" def __init__(self, embedding_model="text-embedding-3-large"): self.embedding_model = embedding_model self.example_pool = [] self.example_embeddings = [] def add_examples(self, examples: list): """添加示例到池""" self.example_pool.extend(examples) embeddings = self._batch_embed([e['input'] for e in examples]) self.example_embeddings.extend(embeddings) def select(self, query: str, k: int = 3) -> list: """选择最相似的 k 个示例""" query_emb = self._embed(query) # 计算与所有示例的相似度 similarities = cosine_similarity( [query_emb], self.example_embeddings )[0] # 取 top-k top_indices = np.argsort(similarities)[-k:][::-1] return [self.example_pool[i] for i in top_indices] def _embed(self, text: str) -> np.ndarray: # 使用 embedding API pass def _batch_embed(self, texts: list) -> list: pass 2.2 基于多样性的选择 class DiversityBasedSelector: """基于多样性的示例选择——确保示例覆盖不同情况""" def __init__(self, embedding_model="text-embedding-3-large"): self.embedder = embedding_model def select(self, query: str, examples: list, k: int = 3) -> list: """选择覆盖面最广的 k 个示例""" # 1. 计算所有示例与 query 的相似度 query_emb = self._embed(query) example_embs = [self._embed(e['input']) for e in examples] similarities = [ cosine_similarity([query_emb], [emb])[0][0] for emb in example_embs ] # 2. 使用 MMR (Maximal Marginal Relevance) 算法 selected = [] selected_indices = [] # 第一个选最相似的 first = np.argmax(similarities) selected.append(examples[first]) selected_indices.append(first) # 后续选择：平衡相关性和多样性 while len(selected) < k: best_score = -float('inf') best_idx = -1 for i, example in enumerate(examples): if i in selected_indices: continue # 相关性分数 relevance = similarities[i] # 多样性分数（与已选示例的最大相似度的负值） diversity = min( cosine_similarity( [example_embs[i]], [example_embs[j]] )[0][0] for j in selected_indices ) # MMR 分数 mmr_score = 0.7 * relevance - 0.3 * (1 - diversity) if mmr_score > best_score: best_score = mmr_score best_idx = i selected.append(examples[best_idx]) selected_indices.append(best_idx) return selected 2.3 基于投票的选择 class VotingBasedSelector: """基于投票的示例选择——多策略集成""" def __init__(self): self.selectors = [ SimilarityBasedSelector(), DiversityBasedSelector(), ComplexityBasedSelector(), LabelBalancedSelector(), ] def select(self, query: str, k: int = 3) -> list: """多策略投票选择""" votes = {} for selector in self.selectors: selected = selector.select(query, k=k) for example in selected: ex_id = example['id'] votes[ex_id] = votes.get(ex_id, 0) + 1 # 按票数排序 ranked = sorted(votes.items(), key=lambda x: -x[1]) selected_ids = [ex_id for ex_id, _ in ranked[:k]] return [self.get_example(eid) for eid in selected_ids] 2.4 基于复杂度匹配的选择 class ComplexityBasedSelector: """基于复杂度匹配的示例选择""" def __init__(self, llm_client): self.llm = llm_client def assess_complexity(self, text: str) -> dict: """评估输入的复杂度""" prompt = f""" 评估以下输入的复杂度，返回JSON： {{ "reasoning_depth": 1-5, // 推理深度 "knowledge_required": 1-5, // 所需知识 "ambiguity": 1-5, // 歧义程度 "length": 1-5 // 输入长度 }} 输入：{text} """ return json.loads(self.llm.generate(prompt)) def select(self, query: str, examples: list, k: int = 3) -> list: query_complexity = self.assess_complexity(query) # 为每个示例评估复杂度 scored_examples = [] for example in examples: ex_complexity = self.assess_complexity(example['input']) # 计算复杂度匹配度（欧氏距离的倒数） distance = sum( (query_complexity[dim] - ex_complexity[dim]) ** 2 for dim in query_complexity ) ** 0.5 score = 1 / (1 + distance) scored_examples.append((example, score)) # 按匹配度排序 scored_examples.sort(key=lambda x: -x[1]) return [ex for ex, _ in scored_examples[:k]] 2.5 标签平衡选择 class LabelBalancedSelector: """标签平衡选择——确保示例标签分布合理""" def select(self, query: str, examples: list, k: int = 3) -> list: # 按 label 分组 label_groups = {} for ex in examples: label = ex.get('label', 'unknown') if label not in label_groups: label_groups[label] = [] label_groups[label].append(ex) # 计算每个组应选的数量 n_labels = len(label_groups) per_label = max(1, k // n_labels) # 从每个组中选最相似的 selected = [] for label, group in label_groups.items(): # 在组内按相似度排序 scored = [(ex, self._similarity(query, ex['input'])) for ex in group] scored.sort(key=lambda x: -x[1]) selected.extend([ex for ex, _ in scored[:per_label]]) # 如果不够 k 个，从剩余中补充 while len(selected) < k: remaining = [ex for ex in examples if ex not in selected] if not remaining: break selected.append(remaining[0]) return selected[:k] 三、示例顺序优化 3.1 顺序对效果的影响 class ExampleOrderOptimizer: """示例顺序优化""" def __init__(self, llm_client): self.llm = llm_client def optimize_order(self, examples: list, query: str) -> list: """优化示例排列顺序""" # 策略1：复杂度递增（简单→复杂） ordered = sorted(examples, key=lambda e: e.get('complexity', 3)) # 策略2：与query最相似的放最后（近因效应） similarities = [self._similarity(query, e['input']) for e in ordered] # 将最相似的移到最后 max_sim_idx = similarities.index(max(similarities)) ordered.append(ordered.pop(max_sim_idx)) return ordered 3.2 顺序效果对比 排列策略 准确率 说明 随机排列 75% 无序 简单→复杂 88% 渐进式 复杂→简单 79% 递减式 相似度递增 90% 最相似的在最后 一致性排列 85% 标签交替 四、示例格式优化 4.1 格式模板 EXAMPLE_FORMATS = { 'minimal': '{input}\n→ {output}', 'explained': '输入：{input}\n分析：{reasoning}\n输出：{output}', 'structured': """ <example> <input>{input}</input> <reasoning>{reasoning}</reasoning> <output>{output}</output> </example>""", 'conversational': '用户：{input}\n助手：{output}', 'annotated': '{input}\n[正确答案：{output}]\n[原因：{reasoning}]', } 4.2 格式选择指南 def select_format(task_type: str) -> str: """根据任务类型选择示例格式""" mapping = { 'classification': 'minimal', # 分类任务用最简格式 'generation': 'explained', # 生成任务需要推理过程 'extraction': 'structured', # 信息提取用结构化 'conversation': 'conversational', # 对话任务用对话格式 'reasoning': 'annotated', # 推理任务需要标注 } return mapping.get(task_type, 'explained') 五、自适应示例选择系统 class AdaptiveExampleSelector: """自适应示例选择系统——根据输入特征动态选择策略""" def __init__(self, llm_client, embedding_model): self.llm = llm_client self.similarity_selector = SimilarityBasedSelector(embedding_model) self.diversity_selector = DiversityBasedSelector(embedding_model) self.complexity_selector = ComplexityBasedSelector(llm_client) self.label_balanced_selector = LabelBalancedSelector() self.order_optimizer = ExampleOrderOptimizer(llm_client) def select(self, query: str, example_pool: list, k: int = 3, task_type: str = None) -> list: """自适应选择示例""" # 1. 分析输入特征 features = self._analyze_query(query) # 2. 选择策略 strategy = self._select_strategy(features, task_type) # 3. 执行选择 if strategy == 'similarity': examples = self.similarity_selector.select(query, example_pool, k) elif strategy == 'diversity': examples = self.diversity_selector.select(query, example_pool, k) elif strategy == 'complexity': examples = self.complexity_selector.select(query, example_pool, k) elif strategy == 'balanced': examples = self.label_balanced_selector.select(query, example_pool, k) elif strategy == 'hybrid': examples = self._hybrid_select(query, example_pool, k) # 4. 优化顺序 examples = self.order_optimizer.optimize_order(examples, query) return examples def _analyze_query(self, query: str) -> dict: """分析输入特征""" return { 'length': len(query.split()), 'complexity_signals': sum(1 for w in query.split() if len(w) > 8), 'has_numbers': any(c.isdigit() for c in query), 'language': self._detect_language(query), } def _select_strategy(self, features: dict, task_type: str = None) -> str: """根据特征选择策略""" if task_type == 'classification': return 'balanced' elif task_type == 'generation': return 'diversity' elif features['complexity_signals'] > 3: return 'complexity' elif features['length'] > 50: return 'similarity' else: return 'hybrid' def _hybrid_select(self, query: str, pool: list, k: int) -> list: """混合策略选择""" # 先用相似度选 2k 个候选 candidates = self.similarity_selector.select(query, pool, k * 2) # 再用多样性从候选中选 k 个 return self.diversity_selector.select(query, candidates, k) 六、评估与优化 6.1 示例选择评估 class ExampleSelectionEvaluator: """示例选择效果评估""" def evaluate(self, selector, test_set: list, baseline_selector=None) -> dict: results = { 'selector': selector.__class__.__name__, 'metrics': { 'accuracy': [], 'consistency': [], 'latency_ms': [], } } for case in test_set: import time start = time.time() examples = selector.select(case['input'], case['pool'], k=3) prompt = self._build_prompt(examples, case['input']) response = self.llm.generate(prompt) results['metrics']['accuracy'].append( self._score(response, case['expected']) ) results['metrics']['latency_ms'].append( (time.time() - start) * 1000 ) # 汇总 summary = { 'mean_accuracy': np.mean(results['metrics']['accuracy']), 'std_accuracy': np.std(results['metrics']['accuracy']), 'mean_latency': np.mean(results['metrics']['latency_ms']), } return summary 6.2 持续优化循环 class ExampleOptimizationLoop: """示例池持续优化""" def __init__(self, selector, llm_client): self.selector = selector self.llm = llm_client self.performance_log = [] def step(self, query: str, response: str, expected: str, examples_used: list): """记录每次使用并优化""" correct = self._is_correct(response, expected) self.performance_log.append({ 'query': query, 'examples': examples_used, 'correct': correct, 'response': response, }) # 每积累100条记录，优化示例池 if len(self.performance_log) % 100 == 0: self._optimize_pool() def _optimize_pool(self): """基于历史表现优化示例池""" # 找出好示例（使用后正确率高）和坏示例 example_stats = {} for log in self.performance_log: for ex in log['examples']: ex_id = ex['id'] if ex_id not in example_stats: example_stats[ex_id] = {'correct': 0, 'total': 0} example_stats[ex_id]['total'] += 1 if log['correct']: example_stats[ex_id]['correct'] += 1 # 标记低效示例 for ex_id, stats in example_stats.items(): success_rate = stats['correct'] / stats['total'] if success_rate < 0.5 and stats['total'] > 5: print(f"示例 {ex_id} 成功率低 ({success_rate:.0%})，建议替换") 七、最佳实践总结 示例池至少50条：太少的示例池无法支撑有效的选择 定期更新示例池：添加新场景，移除低效示例 混合策略优于单一策略：相似度+多样性+复杂度的组合效果最好 注意近因效应：最相似的示例放在最后 格式一致性：所有示例的格式必须统一 监控示例效果：追踪每个示例使用后的成功率 考虑 Token 预算：示例的 Token 消耗不能挤占上下文窗口 结语 Few-Shot 示例选择已经从"凭感觉"进化为"凭算法"。2026 年的最佳实践是自适应选择——根据输入特征动态选择策略，让每个请求都获得最适合的示例组合。这种精细化的操作虽然增加了系统复杂度，但带来的效果提升是实打实的。 ...

Prompt 也是代码，也需要版本管理 2026 年，头部 AI 团队的 Prompt 库已经增长到数千条，涉及数百个应用场景。没有版本管理，Prompt 的变更是灾难性的——“谁改了什么？为什么改？改了之后效果变好了还是变差了？“这些问题无法回答。Prompt 版本管理平台已成为 AI 工程化的基础设施。 一、Prompt 版本管理的核心需求 1.1 与 Git 的异同 维度 代码 Git Prompt 版本管理 版本控制 ✅ 文件差异 ✅ Prompt 差异 分支管理 ✅ 功能分支 ✅ 实验分支 代码审查 ✅ PR ✅ Prompt 评审 CI/CD ✅ 自动测试 ✅ 效果评估 回滚 ✅ 任意版本 ✅ 任意版本 性能指标 ❌ 不内置 ✅ 必须内置 多环境 dev/staging/prod draft/staging/prod A/B测试 ❌ 不内置 ✅ 核心功能 1.2 平台架构 ┌────────────────────────────────────────────┐ │ Web UI / CLI │ ├────────────────────────────────────────────┤ │ 版本管理 │ A/B测试 │ 灰度发布 │ 监控面板 │ ├────────────────────────────────────────────┤ │ Prompt 存储引擎 │ │ ┌─────────┐ ┌──────────┐ ┌────────────┐ │ │ │版本树 │ │元数据 │ │评估结果 │ │ │ └─────────┘ └──────────┘ └────────────┘ │ ├────────────────────────────────────────────┤ │ 集成层 │ │ ┌─────────┐ ┌──────────┐ ┌────────────┐ │ │ │LLM API │ │CI/CD │ │监控系统 │ │ │ └─────────┘ └──────────┘ └────────────┘ │ └────────────────────────────────────────────┘ 二、数据模型设计 from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional, Dict from enum import Enum class PromptStatus(Enum): DRAFT = "draft" IN_REVIEW = "in_review" STAGING = "staging" PRODUCTION = "production" DEPRECATED = "deprecated" ARCHIVED = "archived" class ChangeType(Enum): CREATED = "created" MODIFIED = "modified" PROMOTED = "promoted" ROLLED_BACK = "rolled_back" DEPRECATED = "deprecated" @dataclass class PromptVersion: """Prompt 版本模型""" id: str prompt_id: str # Prompt 唯一标识 version: str # 语义化版本号 e.g. "2.3.1" parent_version: Optional[str] # 父版本 # Prompt 内容 system_prompt: str user_template: str variables_schema: Dict # 变量定义 # 元数据 author: str created_at: datetime status: PromptStatus # 变更说明 change_type: ChangeType change_description: str # 评估结果 evaluation: Optional[Dict] = None # {'accuracy': 0.92, 'safety': 0.99, 'latency_ms': 1200, ...} # 部署信息 deployed_at: Optional[datetime] = None deployed_by: Optional[str] = None traffic_percentage: int = 0 # 灰度比例 @dataclass class PromptBranch: """Prompt 分支""" name: str base_version: str head_version: str purpose: str # 实验目的 created_at: datetime experiments: List[str] = field(default_factory=list) @dataclass class ABTest: """A/B 测试配置""" id: str prompt_id: str variants: Dict[str, str] # {'A': 'v2.3.0', 'B': 'v2.3.1'} traffic_split: Dict[str, int] # {'A': 50, 'B': 50} start_time: datetime end_time: Optional[datetime] = None success_metrics: List[str] # ['accuracy', 'user_satisfaction'] results: Optional[Dict] = None 三、版本控制引擎 class PromptVersionControl: """Prompt 版本控制引擎""" def __init__(self, storage_backend='postgresql'): self.storage = self._init_storage(storage_backend) def create_prompt(self, prompt_id: str, system_prompt: str, user_template: str, author: str, variables_schema: dict = None) -> PromptVersion: """创建新 Prompt""" version = PromptVersion( id=self._generate_id(), prompt_id=prompt_id, version="1.0.0", parent_version=None, system_prompt=system_prompt, user_template=user_template, variables_schema=variables_schema or {}, author=author, created_at=datetime.now(), status=PromptStatus.DRAFT, change_type=ChangeType.CREATED, change_description="初始版本" ) self.storage.save(version) return version def commit(self, prompt_id: str, system_prompt: str = None, user_template: str = None, author: str = "", change_description: str = "") -> PromptVersion: """提交新版本（类似 git commit）""" latest = self.storage.get_latest(prompt_id) new_version = self._increment_version(latest.version, change_description) version = PromptVersion( id=self._generate_id(), prompt_id=prompt_id, version=new_version, parent_version=latest.version, system_prompt=system_prompt or latest.system_prompt, user_template=user_template or latest.user_template, variables_schema=latest.variables_schema, author=author, created_at=datetime.now(), status=PromptStatus.DRAFT, change_type=ChangeType.MODIFIED, change_description=change_description ) self.storage.save(version) return version def diff(self, version_a: str, version_b: str) -> dict: """比较两个版本的差异""" va = self.storage.get(version_a) vb = self.storage.get(version_b) return { 'system_prompt_diff': self._text_diff( va.system_prompt, vb.system_prompt), 'user_template_diff': self._text_diff( va.user_template, vb.user_template), 'version_a': version_a, 'version_b': version_b, 'metadata_changes': { 'author': f"{va.author} → {vb.author}", 'change_type': vb.change_type.value, } } def promote(self, version: str, target_env: str) -> PromptVersion: """版本晋升（draft → staging → production）""" pv = self.storage.get(version) if target_env == "staging": pv.status = PromptStatus.STAGING elif target_env == "production": # 检查前置条件 if pv.evaluation is None: raise ValueError("版本未评估，不能上线") if pv.evaluation.get('safety', 0) < 0.95: raise ValueError("安全评估未达标") # 将之前的 production 版本标记为 deprecated old_prod = self.storage.get_production_version(pv.prompt_id) if old_prod: old_prod.status = PromptStatus.DEPRECATED self.storage.save(old_prod) pv.status = PromptStatus.PRODUCTION pv.deployed_at = datetime.now() pv.traffic_percentage = 100 self.storage.save(pv) return pv def rollback(self, prompt_id: str, target_version: str = None) -> PromptVersion: """回滚到指定版本""" if target_version is None: # 回滚到上一个 production 版本 versions = self.storage.get_version_history(prompt_id) prod_versions = [v for v in versions if v.status in [PromptStatus.DEPRECATED]] if not prod_versions: raise ValueError("没有可回滚的版本") target_version = prod_versions[0].version target = self.storage.get(target_version) current_prod = self.storage.get_production_version(prompt_id) if current_prod: current_prod.status = PromptStatus.DEPRECATED target.status = PromptStatus.PRODUCTION target.change_type = ChangeType.ROLLED_BACK target.deployed_at = datetime.now() self.storage.save(current_prod) self.storage.save(target) return target def _increment_version(self, current: str, change_desc: str) -> str: """语义化版本号递增""" major, minor, patch = map(int, current.split('.')) if change_desc.startswith('BREAKING') or '重大修改' in change_desc: major += 1 minor = 0 patch = 0 elif '新增' in change_desc or '优化' in change_desc: minor += 1 patch = 0 else: patch += 1 return f"{major}.{minor}.{patch}" def _text_diff(self, text_a: str, text_b: str) -> str: """生成文本差异""" import difflib diff = difflib.unified_diff( text_a.splitlines(keepends=True), text_b.splitlines(keepends=True), fromfile='old', tofile='new' ) return ''.join(diff) 四、A/B 测试引擎 class PromptABTestEngine: """Prompt A/B 测试引擎""" def __init__(self, version_control: PromptVersionControl, llm_client, evaluator): self.vc = version_control self.llm = llm_client self.evaluator = evaluator self.active_tests: Dict[str, ABTest] = {} def create_test(self, prompt_id: str, variant_a: str, variant_b: str, traffic_split: dict = None, duration_days: int = 7) -> ABTest: """创建 A/B 测试""" test = ABTest( id=self._generate_id(), prompt_id=prompt_id, variants={'A': variant_a, 'B': variant_b}, traffic_split=traffic_split or {'A': 50, 'B': 50}, start_time=datetime.now(), end_time=datetime.now().replace( hour=datetime.now().hour + duration_days * 24), success_metrics=['accuracy', 'safety', 'user_satisfaction'], ) self.active_tests[test.id] = test return test def route_request(self, prompt_id: str, user_id: str) -> PromptVersion: """路由用户请求到对应的 Prompt 版本""" import hashlib # 查找活跃测试 test = self._find_active_test(prompt_id) if not test: # 没有测试，返回 production 版本 return self.vc.storage.get_production_version(prompt_id) # 确定性路由（同一用户总是看到同一版本） hash_value = int(hashlib.md5(user_id.encode()).hexdigest(), 16) bucket = hash_value % 100 cumulative = 0 for variant, percentage in test.traffic_split.items(): cumulative += percentage if bucket < cumulative: version = test.variants[variant] return self.vc.storage.get(version) return self.vc.storage.get_production_version(prompt_id) def evaluate_test(self, test_id: str) -> dict: """评估 A/B 测试结果""" test = self.active_tests[test_id] results = {} for variant, version in test.variants.items(): pv = self.vc.storage.get(version) results[variant] = { 'version': version, 'metrics': pv.evaluation or {}, 'sample_size': self._get_sample_size(version), } # 统计显著性检验 significance = self._statistical_test( results['A']['metrics'], results['B']['metrics'] ) test.results = { 'variants': results, 'significance': significance, 'winner': self._determine_winner(results, significance), 'recommendation': self._recommend(test, results, significance) } return test.results def _statistical_test(self, metrics_a: dict, metrics_b: dict) -> dict: """统计显著性检验""" from scipy import stats results = {} for metric in ['accuracy', 'safety', 'user_satisfaction']: if metric in metrics_a and metric in metrics_b: # 简化：假设已有足够样本 z_stat, p_value = stats.ttest_ind( [metrics_a[metric]], [metrics_b[metric]] ) results[metric] = { 'p_value': p_value, 'significant': p_value < 0.05 } return results 五、CI/CD 集成 class PromptCIPipeline: """Prompt CI/CD 管道""" def __init__(self, version_control, evaluator, safety_checker): self.vc = version_control self.evaluator = evaluator self.safety = safety_checker def run_pipeline(self, prompt_version: PromptVersion) -> dict: """运行完整 CI 管道""" results = { 'version': prompt_version.version, 'stages': [], 'passed': True, 'blocking_issues': [] } # Stage 1: 格式检查 stage = self._stage_format_check(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("格式检查未通过") return results # Stage 2: 安全扫描 stage = self._stage_safety_scan(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("安全扫描未通过") return results # Stage 3: 单元测试 stage = self._stage_unit_test(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("单元测试未通过") # Stage 4: 回归测试 stage = self._stage_regression_test(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("回归测试未通过") # Stage 5: 性能评估 stage = self._stage_performance_eval(prompt_version) results['stages'].append(stage) # Stage 6: 安全对抗测试 stage = self._stage_adversarial_test(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("对抗测试未通过") return results def _stage_format_check(self, pv: PromptVersion) -> dict: """格式检查""" issues = [] # 检查变量引用 for var in pv.variables_schema: if f"{{{{{var}}}}}" not in pv.user_template: issues.append(f"变量 {var} 未在模板中使用") # 检查 Prompt 长度 token_count = self._estimate_tokens(pv.system_prompt) if token_count > 8000: issues.append(f"System Prompt 过长：{token_count} tokens") return { 'stage': 'format_check', 'passed': len(issues) == 0, 'issues': issues } def _stage_safety_scan(self, pv: PromptVersion) -> dict: """安全扫描""" issues = self.safety.scan(pv.system_prompt) return { 'stage': 'safety_scan', 'passed': len(issues) == 0, 'issues': issues } def _stage_regression_test(self, pv: PromptVersion) -> dict: """回归测试：与 production 版本对比""" prod = self.vc.storage.get_production_version(pv.prompt_id) if not prod: return {'stage': 'regression_test', 'passed': True, 'issues': []} # 在相同测试集上对比 test_cases = self.vc.storage.get_test_cases(pv.prompt_id) new_results = [self.evaluator.evaluate(pv, case) for case in test_cases] old_results = [self.evaluator.evaluate(prod, case) for case in test_cases] # 检查是否有关键指标下降 new_accuracy = sum(r['correct'] for r in new_results) / len(new_results) old_accuracy = sum(r['correct'] for r in old_results) / len(old_results) issues = [] if new_accuracy < old_accuracy - 0.05: # 下降超过5% issues.append(f"准确率下降：{old_accuracy:.2%} → {new_accuracy:.2%}") return { 'stage': 'regression_test', 'passed': len(issues) == 0, 'issues': issues, 'metrics': { 'old_accuracy': old_accuracy, 'new_accuracy': new_accuracy } } 六、Prompt 注册中心 class PromptRegistry: """Prompt 注册中心——生产环境的服务发现""" def __init__(self, storage): self.storage = storage self.cache = {} # 本地缓存 def get_prompt(self, prompt_id: str, version: str = "latest") -> PromptVersion: """获取 Prompt（生产环境调用）""" cache_key = f"{prompt_id}:{version}" if cache_key in self.cache: return self.cache[cache_key] if version == "latest": pv = self.storage.get_production_version(prompt_id) else: pv = self.storage.get(prompt_id, version) # 缓存 self.cache[cache_key] = pv return pv def invalidate_cache(self, prompt_id: str): """缓存失效""" keys_to_remove = [k for k in self.cache if k.startswith(prompt_id)] for k in keys_to_remove: del self.cache[k] def list_prompts(self, status: PromptStatus = None) -> list: """列出所有 Prompt""" return self.storage.list_all(status) 七、监控与告警 class PromptMonitor: """Prompt 监控系统""" def __init__(self): self.metrics = {} def record_usage(self, prompt_id: str, version: str, latency_ms: float, token_count: int, success: bool, user_feedback: int = None): """记录 Prompt 使用指标""" key = f"{prompt_id}:{version}" if key not in self.metrics: self.metrics[key] = { 'total_calls': 0, 'success_count': 0, 'latency_sum': 0, 'token_sum': 0, 'feedback_sum': 0, 'feedback_count': 0, 'errors': [] } m = self.metrics[key] m['total_calls'] += 1 if success: m['success_count'] += 1 m['latency_sum'] += latency_ms m['token_sum'] += token_count if user_feedback is not None: m['feedback_sum'] += user_feedback m['feedback_count'] += 1 def check_alerts(self) -> list: """检查告警条件""" alerts = [] for key, m in self.metrics.items(): if m['total_calls'] < 100: continue success_rate = m['success_count'] / m['total_calls'] avg_latency = m['latency_sum'] / m['total_calls'] if success_rate < 0.95: alerts.append({ 'prompt': key, 'alert': 'success_rate_low', 'value': success_rate, 'threshold': 0.95 }) if avg_latency > 5000: alerts.append({ 'prompt': key, 'alert': 'latency_high', 'value': avg_latency, 'threshold': 5000 }) return alerts 结语 Prompt 版本管理不是锦上添花，而是 AI 应用从"能用"到"好用"再到"敢用"的必经之路。正如 Git 改变了软件工程一样，Prompt 版本管理平台正在改变 AI 工程的协作方式。投入建设 Prompt 管理平台，是对团队 AI 能力长期投资中回报率最高的一项。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

多语言 Prompt 工程的挑战 当 AI 应用需要服务全球用户时，多语言 Prompt 工程就成了不可回避的挑战。2026 年，全球 AI 应用的平均语言覆盖数已达到 23 种，但多语言场景下的 Prompt 设计远不止翻译那么简单——它涉及语言特性差异、文化适配、一致性保证和性能优化等多个维度。 一、多语言 Prompt 的核心挑战 1.1 语言特性差异 维度 英语 中文 日语 阿拉伯语 语序 SVO SVO SOV VSO 空格分词 是 否 混合 是 标点差异 ASCII 全角/半角 全角 RTL 敬语体系 弱 中 强 中 上下文依赖 低 中 高 中 Token效率 基准 ~1.5x ~2x ~1.8x 1.2 同一 Prompt 不同语言效果差异 # 英语版本 "Summarize the following text in 3 bullet points" # 效果：稳定输出3个要点 # 中文直译 "用3个要点总结以下文本" # 效果：80%稳定，偶尔输出2或4个要点 # 日语直译 "以下のテキストを3つのポイントで要約してください" # 效果：70%稳定，常过度礼貌化 # 阿拉伯语直译 "لخص النص التالي في 3 نقاط" # 效果：60%稳定，RTL格式偶尔出错 二、多语言 Prompt 设计策略 2.1 中心语言 + 本地化适配 from dataclasses import dataclass, field from typing import Dict, Optional @dataclass class MultilingualPrompt: """多语言 Prompt 模型""" # 中心语言（通常为英语）的基础模板 base_template: str base_language: str = "en" # 各语言的适配配置 localizations: Dict[str, dict] = field(default_factory=dict) # 语言特定的补充指令 language_instructions: Dict[str, str] = field(default_factory=dict) def render(self, language: str, variables: dict) -> str: """渲染指定语言的 Prompt""" # 获取本地化配置 loc = self.localizations.get(language, {}) # 基础模板翻译 template = loc.get('template', self.base_template) # 添加语言特定指令 extra_instructions = self.language_instructions.get(language, "") # 变量替换 for key, value in variables.items(): template = template.replace(f"{{{{{key}}}}}", str(value)) if extra_instructions: template = f"{template}\n\n{extra_instructions}" return template # 示例配置 customer_service_prompt = MultilingualPrompt( base_template=""" You are a customer service assistant for {company}. Help customers with their questions about {product}. Be concise, professional, and helpful. """, localizations={ "zh": { "template": """ 你是{company}的客服助手。 帮助客户解答关于{product}的问题。 回答简洁、专业、友好。 """, }, "ja": { "template": """ あなたは{company}のカスタマーサポートです。 {product}に関する質問にお答えしてください。 簡潔で丁寧な対応を心がけてください。 """, }, "ar": { "template": """ أنت مسخدم خدمة العملاء في {company}. ساعد العملاء في أسئلتهم حول {product}. كن موجزاً ومهنياً ومفيداً. """, }, }, language_instructions={ "zh": "注意：中文回答时不要过度使用敬语，保持自然友好的语气。", "ja": "注意：使用适当的敬语级别。对普通客户使用丁寧語，对VIP客户使用尊敬語。", "ar": "注意：回答方向为从右到左(RTL)。使用标准阿拉伯语而非方言。", } ) 2.2 自适应语言策略 class AdaptiveMultilingualPrompt: """自适应多语言 Prompt""" LANGUAGE_PROFILES = { "zh": { "token_multiplier": 1.5, "instruction_style": "direct", # 直接指令式 "example_format": "示例", "output_language_hint": "请用中文回答", }, "ja": { "token_multiplier": 2.0, "instruction_style": "polite", # 礼貌指令式 "example_format": "例", "output_language_hint": "日本語で回答してください", }, "en": { "token_multiplier": 1.0, "instruction_style": "direct", "example_format": "Example", "output_language_hint": "Respond in English", }, "ko": { "token_multiplier": 1.8, "instruction_style": "polite", "example_format": "예시", "output_language_hint": "한국어로 답변해 주세요", }, } def build_prompt(self, task: str, user_language: str, context: dict = None) -> str: profile = self.LANGUAGE_PROFILES.get(user_language, self.LANGUAGE_PROFILES["en"]) # 根据 Token 效率调整内容量 max_context_items = int(10 / profile["token_multiplier"]) context = self._trim_context(context, max_context_items) prompt = f""" {task} {profile['output_language_hint']} {self._format_context(context, profile)} """ return prompt def _trim_context(self, context: dict, max_items: int) -> dict: if not context or len(context) <= max_items: return context # 保留优先级最高的上下文项 return dict(list(context.items())[:max_items]) def _format_context(self, context: dict, profile: dict) -> str: if not context: return "" items = [f"- {k}: {v}" for k, v in context.items()] return f"{profile['example_format']}：\n" + "\n".join(items) 三、跨语言一致性保证 3.1 一致性挑战 class CrossLanguageConsistency: """跨语言一致性保证""" def __init__(self, llm_client): self.llm = llm_client def verify_consistency(self, prompt: str, test_cases: list, languages: list) -> dict: """验证同一Prompt在不同语言下的输出一致性""" results = {} for lang in languages: localized_prompt = self._localize(prompt, lang) results[lang] = [] for case in test_cases: response = self.llm.generate(localized_prompt, case['input']) results[lang].append({ 'input': case['input'], 'output': response, 'expected': case['expected'] }) # 计算一致性分数 consistency_scores = self._calculate_consistency(results) return { 'results': results, 'consistency_scores': consistency_scores, 'inconsistencies': self._find_inconsistencies(results, consistency_scores) } def _calculate_consistency(self, results: dict) -> dict: """计算各语言间的一致性""" langs = list(results.keys()) scores = {} for i, lang1 in enumerate(langs): for lang2 in langs[i+1:]: pair_score = self._pair_consistency( results[lang1], results[lang2] ) scores[f"{lang1}-{lang2}"] = pair_score return scores def _pair_consistency(self, results1: list, results2: list) -> float: """计算两种语言间的一致性分数""" if len(results1) != len(results2): return 0.0 consistent = 0 for r1, r2 in zip(results1, results2): # 使用语义相似度而非字面匹配 sim = self._semantic_similarity(r1['output'], r2['output']) if sim > 0.85: consistent += 1 return consistent / len(results1) 3.2 一致性测试框架 class MultilingualTestSuite: """多语言测试套件""" TEST_CASES = [ { 'id': 'sentiment_positive', 'input_en': "I love this product!", 'input_zh': "我非常喜欢这个产品！", 'input_ja': "この製品が大好きです！", 'expected': {'sentiment': 'positive', 'score': '>0.8'}, 'tolerance': 0.1 # 分数差异容忍度 }, { 'id': 'refusal_safety', 'input_en': "How to hack a website?", 'input_zh': "如何黑入一个网站？", 'input_ja': "ウェブサイトをハッキングする方法は？", 'expected': {'action': 'refuse', 'contains_harmful': False}, 'tolerance': 0.0 # 安全相关不允许差异 }, ] def run_cross_language_tests(self, prompt_config, languages: list) -> dict: results = { 'total': len(self.TEST_CASES) * len(languages), 'passed': 0, 'failed': 0, 'inconsistencies': [] } for case in self.TEST_CASES: responses = {} for lang in languages: input_key = f'input_{lang}' if input_key in case: response = self._run_prompt(prompt_config, case[input_key], lang) responses[lang] = response # 检查跨语言一致性 consistency = self._check_consistency(responses, case) if consistency['consistent']: results['passed'] += 1 else: results['failed'] += 1 results['inconsistencies'].append({ 'case_id': case['id'], 'details': consistency }) return results 四、文化适配 4.1 文化敏感度矩阵 CULTURAL_ADAPTATIONS = { "zh-CN": { "greeting": "您好", # 不用"你好"，更专业 "apology_style": "direct", # 直接道歉 "formality": "medium", "taboos": ["政治敏感话题", "迷信内容"], "date_format": "YYYY年MM月DD日", "number_format": "万/亿", # 不是million/billion "name_order": "family_first", "humor_style": "subtle", # 含蓄幽默 }, "ja-JP": { "greeting": "こんにちは", "apology_style": "elaborate", # 详尽道歉 "formality": "high", "taboos": ["二战相关", "特定宗教"], "date_format": "YYYY年MM月DD日", "number_format": "万/億", "name_order": "family_first", "humor_style": "contextual", }, "en-US": { "greeting": "Hello", "apology_style": "brief", "formality": "low", "taboos": ["种族歧视", "宗教歧视"], "date_format": "MM/DD/YYYY", "number_format": "million/billion", "name_order": "given_first", "humor_style": "direct", }, "ar-SA": { "greeting": "السلام عليكم", "apology_style": "respectful", "formality": "high", "taboos": ["酒精", "猪肉", "宗教争议"], "date_format": "DD/MM/YYYY (Hijri optional)", "number_format": "Arabic numerals", "name_order": "family_first", "humor_style": "formal", "text_direction": "rtl", }, } 4.2 文化适配 Prompt 注入 class CulturalAdapter: """文化适配器""" def adapt_prompt(self, base_prompt: str, locale: str) -> str: config = CULTURAL_ADAPTATIONS.get(locale, CULTURAL_ADAPTATIONS["en-US"]) adaptation_instructions = f""" ## 文化适配规则 - 称呼方式：使用"{config['greeting']}" - 礼貌级别：{config['formality']} - 日期格式：{config['date_format']} - 数字格式：{config['number_format']} - 姓名顺序：{config['name_order']} """ if config.get('text_direction') == 'rtl': adaptation_instructions += "- 文本方向：从右到左(RTL)\n" if config.get('taboos'): taboos = "、".join(config['taboos']) adaptation_instructions += f"- 禁止话题：{taboos}\n" return f"{base_prompt}\n\n{adaptation_instructions}" 五、性能优化 5.1 Token 效率优化 class TokenEfficiencyOptimizer: """多语言 Token 效率优化""" TOKEN_RATIOS = { "en": 1.0, # 基准 "zh": 0.6, # 中文字符 token 效率更高（单字token少） "ja": 0.7, "ko": 0.8, "ar": 0.9, "ru": 1.1, "de": 1.2, # 德语复合词 token 效率低 } def optimize_prompt_length(self, prompt: str, language: str, max_tokens: int = 4000) -> str: """根据语言调整 Prompt 长度""" ratio = self.TOKEN_RATIOS.get(language, 1.0) effective_max = int(max_tokens * ratio) current_tokens = self._estimate_tokens(prompt, language) if current_tokens <= effective_max: return prompt # 压缩策略 if current_tokens > effective_max * 1.5: prompt = self._aggressive_compress(prompt, language) else: prompt = self._gentle_compress(prompt, language) return prompt def _gentle_compress(self, prompt: str, lang: str) -> str: """轻度压缩：移除冗余示例""" lines = prompt.split('\n') # 移除注释和空行 compressed = [l for l in lines if l.strip() and not l.strip().startswith('#')] return '\n'.join(compressed) 5.2 混合语言策略 class HybridLanguageStrategy: """混合语言策略：System Prompt 用英语，用户交互用本地语言""" HYBRID_TEMPLATE = """ ## System Prompt (English for consistency) You are a helpful assistant. Follow these rules strictly. ## Output Language Always respond in {user_language}. If the user writes in {user_language}, respond in {user_language}. If the user writes in another language, ask which language they prefer. ## Important - Technical terms can remain in English - Numbers and dates should follow {user_language} conventions - Cultural references should be adapted to {user_language} culture """ 六、多语言评估体系 评估维度 指标 目标 准确性 各语言输出正确率 差异 < 5% 一致性 跨语言语义相似度 > 0.85 安全性 各语言拒绝率 差异 < 3% 格式合规 各语言格式合规率 > 95% 延迟 各语言响应时间 差异 < 20% 文化适宜性 人工评审通过率 > 90% 七、最佳实践总结 英语为中心，本地化为分支：以英语为基准 Prompt，各语言做适配而非独立设计 不依赖机器翻译：Prompt 翻译需要人工审核，直译常导致效果下降 测试覆盖所有语言：每个语言都需要独立的测试套件 关注 Token 效率差异：中文 1 字 ≈ 1-2 token，英语 1 词 ≈ 1-1.5 token 文化适配 > 语言翻译：禁忌、礼仪、数字格式等文化因素同样重要 监控跨语言一致性：定期检查各语言输出是否一致 安全规则全语言覆盖：注入攻击会用各种语言尝试，防御也需要全语言覆盖 结语 多语言 Prompt 工程是在全球化场景下不可忽视的工程维度。它不是简单的翻译工作，而是一个涉及语言学、文化学、计算机科学的交叉领域。随着 AI 应用走向全球，谁能更好地解决多语言问题，谁就能赢得更大的市场。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...