本章导读

提示词工程的效率和质量很大程度上取决于所使用的工具。本章将系统介绍当前主流的提示词工程工具链，包括提示词开发工具、测试调试工具、管理工具以及大模型应用开发框架，帮助读者构建完整的提示词工程工具体系。

18.1 提示词开发工具：LangChain Prompt Hub、DSPy

18.1.1 LangChain Prompt Hub

LangChain Prompt Hub是一个用于发现、分享和管理提示词的中心化平台。

核心功能：

# LangChain Prompt Hub 使用示例"""# 安装pip install langchain langchainhub# 从Hub拉取提示词from langchain import hub# 获取特定提示词prompt = hub.pull("hwchase17/openai-functions-template")# 查看提示词内容print(prompt.template)"""class PromptHubConcept:"""Prompt Hub 核心概念说明"""CONCEPTS = {"repository": {"description": "提示词仓库","features": ["版本控制：每个提示词都有版本历史","分类标签：按用途、领域、模型分类","社区贡献：开源社区共享优质提示词","搜索发现：通过关键词快速找到所需提示词"]},"versioning": {"description": "版本管理","features": ["语义化版本：major.minor.patch","变更历史：记录每次修改的内容","回滚能力：随时回退到历史版本","版本对比：比较不同版本的差异"]},"collaboration": {"description": "协作功能","features": ["Fork：基于现有提示词创建变体","PR：提交改进建议","评论：讨论提示词优化方案","评分：社区评分机制"]}}

自定义Prompt Hub实现：

from typing import Dict, List, Optionalfrom dataclasses import dataclass, asdictfrom datetime import datetimeimport json@dataclassclass HubPrompt:"""Hub提示词定义"""id: strname: strdescription: strtemplate: strtags: List[str]author: strversion: strvariables: List[str]created_at: datetimeupdated_at: datetimeusage_count: int = 0rating: float = 0.0def to_dict(self) -> Dict:return asdict(self)@classmethoddef from_dict(cls, data: Dict) -> 'HubPrompt':return cls(**data)class LocalPromptHub:"""本地Prompt Hub实现"""def __init__(self, storage_path: str = "./prompt_hub"):self.storage_path = storage_pathself.prompts: Dict[str, HubPrompt] = {}self.index = {"tags": {}, "authors": {}}self._load_prompts()def _load_prompts(self):"""加载已保存的提示词"""import osif os.path.exists(f"{self.storage_path}/prompts.json"):with open(f"{self.storage_path}/prompts.json", "r") as f:data = json.load(f)for prompt_id, prompt_data in data.items():self.prompts[prompt_id] = HubPrompt.from_dict(prompt_data)self._update_index(self.prompts[prompt_id])def _save_prompts(self):"""保存提示词到文件"""import osos.makedirs(self.storage_path, exist_ok=True)data = {k: v.to_dict() for k, v in self.prompts.items()}with open(f"{self.storage_path}/prompts.json", "w") as f:json.dump(data, f, indent=2, default=str)def _update_index(self, prompt: HubPrompt):"""更新索引"""# 标签索引for tag in prompt.tags:if tag not in self.index["tags"]:self.index["tags"][tag] = []if prompt.id not in self.index["tags"][tag]:self.index["tags"][tag].append(prompt.id)# 作者索引if prompt.author not in self.index["authors"]:self.index["authors"][prompt.author] = []if prompt.id not in self.index["authors"][prompt.author]:self.index["authors"][prompt.author].append(prompt.id)def publish(self, prompt: HubPrompt) -> str:"""发布提示词"""if prompt.id in self.prompts:# 更新现有提示词prompt.updated_at = datetime.now()# 版本升级parts = prompt.version.split(".")parts[2] = str(int(parts[2]) + 1)prompt.version = ".".join(parts)else:prompt.created_at = datetime.now()prompt.updated_at = datetime.now()self.prompts[prompt.id] = promptself._update_index(prompt)self._save_prompts()return prompt.iddef pull(self, prompt_id: str, version: str = None) -> Optional[HubPrompt]:"""拉取提示词"""prompt = self.prompts.get(prompt_id)if prompt:prompt.usage_count += 1self._save_prompts()return promptdef search(self, query: str = None, tags: List[str] = None, author: str = None) -> List[HubPrompt]:"""搜索提示词"""results = list(self.prompts.values())if query:query_lower = query.lower()results = [p for p in resultsif query_lower in p.name.lower() or query_lower in p.description.lower()or query_lower in p.template.lower()]if tags:results = [p for p in resultsif any(tag in p.tags for tag in tags)]if author:results = [p for p in results if p.author == author]# 按评分和使用量排序results.sort(key=lambda p: (p.rating, p.usage_count), reverse=True)return resultsdef get_popular(self, limit: int = 10) -> List[HubPrompt]:"""获取热门提示词"""sorted_prompts = sorted(self.prompts.values(),key=lambda p: p.usage_count,reverse=True)return sorted_prompts[:limit]def rate(self, prompt_id: str, rating: float):"""评分"""if prompt_id in self.prompts:prompt = self.prompts[prompt_id]# 简单的加权平均prompt.rating = (prompt.rating * prompt.usage_count + rating) / (prompt.usage_count + 1)self._save_prompts()# 使用示例hub = LocalPromptHub()# 创建并发布提示词new_prompt = HubPrompt(id="summarization/basic",name="基础摘要生成",description="生成简洁的文本摘要",template="""请为以下文本生成摘要：文本：{text}要求：- 摘要长度：不超过{max_length}字- 包含主要观点- 保持客观中立摘要：""",tags=["summarization", "nlp", "basic"],author="user001",version="1.0.0",variables=["text", "max_length"],created_at=datetime.now(),updated_at=datetime.now())hub.publish(new_prompt)# 搜索提示词results = hub.search(query="摘要", tags=["nlp"])print(f"找到 {len(results)} 个相关提示词")# 拉取提示词pulled = hub.pull("summarization/basic")if pulled:print(f"提示词模板：{pulled.template[:100]}...")

18.1.2 DSPy框架

DSPy（Declarative Self-improving Python）是一个用于算法优化提示词和权重的框架。

核心概念：

"""DSPy 核心概念：1. Signatures（签名） - 声明式定义任务的输入输出 - 替代手工编写的提示词2. Modules（模块） - 预置的提示模式（ChainOfThought, ReAct等） - 可组合的构建块3. Optimizers（优化器） - 自动优化提示词和示例 - 基于训练数据学习最佳配置"""# DSPy 概念性示例（实际使用需要安装dspy库）"""import dspy# 定义签名class Summarize(dspy.Signature):document = dspy.InputField()summary = dspy.OutputField(desc="简洁的文档摘要")# 使用模块summarizer = dspy.ChainOfThought(Summarize)# 执行result = summarizer(document="长文本内容...")print(result.summary)# 优化from dspy.teleprompt import BootstrapFewShotoptimizer = BootstrapFewShot(metric=summarize_metric)optimized_summarizer = optimizer.compile(summarizer, trainset=train_data)"""

DSPy风格提示词优化实现：

from typing import Callable, List, Dict, Anyfrom dataclasses import dataclassimport random@dataclassclass Example:"""训练示例"""inputs: Dict[str, Any]outputs: Dict[str, Any]@dataclassclass Signature:"""任务签名"""name: strinput_fields: List[str]output_fields: List[str]instructions: strclass DSPyStyleOptimizer:"""DSPy风格的提示词优化器"""def __init__(self, signature: Signature, llm_client):self.signature = signatureself.llm_client = llm_clientself.demos = []# 示例集合self.optimized_prompt = Nonedef add_demo(self, example: Example):"""添加示例"""self.demos.append(example)def build_base_prompt(self) -> str:"""构建基础提示词"""prompt = f"{self.signature.instructions}nn"# 添加输入字段说明prompt += "输入字段：n"for field in self.signature.input_fields:prompt += f"- {field}n"# 添加输出字段说明prompt += "n输出字段：n"for field in self.signature.output_fields:prompt += f"- {field}n"return promptdef build_few_shot_prompt(self, num_demos: int = 3) -> str:"""构建Few-shot提示词"""prompt = self.build_base_prompt()# 选择示例selected_demos = random.sample(self.demos, min(num_demos, len(self.demos)))if selected_demos:prompt += "n示例：n"for i, demo in enumerate(selected_demos, 1):prompt += f"n示例 {i}:n"# 输入for field, value in demo.inputs.items():prompt += f"{field}: {value}n"# 输出for field, value in demo.outputs.items():prompt += f"{field}: {value}n"prompt += "n现在请处理以下输入：n"return promptdef optimize(self, metric_fn: Callable, trainset: List[Example],num_iterations: int = 10) -> str:"""优化提示词"""best_score = 0best_prompt = Nonebest_demos = []for iteration in range(num_iterations):# 随机选择示例组合num_demos = random.randint(1, min(5, len(trainset)))selected = random.sample(trainset, num_demos)# 构建提示词self.demos = selectedprompt = self.build_few_shot_prompt(num_demos)# 评估score = self._evaluate_prompt(prompt, metric_fn, trainset)if score > best_score:best_score = scorebest_prompt = promptbest_demos = selected.copy()self.optimized_prompt = best_promptself.demos = best_demosreturn best_promptdef _evaluate_prompt(self, prompt: str, metric_fn: Callable, trainset: List[Example]) -> float:"""评估提示词效果"""scores = []# 在验证集上测试for example in trainset[:10]:# 使用前10个示例评估try:# 构建完整提示词full_prompt = promptfor field, value in example.inputs.items():full_prompt += f"{field}: {value}n"# 调用模型output = self.llm_client.generate(full_prompt)# 解析输出predicted = self._parse_output(output)# 计算分数score = metric_fn(predicted, example.outputs)scores.append(score)except Exception as e:scores.append(0)return sum(scores) / len(scores) if scores else 0def _parse_output(self, output: str) -> Dict[str, Any]:"""解析模型输出"""result = {}lines = output.strip().split('n')for line in lines:if ':' in line:key, value = line.split(':', 1)result[key.strip()] = value.strip()return result# 使用示例def mock_llm_client():"""模拟LLM客户端"""class MockLLM:def generate(self, prompt: str) -> str:return "summary: 这是一个摘要nkey_points: 要点1, 要点2"return MockLLM()signature = Signature(name="text_summarization",input_fields=["document"],output_fields=["summary", "key_points"],instructions="请为输入文档生成摘要和关键要点。")optimizer = DSPyStyleOptimizer(signature, mock_llm_client())# 添加训练示例train_examples = [Example(inputs={"document": "长文本1..."},outputs={"summary": "摘要1", "key_points": "要点1, 要点2"}),Example(inputs={"document": "长文本2..."},outputs={"summary": "摘要2", "key_points": "要点A, 要点B"})]for ex in train_examples:optimizer.add_demo(ex)# 构建提示词prompt = optimizer.build_few_shot_prompt()print("优化后的提示词：")print(prompt)

18.1.3 其他提示词开发工具

OTHER_DEV_TOOLS = """其他提示词开发工具：1. **PromptIDE** - OpenAI提供的提示词开发环境 - 支持版本控制、协作、测试 - 与OpenAI API深度集成2. **Promptmetheus** - 可视化提示词编辑器 - 支持变量、条件、循环 - 实时预览和测试3. **PromptLayer** - 提示词版本管理和追踪 - 性能分析和A/B测试 - 团队协作功能4. **Humanloop** - 提示词工程和评估平台 - 人工反馈集成 - 模型微调支持5. **Vellum** - 企业级提示词管理平台 - 部署和监控功能 - 多环境支持"""

18.2 提示词测试与调试工具：LangSmith、PromptEval

18.2.1 LangSmith

LangSmith是LangChain提供的全生命周期LLM应用开发平台，包含强大的测试和调试功能。

核心功能：

LANGSMITH_FEATURES = """LangSmith 核心功能：1. **Tracing（追踪）** - 记录完整的调用链路 - 可视化Chain执行流程 - 查看每一步的输入输出2. **Debugging（调试）** - 查看实际发送的提示词 - 检查模型参数 - 分析token使用情况3. **Testing（测试）** - 批量测试数据集 - 自动化评估 - 回归测试4. **Monitoring（监控）** - 生产环境监控 - 性能指标追踪 - 错误告警"""# LangSmith 使用示例"""import osos.environ["LANGCHAIN_API_KEY"] = "your-api-key"os.environ["LANGCHAIN_PROJECT"] = "my-project"from langchain import OpenAI, LLMChain, PromptTemplatefrom langchain.callbacks import LangChainTracer# 启用追踪tracer = LangChainTracer()# 创建Chainllm = OpenAI()template = PromptTemplate(input_variables=["product"],template="为{product}写一段营销文案。")chain = LLMChain(llm=llm, prompt=template, callbacks=[tracer])# 执行（会自动记录到LangSmith）result = chain.run("智能手表")"""

本地调试工具实现：

import jsonfrom datetime import datetimefrom typing import Dict, List, Any, Optionalfrom dataclasses import dataclass, field, asdict@dataclassclass TraceStep:"""追踪步骤"""step_name: strstep_type: str# prompt, llm_call, tool_use, etc.inputs: Dict[str, Any]outputs: Dict[str, Any]start_time: datetimeend_time: Optional[datetime] = Nonelatency_ms: float = 0.0token_usage: Dict[str, int] = field(default_factory=dict)metadata: Dict[str, Any] = field(default_factory=dict)@dataclassclass Trace:"""追踪记录"""trace_id: strproject_name: strstart_time: datetimeend_time: Optional[datetime] = Nonesteps: List[TraceStep] = field(default_factory=list)metadata: Dict[str, Any] = field(default_factory=dict)class LocalDebugger:"""本地调试器"""def __init__(self, project_name: str = "default"):self.project_name = project_nameself.traces: List[Trace] = []self.current_trace: Optional[Trace] = Noneself.current_step: Optional[TraceStep] = Nonedef start_trace(self, trace_id: str = None, metadata: Dict = None) -> str:"""开始追踪"""if trace_id is None:trace_id = f"trace_{datetime.now().strftime('%Y%m%d%H%M%S')}"self.current_trace = Trace(trace_id=trace_id,project_name=self.project_name,start_time=datetime.now(),metadata=metadata or {})return trace_iddef start_step(self, step_name: str, step_type: str, inputs: Dict[str, Any]):"""开始步骤"""if self.current_trace is None:self.start_trace()self.current_step = TraceStep(step_name=step_name,step_type=step_type,inputs=inputs,start_time=datetime.now())def end_step(self, outputs: Dict[str, Any], token_usage: Dict[str, int] = None,metadata: Dict[str, Any] = None):"""结束步骤"""if self.current_step is None:returnself.current_step.end_time = datetime.now()self.current_step.outputs = outputsself.current_step.latency_ms = (self.current_step.end_time - self.current_step.start_time).total_seconds() * 1000if token_usage:self.current_step.token_usage = token_usageif metadata:self.current_step.metadata = metadataself.current_trace.steps.append(self.current_step)self.current_step = Nonedef end_trace(self, metadata: Dict = None):"""结束追踪"""if self.current_trace is None:returnself.current_trace.end_time = datetime.now()if metadata:self.current_trace.metadata.update(metadata)self.traces.append(self.current_trace)self.current_trace = Nonedef get_trace(self, trace_id: str) -> Optional[Trace]:"""获取追踪记录"""for trace in self.traces:if trace.trace_id == trace_id:return tracereturn Nonedef visualize_trace(self, trace_id: str) -> str:"""可视化追踪"""trace = self.get_trace(trace_id)if not trace:return "Trace not found"output = f"""Trace ID: {trace.trace_id}Project: {trace.project_name}Duration: {(trace.end_time - trace.start_time).total_seconds():.2f}sSteps:"""for i, step in enumerate(trace.steps, 1):output += f"n{i}. {step.step_name} ({step.step_type})n"output += f" Latency: {step.latency_ms:.2f}msn"if step.token_usage:output += f" Tokens: {step.token_usage}n"output += f" Inputs: {json.dumps(step.inputs, ensure_ascii=False)[:100]}...n"output += f" Outputs: {json.dumps(step.outputs, ensure_ascii=False)[:100]}...n"return outputdef export_traces(self, filepath: str):"""导出追踪记录"""data = [asdict(trace) for trace in self.traces]with open(filepath, 'w') as f:json.dump(data, f, indent=2, default=str)def analyze_performance(self) -> Dict:"""分析性能"""if not self.traces:return {}all_steps = []for trace in self.traces:all_steps.extend(trace.steps)if not all_steps:return {}latencies = [s.latency_ms for s in all_steps]return {"total_traces": len(self.traces),"total_steps": len(all_steps),"avg_latency_ms": sum(latencies) / len(latencies),"max_latency_ms": max(latencies),"min_latency_ms": min(latencies),"step_type_distribution": self._count_by_type(all_steps)}def _count_by_type(self, steps: List[TraceStep]) -> Dict[str, int]:"""按类型统计"""counts = {}for step in steps:counts[step.step_type] = counts.get(step.step_type, 0) + 1return counts# 使用示例debugger = LocalDebugger(project_name="my_chatbot")# 开始追踪debugger.start_trace("chat_001", metadata={"user_id": "user123"})# 记录提示词构建步骤debugger.start_step("build_prompt", "prompt",inputs={"template": "你好，{name}", "variables": {"name": "张三"}})debugger.end_step(outputs={"full_prompt": "你好，张三"})# 记录LLM调用步骤debugger.start_step("llm_call", "llm_call",inputs={"prompt": "你好，张三", "model": "gpt-4"})debugger.end_step(outputs={"response": "你好！有什么可以帮助你的？"},token_usage={"prompt_tokens": 10, "completion_tokens": 15, "total_tokens": 25},metadata={"temperature": 0.7})# 结束追踪debugger.end_trace()# 可视化print(debugger.visualize_trace("chat_001"))# 性能分析perf = debugger.analyze_performance()print(f"平均延迟: {perf.get('avg_latency_ms', 0):.2f}ms")

18.2.2 PromptEval测试框架

from typing import List, Dict, Callable, Anyfrom dataclasses import dataclassfrom concurrent.futures import ThreadPoolExecutor, as_completedimport time@dataclassclass TestCase:"""测试用例"""name: strinputs: Dict[str, Any]expected_outputs: Dict[str, Any] = Noneexpected_behavior: str = Nonetags: List[str] = None@dataclassclass TestResult:"""测试结果"""test_name: strpassed: boolactual_output: Anyexecution_time_ms: floaterror_message: str = Nonedetails: Dict = Noneclass PromptEvalFramework:"""提示词评估框架"""def __init__(self):self.test_cases: List[TestCase] = []self.evaluators: Dict[str, Callable] = {}self.results: List[TestResult] = []def add_test_case(self, test_case: TestCase):"""添加测试用例"""self.test_cases.append(test_case)def register_evaluator(self, name: str, evaluator: Callable):"""注册评估器"""self.evaluators[name] = evaluatordef run_tests(self, prompt_executor: Callable, parallel: bool = False) -> List[TestResult]:"""运行测试"""self.results = []if parallel:with ThreadPoolExecutor(max_workers=5) as executor:futures = {executor.submit(self._run_single_test, tc, prompt_executor): tcfor tc in self.test_cases}for future in as_completed(futures):result = future.result()self.results.append(result)else:for test_case in self.test_cases:result = self._run_single_test(test_case, prompt_executor)self.results.append(result)return self.resultsdef _run_single_test(self, test_case: TestCase, prompt_executor: Callable) -> TestResult:"""运行单个测试"""start_time = time.time()try:# 执行提示词actual_output = prompt_executor(test_case.inputs)# 评估结果passed = self._evaluate_result(test_case, actual_output)execution_time = (time.time() - start_time) * 1000return TestResult(test_name=test_case.name,passed=passed,actual_output=actual_output,execution_time_ms=execution_time)except Exception as e:execution_time = (time.time() - start_time) * 1000return TestResult(test_name=test_case.name,passed=False,actual_output=None,execution_time_ms=execution_time,error_message=str(e))def _evaluate_result(self, test_case: TestCase, actual_output: Any) -> bool:"""评估结果"""# 如果有期望输出，进行精确匹配if test_case.expected_outputs:return actual_output == test_case.expected_outputs# 如果有期望行为描述，使用LLM评估if test_case.expected_behavior:# 这里可以调用LLM进行评估return True# 简化处理# 默认通过return Truedef generate_report(self) -> Dict:"""生成测试报告"""if not self.results:return {"error": "No test results"}total = len(self.results)passed = sum(1 for r in self.results if r.passed)failed = total - passedavg_time = sum(r.execution_time_ms for r in self.results) / totalfailed_tests = [{"name": r.test_name,"error": r.error_message,"output": r.actual_output}for r in self.results if not r.passed]return {"summary": {"total": total,"passed": passed,"failed": failed,"pass_rate": passed / total if total > 0 else 0,"avg_execution_time_ms": avg_time},"failed_tests": failed_tests,"all_results": [{"name": r.test_name,"passed": r.passed,"time_ms": r.execution_time_ms}for r in self.results]}# 使用示例eval_framework = PromptEvalFramework()# 添加测试用例eval_framework.add_test_case(TestCase(name="basic_greeting",inputs={"message": "你好"},expected_behavior="应该礼貌地回应问候"))eval_framework.add_test_case(TestCase(name="empty_input",inputs={"message": ""},expected_behavior="应该优雅地处理空输入"))eval_framework.add_test_case(TestCase(name="long_input",inputs={"message": "很长的文本..." * 100},expected_behavior="应该能够处理长文本"))# 定义执行器def mock_executor(inputs):return {"response": f"收到: {inputs.get('message', '')[:20]}"}# 运行测试results = eval_framework.run_tests(mock_executor)# 生成报告report = eval_framework.generate_report()print(f"通过率: {report['summary']['pass_rate']:.2%}")print(f"失败测试数: {len(report['failed_tests'])}")

18.3 提示词管理工具：版本控制、权限管理、知识库

18.3.1 提示词版本控制系统

import hashlibfrom typing import Dict, List, Optional, Tuplefrom dataclasses import dataclass, asdictfrom datetime import datetimeimport jsonimport difflib@dataclassclass PromptVersion:"""提示词版本"""version_id: strprompt_id: strcontent: strauthor: strcreated_at: datetimecommit_message: strparent_version: Optional[str] = Nonetags: List[str] = Nonemetadata: Dict = Nonedef __post_init__(self):if self.tags is None:self.tags = []if self.metadata is None:self.metadata = {}@propertydef content_hash(self) -> str:"""计算内容哈希"""return hashlib.sha256(self.content.encode()).hexdigest()[:16]class PromptVersionControl:"""提示词版本控制系统"""def __init__(self, storage_path: str = "./prompt_vcs"):self.storage_path = storage_pathself.versions: Dict[str, List[PromptVersion]] = {}self.branches: Dict[str, Dict[str, str]] = {}# prompt_id -> {branch_name: version_id}self._load_data()def _load_data(self):"""加载数据"""import osversions_file = f"{self.storage_path}/versions.json"branches_file = f"{self.storage_path}/branches.json"if os.path.exists(versions_file):with open(versions_file, 'r') as f:data = json.load(f)for prompt_id, versions_data in data.items():self.versions[prompt_id] = [PromptVersion(**v) for v in versions_data]if os.path.exists(branches_file):with open(branches_file, 'r') as f:self.branches = json.load(f)def _save_data(self):"""保存数据"""import osos.makedirs(self.storage_path, exist_ok=True)versions_data = {k: [asdict(v) for v in vs]for k, vs in self.versions.items()}with open(f"{self.storage_path}/versions.json", 'w') as f:json.dump(versions_data, f, indent=2, default=str)with open(f"{self.storage_path}/branches.json", 'w') as f:json.dump(self.branches, f, indent=2)def commit(self, prompt_id: str, content: str, author: str,message: str, tags: List[str] = None) -> str:"""提交新版本"""# 生成版本IDtimestamp = datetime.now().strftime("%Y%m%d%H%M%S")version_id = f"{prompt_id}_v{timestamp}"# 获取父版本parent_version = Noneif prompt_id in self.versions and self.versions[prompt_id]:parent_version = self.versions[prompt_id][-1].version_id# 创建新版本version = PromptVersion(version_id=version_id,prompt_id=prompt_id,content=content,author=author,created_at=datetime.now(),commit_message=message,parent_version=parent_version,tags=tags or [])# 保存if prompt_id not in self.versions:self.versions[prompt_id] = []self.versions[prompt_id].append(version)# 更新主分支if prompt_id not in self.branches:self.branches[prompt_id] = {}self.branches[prompt_id]["main"] = version_idself._save_data()return version_iddef get_version(self, prompt_id: str,version_id: str = None) -> Optional[PromptVersion]:"""获取特定版本"""if prompt_id not in self.versions:return Noneif version_id is None:# 返回最新版本return self.versions[prompt_id][-1] if self.versions[prompt_id] else Nonefor version in self.versions[prompt_id]:if version.version_id == version_id:return versionreturn Nonedef get_history(self, prompt_id: str) -> List[PromptVersion]:"""获取版本历史"""return self.versions.get(prompt_id, [])def diff(self, prompt_id: str, version_id1: str, version_id2: str) -> str:"""比较两个版本的差异"""v1 = self.get_version(prompt_id, version_id1)v2 = self.get_version(prompt_id, version_id2)if not v1 or not v2:return "Version not found"diff = difflib.unified_diff(v1.content.splitlines(keepends=True),v2.content.splitlines(keepends=True),fromfile=version_id1,tofile=version_id2)return ''.join(diff)def checkout(self, prompt_id: str, version_id: str) -> Optional[str]:"""检出特定版本"""version = self.get_version(prompt_id, version_id)if version:return version.contentreturn Nonedef create_branch(self, prompt_id: str, branch_name: str,from_version: str = None):"""创建分支"""if prompt_id not in self.branches:self.branches[prompt_id] = {}if from_version is None:# 从最新版本创建latest = self.get_version(prompt_id)from_version = latest.version_id if latest else Noneself.branches[prompt_id][branch_name] = from_versionself._save_data()def merge_branch(self, prompt_id: str, branch_name: str, target_branch: str = "main"):"""合并分支"""# 简化实现：将分支的最新版本复制到目标分支if prompt_id not in self.branches:return Falseif branch_name not in self.branches[prompt_id]:return Falsebranch_version_id = self.branches[prompt_id][branch_name]self.branches[prompt_id][target_branch] = branch_version_idself._save_data()return Truedef tag(self, prompt_id: str, version_id: str, tag: str):"""给版本打标签"""version = self.get_version(prompt_id, version_id)if version:version.tags.append(tag)self._save_data()def find_by_tag(self, prompt_id: str, tag: str) -> List[PromptVersion]:"""通过标签查找版本"""versions = self.versions.get(prompt_id, [])return [v for v in versions if tag in v.tags]# 使用示例vcs = PromptVersionControl()# 提交版本v1 = vcs.commit(prompt_id="customer_service",content="你是客服助手...",author="alice",message="初始版本")v2 = vcs.commit(prompt_id="customer_service",content="你是专业的客服助手...",author="bob",message="增加专业性描述")# 查看历史history = vcs.get_history("customer_service")print(f"共有 {len(history)} 个版本")# 比较差异diff = vcs.diff("customer_service", v1, v2)print("版本差异：")print(diff)# 打标签vcs.tag("customer_service", v2, "stable")# 查找标签stable_versions = vcs.find_by_tag("customer_service", "stable")print(f"稳定版本数: {len(stable_versions)}")

18.3.2 权限管理系统

from enum import Enumfrom typing import Dict, List, Setfrom dataclasses import dataclassclass Permission(Enum):"""权限枚举"""READ = "read"WRITE = "write"DELETE = "delete"EXECUTE = "execute"ADMIN = "admin"@dataclassclass User:"""用户"""user_id: strname: strrole: strgroups: List[str]@dataclassclass AccessControl:"""访问控制"""resource_id: strowner: strpermissions: Dict[str, List[Permission]]# user/group -> permissionsclass PromptAccessManager:"""提示词访问管理器"""def __init__(self):self.users: Dict[str, User] = {}self.access_controls: Dict[str, AccessControl] = {}self.role_permissions = {"admin": [Permission.READ, Permission.WRITE, Permission.DELETE,Permission.EXECUTE, Permission.ADMIN],"editor": [Permission.READ, Permission.WRITE, Permission.EXECUTE],"viewer": [Permission.READ, Permission.EXECUTE],"guest": [Permission.READ]}def register_user(self, user: User):"""注册用户"""self.users[user.user_id] = userdef create_resource(self, resource_id: str, owner_id: str):"""创建资源"""self.access_controls[resource_id] = AccessControl(resource_id=resource_id,owner=owner_id,permissions={owner_id: self.role_permissions["admin"]})def grant_permission(self, resource_id: str, user_id: str, permissions: List[Permission]):"""授予权限"""if resource_id not in self.access_controls:return Falseac = self.access_controls[resource_id]# 检查授予者是否有ADMIN权限# 简化处理：假设调用者有权限ac.permissions[user_id] = permissionsreturn Truedef revoke_permission(self, resource_id: str, user_id: str):"""撤销权限"""if resource_id not in self.access_controls:return Falseac = self.access_controls[resource_id]if user_id in ac.permissions:del ac.permissions[user_id]return Truereturn Falsedef check_permission(self, resource_id: str, user_id: str, permission: Permission) -> bool:"""检查权限"""if resource_id not in self.access_controls:return Falseac = self.access_controls[resource_id]user = self.users.get(user_id)if not user:return False# 检查直接权限if user_id in ac.permissions:if permission in ac.permissions[user_id]:return True# 检查组权限for group in user.groups:if group in ac.permissions:if permission in ac.permissions[group]:return True# 检查角色权限if user.role in self.role_permissions:if permission in self.role_permissions[user.role]:return Truereturn Falsedef get_accessible_resources(self, user_id: str,permission: Permission = Permission.READ) -> List[str]:"""获取用户可访问的资源"""resources = []for resource_id, ac in self.access_controls.items():if self.check_permission(resource_id, user_id, permission):resources.append(resource_id)return resourcesdef audit_log(self, resource_id: str, user_id: str, action: str, success: bool):"""审计日志"""log_entry = {"timestamp": datetime.now().isoformat(),"resource_id": resource_id,"user_id": user_id,"action": action,"success": success}# 实际应用中应写入日志系统print(f"[AUDIT] {log_entry}")# 使用示例access_manager = PromptAccessManager()# 注册用户access_manager.register_user(User(user_id="user001",name="Alice",role="editor",groups=["engineering"]))access_manager.register_user(User(user_id="user002",name="Bob",role="viewer",groups=["engineering"]))# 创建资源access_manager.create_resource("prompt_customer_service", "user001")# 授予权限access_manager.grant_permission("prompt_customer_service","user002",[Permission.READ, Permission.EXECUTE])# 检查权限can_write = access_manager.check_permission("prompt_customer_service", "user002", Permission.WRITE)print(f"Bob can write: {can_write}")# Falsecan_read = access_manager.check_permission("prompt_customer_service", "user002", Permission.READ)print(f"Bob can read: {can_read}")# True

18.3.3 提示词知识库

from typing import Dict, List, Optional, Setfrom dataclasses import dataclass, fieldimport re@dataclassclass KnowledgeEntry:"""知识条目"""entry_id: strtitle: strcontent: strcategory: strtags: List[str] = field(default_factory=list)related_prompts: List[str] = field(default_factory=list)created_by: str = ""created_at: datetime = field(default_factory=datetime.now)class PromptKnowledgeBase:"""提示词知识库"""def __init__(self):self.entries: Dict[str, KnowledgeEntry] = {}self.category_index: Dict[str, List[str]] = {}self.tag_index: Dict[str, List[str]] = {}self.search_index: Dict[str, Set[str]] = {}# 倒排索引def add_entry(self, entry: KnowledgeEntry):"""添加条目"""self.entries[entry.entry_id] = entry# 更新分类索引if entry.category not in self.category_index:self.category_index[entry.category] = []self.category_index[entry.category].append(entry.entry_id)# 更新标签索引for tag in entry.tags:if tag not in self.tag_index:self.tag_index[tag] = []self.tag_index[tag].append(entry.entry_id)# 更新搜索索引self._update_search_index(entry)def _update_search_index(self, entry: KnowledgeEntry):"""更新搜索索引"""# 分词并建立倒排索引text = f"{entry.title} {entry.content}"words = self._tokenize(text)for word in words:if word not in self.search_index:self.search_index[word] = set()self.search_index[word].add(entry.entry_id)def _tokenize(self, text: str) -> List[str]:"""分词（简化实现）"""# 实际应用中应使用专业的分词工具words = re.findall(r'bw+b', text.lower())return wordsdef search(self, query: str, category: str = None, tags: List[str] = None) -> List[KnowledgeEntry]:"""搜索"""query_words = self._tokenize(query)# 获取候选条目candidate_ids = Nonefor word in query_words:if word in self.search_index:if candidate_ids is None:candidate_ids = self.search_index[word].copy()else:candidate_ids &= self.search_index[word]if candidate_ids is None:candidate_ids = set(self.entries.keys())# 过滤results = []for entry_id in candidate_ids:entry = self.entries[entry_id]# 分类过滤if category and entry.category != category:continue# 标签过滤if tags and not all(tag in entry.tags for tag in tags):continueresults.append(entry)# 按相关性排序（简化：匹配词数）results.sort(key=lambda e: self._relevance_score(e, query_words), reverse=True)return resultsdef _relevance_score(self, entry: KnowledgeEntry, query_words: List[str]) -> int:"""计算相关性分数"""text = f"{entry.title} {entry.content}".lower()score = 0for word in query_words:score += text.count(word)return scoredef get_by_category(self, category: str) -> List[KnowledgeEntry]:"""按分类获取"""entry_ids = self.category_index.get(category, [])return [self.entries[eid] for eid in entry_ids]def get_by_tag(self, tag: str) -> List[KnowledgeEntry]:"""按标签获取"""entry_ids = self.tag_index.get(tag, [])return [self.entries[eid] for eid in entry_ids]def link_to_prompt(self, entry_id: str, prompt_id: str):"""关联到提示词"""if entry_id in self.entries:if prompt_id not in self.entries[entry_id].related_prompts:self.entries[entry_id].related_prompts.append(prompt_id)def get_related_knowledge(self, prompt_id: str) -> List[KnowledgeEntry]:"""获取与提示词相关的知识"""related = []for entry in self.entries.values():if prompt_id in entry.related_prompts:related.append(entry)return related# 使用示例kb = PromptKnowledgeBase()# 添加知识条目entry1 = KnowledgeEntry(entry_id="kb_001",title="Few-shot提示最佳实践",content="""Few-shot提示是通过提供示例来指导模型完成任务的技术。关键要点：1. 示例要具有代表性2. 示例数量适中（3-5个）3. 示例格式一致4. 涵盖边界情况""",category="techniques",tags=["few-shot", "best-practices", "prompting"],created_by="expert001")kb.add_entry(entry1)# 搜索results = kb.search("few shot examples")print(f"找到 {len(results)} 条相关知识")# 按分类获取techniques = kb.get_by_category("techniques")print(f"技术类知识: {len(techniques)} 条")# 关联到提示词kb.link_to_prompt("kb_001", "prompt_classification")

18.4 大模型应用开发框架：LangChain、LlamaIndex、Semantic Kernel

18.4.1 LangChain

LangChain是目前最流行的LLM应用开发框架，提供了完整的组件生态。

核心组件：

LANGCHAIN_COMPONENTS = """LangChain 核心组件：1. **Model I/O** - LLM：语言模型接口 - Chat Model：对话模型接口 - Prompts：提示词管理 - Output Parsers：输出解析2. **Chains** - LLMChain：基础链 - SequentialChain：顺序链 - RouterChain：路由链 - TransformChain：转换链3. **Agents** - Tool：工具定义 - Agent：智能体 - AgentExecutor：执行器4. **Memory** - ConversationBufferMemory：缓冲区记忆 - ConversationBufferWindowMemory：窗口记忆 - VectorStoreRetrieverMemory：向量检索记忆5. **Retrieval** - Document Loaders：文档加载器 - Text Splitters：文本分割器 - Embeddings：嵌入模型 - Vector Stores：向量存储 - Retrievers：检索器"""# LangChain 使用示例"""from langchain import OpenAI, LLMChain, PromptTemplatefrom langchain.memory import ConversationBufferMemoryfrom langchain.agents import initialize_agent, Tool# 1. 基础Chainllm = OpenAI(temperature=0.7)prompt = PromptTemplate(input_variables=["product"],template="为{product}写一段营销文案。")chain = LLMChain(llm=llm, prompt=prompt)result = chain.run("智能手表")# 2. 带记忆的Chainmemory = ConversationBufferMemory()conversation = LLMChain(llm=llm,prompt=PromptTemplate(input_variables=["history", "input"],template="历史：{history}n用户：{input}n助手："),memory=memory)# 3. Agentfrom langchain.tools import DuckDuckGoSearchRunsearch = DuckDuckGoSearchRun()tools = [Tool(name="Search",func=search.run,description="用于搜索最新信息")]agent = initialize_agent(tools, llm, agent="zero-shot-react-description")agent.run("今天北京的天气如何？")"""

LCEL（LangChain Expression Language）：

LCEL_CONCEPT = """LCEL 是 LangChain Expression Language 的缩写，是LangChain 0.1.0+版本引入的声明式链组合语法。核心概念：1. Runnable：可运行组件的统一接口2. Pipes (|)：组合操作符3. 内置组件：RunnableParallel, RunnablePassthrough等优势：- 简洁的语法- 自动的流式支持- 自动的异步支持- 优化的并行执行"""# LCEL 使用示例"""from langchain_core.runnables import RunnablePassthrough, RunnableParallelfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI# 基础链model = ChatOpenAI()prompt = ChatPromptTemplate.from_template("告诉我一个关于{topic}的笑话")chain = prompt | model# 并行执行chain = RunnableParallel(joke=prompt | model,topic=RunnablePassthrough())# 复杂链retrieval_chain = ({"context": retriever, "question": RunnablePassthrough()}| prompt| model)"""

18.4.2 LlamaIndex

LlamaIndex专注于数据索引和检索，是构建RAG应用的强大工具。

LLAMAINDEX_CONCEPT = """LlamaIndex 核心概念：1. **Indexing（索引）** - VectorStoreIndex：向量存储索引 - ListIndex：列表索引 - TreeIndex：树形索引 - KeywordTableIndex：关键词表索引2. **Querying（查询）** - Retriever：检索器 - Query Engine：查询引擎 - Chat Engine：对话引擎3. **Data Connectors（数据连接器）** - 支持多种数据源：文件、数据库、API等 - 自动数据加载和解析4. **Node Parser（节点解析器）** - 文档分块策略 - 元数据提取"""# LlamaIndex 使用示例"""from llama_index import VectorStoreIndex, SimpleDirectoryReader, ServiceContextfrom llama_index.llms import OpenAI# 加载文档documents = SimpleDirectoryReader('data').load_data()# 创建索引index = VectorStoreIndex.from_documents(documents)# 创建查询引擎query_engine = index.as_query_engine()# 查询response = query_engine.query("文档的主要内容是什么？")print(response)# 创建对话引擎chat_engine = index.as_chat_engine()response = chat_engine.chat("告诉我更多关于...")"""

LlamaIndex核心实现：

from typing import List, Dict, Anyfrom dataclasses import dataclassimport numpy as np@dataclassclass Document:"""文档"""text: strdoc_id: strmetadata: Dict[str, Any] = None@dataclassclass Node:"""节点（文档块）"""text: strnode_id: strdoc_id: strembedding: List[float] = Nonemetadata: Dict[str, Any] = Noneclass SimpleVectorStore:"""简单向量存储"""def __init__(self, embedding_dim: int = 1536):self.embedding_dim = embedding_dimself.nodes: Dict[str, Node] = {}self.embeddings: np.ndarray = Noneself.node_ids: List[str] = []def add_nodes(self, nodes: List[Node]):"""添加节点"""embeddings = []for node in nodes:self.nodes[node.node_id] = nodeself.node_ids.append(node.node_id)embeddings.append(node.embedding or [0.0] * self.embedding_dim)# 更新嵌入矩阵if self.embeddings is None:self.embeddings = np.array(embeddings)else:self.embeddings = np.vstack([self.embeddings, embeddings])def similarity_search(self, query_embedding: List[float],top_k: int = 5) -> List[Node]:"""相似度搜索"""if self.embeddings is None or len(self.node_ids) == 0:return []query_vec = np.array(query_embedding).reshape(1, -1)# 计算余弦相似度similarities = np.dot(self.embeddings, query_vec.T).flatten()# 获取top-ktop_indices = np.argsort(similarities)[-top_k:][::-1]return [self.nodes[self.node_ids[i]] for i in top_indices]class SimpleIndex:"""简单索引"""def __init__(self, embedding_fn=None):self.documents: Dict[str, Document] = {}self.nodes: List[Node] = []self.vector_store = SimpleVectorStore()self.embedding_fn = embedding_fn or self._default_embeddingdef _default_embedding(self, text: str) -> List[float]:"""默认嵌入函数（简化实现）"""# 实际应调用嵌入模型APIimport hashlibhash_val = hashlib.md5(text.encode()).hexdigest()# 生成伪随机向量np.random.seed(int(hash_val[:8], 16))return np.random.randn(1536).tolist()def add_documents(self, documents: List[Document]):"""添加文档"""for doc in documents:self.documents[doc.doc_id] = doc# 分块（简化：每个文档作为一个节点）node = Node(text=doc.text,node_id=f"node_{doc.doc_id}",doc_id=doc.doc_id,embedding=self.embedding_fn(doc.text),metadata=doc.metadata)self.nodes.append(node)# 添加到向量存储self.vector_store.add_nodes(self.nodes)def query(self, query_text: str, top_k: int = 5) -> List[Node]:"""查询"""query_embedding = self.embedding_fn(query_text)return self.vector_store.similarity_search(query_embedding, top_k)# 使用示例index = SimpleIndex()# 添加文档docs = [Document(doc_id="doc1",text="机器学习是人工智能的一个分支，它使计算机能够从数据中学习。",metadata={"category": "AI"}),Document(doc_id="doc2",text="深度学习是机器学习的一种方法，使用多层神经网络。",metadata={"category": "AI"})]index.add_documents(docs)# 查询results = index.query("什么是人工智能？", top_k=2)for node in results:print(f"相关文档: {node.text[:50]}...")

18.4.3 Semantic Kernel

Semantic Kernel是微软开发的AI开发SDK，支持多种编程语言。

SEMANTIC_KERNEL_CONCEPT = """Semantic Kernel 核心概念：1. **Kernel（内核）** - 核心编排引擎 - 管理AI服务和插件2. **Plugins（插件）** - 语义函数（Semantic Functions）：基于提示词 - 原生函数（Native Functions）：基于代码3. **Planner（规划器）** - 自动任务分解 - 动态计划生成4. **Memory（记忆）** - 语义记忆 - 上下文管理5. **Connectors（连接器）** - AI服务连接 - 内存存储连接"""# Semantic Kernel 使用示例（Python）"""import semantic_kernel as skfrom semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion# 创建Kernelkernel = sk.Kernel()# 添加AI服务api_key = "your-api-key"kernel.add_chat_service("gpt-4", OpenAIChatCompletion("gpt-4", api_key))# 定义语义函数（提示词）prompt = """将以下文本翻译成{{$target_language}}：{{$input}}"""translate = kernel.create_semantic_function(prompt,function_name="Translate",plugin_name="Translation")# 执行result = translate("Hello, world!", target_language="中文")print(result)# 定义原生函数from semantic_kernel.skill_definition import sk_functionclass MathPlugin:@sk_function(description="计算两个数的和",name="Add")def add(self, input: str) -> str:numbers = input.split(',')result = float(numbers[0]) + float(numbers[1])return str(result)# 注册插件math_plugin = kernel.import_skill(MathPlugin())# 使用Plannerfrom semantic_kernel.planning import BasicPlannerplanner = BasicPlanner()plan = await planner.create_plan_async("计算3+5并翻译成法语", kernel)result = await plan.invoke_async()"""

Semantic Kernel风格实现：

from typing import Dict, Callable, Any, Listfrom dataclasses import dataclassimport inspect@dataclassclass FunctionResult:"""函数执行结果"""value: strmetadata: Dict[str, Any] = Noneclass SemanticFunction:"""语义函数"""def __init__(self, name: str, plugin_name: str,template: str, llm_client):self.name = nameself.plugin_name = plugin_nameself.template = templateself.llm_client = llm_clientself.parameters = self._extract_parameters()def _extract_parameters(self) -> List[str]:"""提取模板参数"""import repattern = r'{{($w+)}}'matches = re.findall(pattern, self.template)return [m[1:] for m in matches]# 去掉$前缀def invoke(self, **kwargs) -> FunctionResult:"""执行函数"""# 填充模板prompt = self.templatefor param in self.parameters:value = kwargs.get(param, "")prompt = prompt.replace(f"{{{{${param}}}}}", str(value))# 调用LLMresponse = self.llm_client.generate(prompt)return FunctionResult(value=response,metadata={"function": f"{self.plugin_name}.{self.name}"})class NativeFunction:"""原生函数"""def __init__(self, name: str, plugin_name: str,func: Callable, description: str):self.name = nameself.plugin_name = plugin_nameself.func = funcself.description = descriptiondef invoke(self, **kwargs) -> FunctionResult:"""执行函数"""result = self.func(**kwargs)return FunctionResult(value=str(result),metadata={"function": f"{self.plugin_name}.{self.name}"})class SimpleKernel:"""简化版Semantic Kernel"""def __init__(self):self.semantic_functions: Dict[str, SemanticFunction] = {}self.native_functions: Dict[str, NativeFunction] = {}self.llm_client = Nonedef set_llm_client(self, client):"""设置LLM客户端"""self.llm_client = clientdef register_semantic_function(self, name: str, plugin_name: str,template: str):"""注册语义函数"""func = SemanticFunction(name, plugin_name, template, self.llm_client)key = f"{plugin_name}.{name}"self.semantic_functions[key] = funcdef register_native_function(self, name: str, plugin_name: str,func: Callable, description: str = ""):"""注册原生函数"""native_func = NativeFunction(name, plugin_name, func, description)key = f"{plugin_name}.{name}"self.native_functions[key] = native_funcdef invoke(self, function_name: str, **kwargs) -> FunctionResult:"""调用函数"""if function_name in self.semantic_functions:return self.semantic_functions[function_name].invoke(**kwargs)elif function_name in self.native_functions:return self.native_functions[function_name].invoke(**kwargs)else:raise ValueError(f"Function {function_name} not found")# 使用示例def mock_llm():class MockLLM:def generate(self, prompt: str) -> str:return f"[LLM响应] {prompt[:50]}..."return MockLLM()kernel = SimpleKernel()kernel.set_llm_client(mock_llm())# 注册语义函数kernel.register_semantic_function(name="Summarize",plugin_name="TextProcessing",template="请为以下文本生成摘要：n{{$input}}nn摘要：")# 注册原生函数def calculate_length(text: str) -> int:return len(text)kernel.register_native_function(name="Length",plugin_name="TextProcessing",func=calculate_length,description="计算文本长度")# 调用result = kernel.invoke("TextProcessing.Summarize", input="这是一段长文本...")print(f"语义函数结果: {result.value}")result = kernel.invoke("TextProcessing.Length", text="Hello")print(f"原生函数结果: {result.value}")

18.5 提示词工程的 CI/CD 与自动化部署

18.5.1 LLMOps概述

LLMOPS_CONCEPT = """LLMOps（Large Language Model Operations）是MLOps在LLM领域的延伸，专注于大语言模型应用的全生命周期管理。LLMOps 核心流程：1. **开发（Develop）** - 提示词工程 - 原型开发 - 本地测试2. **版本控制（Version）** - 提示词版本管理 - 数据版本管理 - 模型版本管理3. **测试（Test）** - 单元测试 - 集成测试 - 评估测试4. **部署（Deploy）** - 环境管理 - 灰度发布 - A/B测试5. **监控（Monitor）** - 性能监控 - 质量监控 - 成本监控6. **反馈（Feedback）** - 用户反馈收集 - 模型改进 - 持续优化"""

18.5.2 CI/CD流水线设计

from typing import List, Dict, Callable, Anyfrom dataclasses import dataclassfrom enum import Enumimport timeclass PipelineStage(Enum):"""流水线阶段"""LINT = "lint"TEST = "test"EVALUATE = "evaluate"BUILD = "build"DEPLOY = "deploy"@dataclassclass PipelineResult:"""流水线结果"""stage: strsuccess: boolduration_ms: floatoutput: strerror: str = Noneclass PromptCIPipeline:"""提示词CI流水线"""def __init__(self):self.stages: Dict[PipelineStage, List[Callable]] = {stage: [] for stage in PipelineStage}self.results: List[PipelineResult] = []def add_stage(self, stage: PipelineStage, func: Callable):"""添加阶段任务"""self.stages[stage].append(func)def run(self, context: Dict[str, Any]) -> List[PipelineResult]:"""运行流水线"""self.results = []for stage in PipelineStage:print(f"Running stage: {stage.value}")for func in self.stages[stage]:start_time = time.time()try:output = func(context)duration = (time.time() - start_time) * 1000result = PipelineResult(stage=stage.value,success=True,duration_ms=duration,output=str(output))except Exception as e:duration = (time.time() - start_time) * 1000result = PipelineResult(stage=stage.value,success=False,duration_ms=duration,output="",error=str(e))self.results.append(result)print(f"Stage {stage.value} failed: {e}")return self.results# 失败时停止self.results.append(result)return self.resultsdef generate_report(self) -> Dict:"""生成报告"""total_duration = sum(r.duration_ms for r in self.results)success_count = sum(1 for r in self.results if r.success)return {"total_stages": len(self.results),"success_count": success_count,"failure_count": len(self.results) - success_count,"total_duration_ms": total_duration,"stages": [{"stage": r.stage,"success": r.success,"duration_ms": r.duration_ms,"error": r.error}for r in self.results]}# 使用示例pipeline = PromptCIPipeline()# 添加Lint阶段def lint_prompts(context):"""检查提示词格式"""prompt = context.get("prompt", "")if not prompt:raise ValueError("Empty prompt")if "{{" in prompt and "}}" not in prompt:raise ValueError("Unclosed template variable")return "Lint passed"pipeline.add_stage(PipelineStage.LINT, lint_prompts)# 添加测试阶段def run_unit_tests(context):"""运行单元测试"""# 模拟测试return "All tests passed"pipeline.add_stage(PipelineStage.TEST, run_unit_tests)# 添加评估阶段def evaluate_prompt(context):"""评估提示词质量"""# 模拟评估return "Quality score: 0.92"pipeline.add_stage(PipelineStage.EVALUATE, evaluate_prompt)# 运行流水线context = {"prompt": "你好，{{name}}"}results = pipeline.run(context)report = pipeline.generate_report()print(f"流水线完成: {report['success_count']}/{report['total_stages']} 阶段成功")

18.5.3 自动化部署系统

from typing import Dict, List, Optionalfrom dataclasses import dataclassfrom datetime import datetimeimport json@dataclassclass DeploymentConfig:"""部署配置"""environment: str# dev, staging, productionprompt_id: strversion: strtraffic_percentage: float = 100.0rollback_on_error: bool = Truehealth_check_url: str = Noneclass PromptDeploymentManager:"""提示词部署管理器"""def __init__(self):self.deployments: Dict[str, List[Dict]] = {}# env -> deploymentsself.current_versions: Dict[str, Dict[str, str]] = {}# env -> {prompt_id: version}def deploy(self, config: DeploymentConfig) -> Dict:"""部署提示词"""deployment_id = f"deploy_{datetime.now().strftime('%Y%m%d%H%M%S')}"deployment = {"id": deployment_id,"config": config,"status": "in_progress","started_at": datetime.now(),"completed_at": None,"error": None}if config.environment not in self.deployments:self.deployments[config.environment] = []self.deployments[config.environment].append(deployment)try:# 1. 验证配置self._validate_config(config)# 2. 健康检查if config.health_check_url:self._health_check(config.health_check_url)# 3. 执行部署self._execute_deployment(config)# 4. 更新当前版本if config.environment not in self.current_versions:self.current_versions[config.environment] = {}self.current_versions[config.environment][config.prompt_id] = config.versiondeployment["status"] = "completed"deployment["completed_at"] = datetime.now()except Exception as e:deployment["status"] = "failed"deployment["error"] = str(e)if config.rollback_on_error:self._rollback(config)return deploymentdef _validate_config(self, config: DeploymentConfig):"""验证配置"""if config.traffic_percentage < 0 or config.traffic_percentage > 100:raise ValueError("Traffic percentage must be between 0 and 100")if config.environment not in ["dev", "staging", "production"]:raise ValueError("Invalid environment")def _health_check(self, url: str):"""健康检查"""# 模拟健康检查import randomif random.random() < 0.1:# 10%失败率模拟raise Exception("Health check failed")def _execute_deployment(self, config: DeploymentConfig):"""执行部署"""# 模拟部署过程print(f"Deploying {config.prompt_id}:{config.version} to {config.environment}")time.sleep(1)# 模拟部署时间def _rollback(self, config: DeploymentConfig):"""回滚部署"""print(f"Rolling back {config.prompt_id} in {config.environment}")# 回滚逻辑def get_deployment_history(self, environment: str) -> List[Dict]:"""获取部署历史"""return self.deployments.get(environment, [])def get_current_version(self, environment: str, prompt_id: str) -> Optional[str]:"""获取当前版本"""return self.current_versions.get(environment, {}).get(prompt_id)def promote(self, prompt_id: str, from_env: str, to_env: str):"""提升部署（从低级环境到高级环境）"""current_version = self.get_current_version(from_env, prompt_id)if not current_version:raise ValueError(f"No version found in {from_env}")config = DeploymentConfig(environment=to_env,prompt_id=prompt_id,version=current_version)return self.deploy(config)# 使用示例deploy_manager = PromptDeploymentManager()# 部署到开发环境dev_config = DeploymentConfig(environment="dev",prompt_id="customer_service",version="v1.2.0")dev_result = deploy_manager.deploy(dev_config)print(f"开发环境部署: {dev_result['status']}")# 部署到生产环境（灰度）prod_config = DeploymentConfig(environment="production",prompt_id="customer_service",version="v1.2.0",traffic_percentage=10# 10%流量)prod_result = deploy_manager.deploy(prod_config)print(f"生产环境部署: {prod_result['status']}")# 提升部署promote_result = deploy_manager.promote("customer_service", "dev", "staging")print(f"提升到Staging: {promote_result['status']}")

18.5.4 配置即代码

"""提示词工程的配置即代码（Configuration as Code）实践"""# prompt_config.yaml 示例PROMPT_CONFIG_YAML = """prompts:customer_service:name: "Customer Service Assistant"description: "Handles customer inquiries"versions:v1.0.0:template: |You are a helpful customer service assistant.Customer query: {{query}}Please provide a helpful response.variables:- queryparameters:temperature: 0.7max_tokens: 500v1.1.0:template: |You are a professional customer service assistant for {{company}}.Customer query: {{query}}Customer history: {{history}}Please provide a helpful and personalized response.variables:- company- query- historyparameters:temperature: 0.5max_tokens: 800evaluation:test_cases:- name: "basic_query"inputs:query: "How do I reset my password?"expected_behavior: "Provides password reset instructions"metrics:- helpfulness- accuracy- response_timedeployment:environments:dev:model: "gpt-3.5-turbo"staging:model: "gpt-4"production:model: "gpt-4"traffic_split:v1.0.0: 90v1.1.0: 10"""class PromptConfigManager:"""提示词配置管理器"""def __init__(self, config_path: str):self.config_path = config_pathself.config = self._load_config()def _load_config(self) -> Dict:"""加载配置"""# 实际应解析YAML文件import yamlwith open(self.config_path, 'r') as f:return yaml.safe_load(f)def get_prompt_config(self, prompt_id: str, version: str = None) -> Dict:"""获取提示词配置"""prompt_config = self.config.get("prompts", {}).get(prompt_id)if not prompt_config:return Noneif version:return prompt_config.get("versions", {}).get(version)# 返回最新版本versions = prompt_config.get("versions", {})if versions:latest = sorted(versions.keys())[-1]return versions[latest]return Nonedef get_evaluation_config(self, prompt_id: str) -> Dict:"""获取评估配置"""prompt_config = self.config.get("prompts", {}).get(prompt_id)return prompt_config.get("evaluation", {}) if prompt_config else {}def get_deployment_config(self, prompt_id: str, environment: str) -> Dict:"""获取部署配置"""prompt_config = self.config.get("prompts", {}).get(prompt_id)if not prompt_config:return Nonedeployment = prompt_config.get("deployment", {})environments = deployment.get("environments", {})return environments.get(environment)def validate_config(self) -> List[str]:"""验证配置"""errors = []prompts = self.config.get("prompts", {})for prompt_id, prompt_config in prompts.items():# 验证版本versions = prompt_config.get("versions", {})if not versions:errors.append(f"{prompt_id}: No versions defined")for version, version_config in versions.items():if "template" not in version_config:errors.append(f"{prompt_id}:{version}: Missing template")# 验证变量template = version_config.get("template", "")declared_vars = set(version_config.get("variables", []))import reused_vars = set(re.findall(r'{{(w+)}}', template))undeclared = used_vars - declared_varsif undeclared:errors.append(f"{prompt_id}:{version}: Undeclared variables: {undeclared}")return errors# 使用示例# config_manager = PromptConfigManager("prompt_config.yaml")# errors = config_manager.validate_config()# if errors:# print("配置错误:")# for error in errors:# print(f"- {error}")

本章小结

本章全面介绍了提示词工程工具链：

1. 提示词开发工具：学习了LangChain Prompt Hub的使用方法，了解了DSPy框架的核心理念和优化方法，掌握了提示词模板化和参数化技术。

2. 测试与调试工具：掌握了LangSmith的追踪和调试功能，学习了PromptEval测试框架的构建方法，能够建立完整的提示词测试体系。

3. 提示词管理工具：实现了提示词版本控制系统、权限管理系统和知识库，能够支撑企业级的提示词管理需求。

4. 大模型应用开发框架：深入了解了LangChain、LlamaIndex和Semantic Kernel三大框架的核心概念和使用方法，能够根据场景选择合适的框架。

5. CI/CD与自动化部署：掌握了LLMOps的核心理念，能够设计提示词CI/CD流水线，实现提示词的自动化测试和部署。

通过构建完整的工具链，可以大幅提升提示词工程的效率和质量，实现提示词开发的标准化和工程化。

参考资源

1. LangChain Documentation: python.langchain.com/
2. LangSmith Documentation: docs.smith.langchain.com/
3. LlamaIndex Documentation: docs.llamaindex.ai/
4. Semantic Kernel Documentation: learn.microsoft.com/en-us/seman…
5. DSPy Documentation: dspy-docs.vercel.app/
6. Prompt Flow: microsoft.github.io/promptflow/
7. Weights & Biases Prompts: docs.wandb.ai/guides/prom…
8. MLflow LLM Tracking: mlflow.org/docs/latest…