LangChain Prompt管理核心:PromptTemplate与ChatPromptTemplate全解析
LangChain Prompt管理核心:PromptTemplate与ChatPromptTemplate全解析
1. 核心定义与价值
1.1 本质定义
PromptTemplate:通用文本提示词的"动态构建工具",是LangChain中最基础的模板类,专门用于解决硬编码问题。它继承自StringPromptTemplate
,提供了灵活的字符串模板管理能力。
ChatPromptTemplate:对话场景专属模板,基于"角色-消息"结构设计,继承自BaseChatPromptTemplate
。它专门为Chat模型优化,支持多轮对话和复杂的消息组合。
1.2 无模板的痛点对比
硬编码方式的缺陷:
# ❌ 硬编码方式 - 维护困难
def generate_product_description_bad(product_name, features, price):return f"产品名称:{product_name}\n特性:{features}\n价格:{price}\n请为这个产品写一个吸引人的描述。"# 问题:
# 1. 模板分散在代码各处,难以统一管理
# 2. 修改模板需要修改代码,增加出错风险
# 3. 无法复用,每个场景都要重写
# 4. 缺乏参数验证,容易出现运行时错误
使用PromptTemplate的优势:
# ✅ 模板化方式 - 灵活可维护
from langchain_core.prompts import PromptTemplateproduct_template = PromptTemplate.from_template("产品名称:{product_name}\n""特性:{features}\n""价格:{price}\n""请为这个产品写一个吸引人的描述。"
)# 优势:
# 1. 模板集中管理,易于维护
# 2. 参数自动验证,减少错误
# 3. 支持复用和组合
# 4. 与LangChain生态无缝集成
1.3 适用场景区分
场景类型 | PromptTemplate | ChatPromptTemplate |
---|---|---|
简单文本生成 | ✅ 适用 | ❌ 过度设计 |
多轮对话 | ❌ 功能不足 | ✅ 专门优化 |
角色扮演 | ❌ 缺乏角色概念 | ✅ 原生支持 |
文档处理 | ✅ 简单直接 | ❌ 不必要 |
客服系统 | ❌ 缺乏上下文 | ✅ 完美匹配 |
2. 实现逻辑
2.1 共性逻辑:模板定义→参数注入→结果输出
两个模板类都遵循相同的核心流程:
# 共同的处理流程
class BasePromptTemplate:def invoke(self, input: dict) -> PromptValue:"""统一的调用入口"""# 1. 验证输入参数validated_input = self._validate_input(input)# 2. 格式化模板return self.format_prompt(**validated_input)
核心步骤解析:
- 模板定义:使用特定语法定义变量占位符
- 参数验证:检查必需参数是否提供,类型是否匹配
- 变量合并:合并部分变量(partial_variables)和用户输入
- 格式化输出:根据模板类型生成相应的输出格式
2.2 差异逻辑:format() vs format_messages()
PromptTemplate的format()方法:
def format(self, **kwargs: Any) -> str:"""返回格式化的字符串"""kwargs = self._merge_partial_and_user_variables(**kwargs)return DEFAULT_FORMATTER_MAPPING[self.template_format](self.template, **kwargs)
ChatPromptTemplate的format_messages()方法:
def format_messages(self, **kwargs: Any) -> list[BaseMessage]:"""返回格式化的消息列表"""kwargs = self._merge_partial_and_user_variables(**kwargs)result = []for message_template in self.messages:if isinstance(message_template, BaseMessage):result.extend([message_template])elif isinstance(message_template, BaseMessagePromptTemplate):message = message_template.format_messages(**kwargs)result.extend(message)return result
2.3 关键特性
参数校验机制:
def _validate_input(self, inner_input: Any) -> dict:"""输入验证逻辑"""# 单变量简化处理if not isinstance(inner_input, dict):if len(self.input_variables) == 1:var_name = self.input_variables[0]inner_input = {var_name: inner_input}# 检查缺失变量missing = set(self.input_variables).difference(inner_input)if missing:raise KeyError(f"缺少必需的变量: {missing}")return inner_input
部分渲染(partial_variables):
def partial(self, **kwargs: Any) -> BasePromptTemplate:"""创建部分填充的模板副本"""prompt_dict = self.__dict__.copy()prompt_dict["input_variables"] = list(set(self.input_variables).difference(kwargs))prompt_dict["partial_variables"] = {**self.partial_variables, **kwargs}return type(self)(**prompt_dict)
3. 代码实践
3.1 基础实践1:PromptTemplate实现多场景复用
"""
依赖安装:
pip install langchain-core
"""from langchain_core.prompts import PromptTemplate
from typing import Dict, Anyclass ProductDescriptionGenerator:"""商品描述生成器 - 展示PromptTemplate的复用能力"""def __init__(self):# 基础模板self.base_template = PromptTemplate.from_template("产品:{product_name}\n""类别:{category}\n""特点:{features}\n""目标用户:{target_audience}\n""\n请为这个{category}产品写一个{style}的描述,突出其{features}特点。")# 营销模板(继承基础模板并添加营销元素)self.marketing_template = PromptTemplate.from_template("🔥 限时优惠!\n\n""产品:{product_name}\n""原价:{original_price}\n""现价:{current_price}\n""特点:{features}\n""\n写一个有紧迫感的营销文案,强调价格优势和{features}。")# 技术规格模板self.tech_spec_template = PromptTemplate.from_template("产品技术规格说明\n""================\n""产品名称:{product_name}\n""技术参数:{tech_specs}\n""兼容性:{compatibility}\n""使用场景:{use_cases}\n""\n请写一个专业的技术说明,面向{target_audience}用户。")def generate_basic_description(self, **kwargs) -> str:"""生成基础产品描述"""try:return self.base_template.format(**kwargs)except KeyError as e:return f"错误:缺少必需参数 {e}"def generate_marketing_copy(self, **kwargs) -> str:"""生成营销文案"""return self.marketing_template.format(**kwargs)def generate_tech_description(self, **kwargs) -> str:"""生成技术说明"""return self.tech_spec_template.format(**kwargs)def batch_generate(self, products: list[Dict[str, Any]]) -> Dict[str, list[str]]:"""批量生成多种描述"""results = {"basic": [], "marketing": [], "tech": []}for product in products:# 基础描述if all(key in product for key in ["product_name", "category", "features", "target_audience", "style"]):results["basic"].append(self.generate_basic_description(**product))# 营销文案if all(key in product for key in ["product_name", "original_price", "current_price", "features"]):results["marketing"].append(self.generate_marketing_copy(**product))# 技术说明if all(key in product for key in ["product_name", "tech_specs", "compatibility", "use_cases", "target_audience"]):results["tech"].append(self.generate_tech_description(**product))return results# 使用示例
if __name__ == "__main__":generator = ProductDescriptionGenerator()# 单个产品示例product_data = {"product_name": "智能蓝牙耳机 Pro","category": "数码产品","features": "主动降噪、30小时续航、快速充电","target_audience": "商务人士","style": "专业简洁"}basic_desc = generator.generate_basic_description(**product_data)print("基础描述:")print(basic_desc)print("\n" + "="*50 + "\n")# 营销文案示例marketing_data = {"product_name": "智能蓝牙耳机 Pro","original_price": "¥599","current_price": "¥399","features": "主动降噪和超长续航"}marketing_copy = generator.generate_marketing_copy(**marketing_data)print("营销文案:")print(marketing_copy)# 预期输出:# 基础描述:# 产品:智能蓝牙耳机 Pro# 类别:数码产品# 特点:主动降噪、30小时续航、快速充电# 目标用户:商务人士# # 请为这个数码产品产品写一个专业简洁的描述,突出其主动降噪、30小时续航、快速充电特点。
3.2 基础实践2:ChatPromptTemplate实现多轮对话
"""
依赖安装:
pip install langchain-core
"""from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing import List, Dict, Anyclass CustomerServiceChatbot:"""客服聊天机器人 - 展示ChatPromptTemplate的对话能力"""def __init__(self):# 基础客服模板self.service_template = ChatPromptTemplate.from_messages([("system", "你是{company_name}的专业客服代表。你的任务是:\n""1. 友好、专业地回答客户问题\n""2. 了解客户的{service_type}需求\n""3. 提供准确的产品信息\n""4. 如果无法解决问题,引导客户联系人工客服\n""请保持{tone}的语调。"),MessagesPlaceholder(variable_name="chat_history"),("human", "{user_input}")])# 技术支持模板self.tech_support_template = ChatPromptTemplate.from_messages([("system", "你是{company_name}的技术支持专家。客户的设备信息:\n""- 产品型号:{product_model}\n""- 操作系统:{os_version}\n""- 问题类型:{issue_type}\n\n""请提供专业的技术解决方案,使用简单易懂的语言。"),MessagesPlaceholder(variable_name="chat_history"),("human", "问题描述:{problem_description}")])# 销售咨询模板self.sales_template = ChatPromptTemplate.from_messages([("system", "你是{company_name}的销售顾问。客户信息:\n""- 预算范围:{budget_range}\n""- 使用场景:{use_case}\n""- 关注重点:{key_concerns}\n\n""请推荐最适合的产品,并说明理由。"),("ai", "您好!我是{company_name}的销售顾问,很高兴为您服务。根据您的需求,我来为您推荐最合适的产品。"),MessagesPlaceholder(variable_name="chat_history"),("human", "{customer_query}")])def create_service_conversation(self, company_name: str,service_type: str,tone: str,chat_history: List[tuple],user_input: str) -> ChatPromptTemplate:"""创建客服对话"""# 转换聊天历史为消息对象history_messages = []for role, content in chat_history:if role == "human":history_messages.append(HumanMessage(content=content))elif role == "ai":history_messages.append(AIMessage(content=content))return self.service_template.format_messages(company_name=company_name,service_type=service_type,tone=tone,chat_history=history_messages,user_input=user_input)def create_tech_support_conversation(self,company_name: str,product_model: str,os_version: str,issue_type: str,chat_history: List[tuple],problem_description: str) -> List:"""创建技术支持对话"""history_messages = []for role, content in chat_history:if role == "human":history_messages.append(HumanMessage(content=content))elif role == "ai":history_messages.append(AIMessage(content=content))return self.tech_support_template.format_messages(company_name=company_name,product_model=product_model,os_version=os_version,issue_type=issue_type,chat_history=history_messages,problem_description=problem_description)def create_sales_conversation(self,company_name: str,budget_range: str,use_case: str,key_concerns: str,chat_history: List[tuple],customer_query: str) -> List:"""创建销售咨询对话"""history_messages = []for role, content in chat_history:if role == "human":history_messages.append(HumanMessage(content=content))elif role == "ai":history_messages.append(AIMessage(content=content))return self.sales_template.format_messages(company_name=company_name,budget_range=budget_range,use_case=use_case,key_concerns=key_concerns,chat_history=history_messages,customer_query=customer_query)def simulate_conversation_flow(self) -> None:"""模拟完整的对话流程"""print("=== 客服对话模拟 ===\n")# 模拟客服对话chat_history = [("human", "你好,我想了解一下你们的产品"),("ai", "您好!欢迎咨询我们的产品。请问您对哪类产品感兴趣呢?"),("human", "我想买一个笔记本电脑,主要用于办公")]messages = self.create_service_conversation(company_name="TechCorp科技",service_type="产品咨询",tone="友好专业",chat_history=chat_history,user_input="预算在8000元左右,需要轻薄便携")print("客服对话消息:")for i, msg in enumerate(messages):print(f"{i+1}. [{msg.__class__.__name__}] {msg.content}")print("\n" + "="*50 + "\n")# 模拟技术支持对话tech_history = [("human", "我的电脑开机很慢"),("ai", "我来帮您诊断这个问题。请问您的电脑使用多长时间了?")]tech_messages = self.create_tech_support_conversation(company_name="TechCorp科技",product_model="ThinkPad X1 Carbon",os_version="Windows 11",issue_type="性能问题",chat_history=tech_history,problem_description="开机需要3-4分钟,运行程序也很卡顿")print("技术支持对话消息:")for i, msg in enumerate(tech_messages):print(f"{i+1}. [{msg.__class__.__name__}] {msg.content}")# 使用示例
if __name__ == "__main__":chatbot = CustomerServiceChatbot()chatbot.simulate_conversation_flow()# 预期输出:# === 客服对话模拟 ===# # 客服对话消息:# 1. [SystemMessage] 你是TechCorp科技的专业客服代表。你的任务是:...# 2. [HumanMessage] 你好,我想了解一下你们的产品# 3. [AIMessage] 您好!欢迎咨询我们的产品。请问您对哪类产品感兴趣呢?# 4. [HumanMessage] 我想买一个笔记本电脑,主要用于办公# 5. [HumanMessage] 预算在8000元左右,需要轻薄便携
3.3 进阶实践:与LangChain其他组件结合
"""
依赖安装:
pip install langchain-core langchain-openai
"""from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParser, PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List, Dict, Any
import jsonclass ProductAnalysis(BaseModel):"""产品分析结果的结构化输出"""product_name: str = Field(description="产品名称")strengths: List[str] = Field(description="产品优势列表")weaknesses: List[str] = Field(description="产品劣势列表")target_market: str = Field(description="目标市场")price_range: str = Field(description="价格区间")recommendation: str = Field(description="推荐建议")class AdvancedPromptChain:"""高级Prompt链 - 展示与Runnable的集成"""def __init__(self):# 设置输出解析器self.analysis_parser = PydanticOutputParser(pydantic_object=ProductAnalysis)# 产品信息提取模板self.info_extraction_template = PromptTemplate.from_template("从以下产品描述中提取关键信息:\n\n""产品描述:{product_description}\n\n""请提取以下信息:\n""1. 产品名称\n""2. 主要特性\n""3. 价格信息\n""4. 目标用户\n\n""以JSON格式输出结果。")# 竞品分析模板self.competitor_analysis_template = ChatPromptTemplate.from_messages([("system", "你是一个专业的市场分析师,擅长产品竞争分析。"),("human", "请分析以下产品的竞争优势:\n\n""产品信息:{product_info}\n""竞品信息:{competitor_info}\n\n""请从以下角度分析:\n""1. 功能对比\n""2. 价格优势\n""3. 市场定位\n""4. 用户体验")])# 结构化分析模板self.structured_analysis_template = PromptTemplate.from_template("基于以下信息,生成结构化的产品分析报告:\n\n""产品基础信息:{basic_info}\n""竞争分析:{competition_analysis}\n\n""{format_instructions}\n\n""请确保分析客观、准确,并提供实用的建议。")# 添加格式说明self.structured_analysis_template = self.structured_analysis_template.partial(format_instructions=self.analysis_parser.get_format_instructions())def create_analysis_chain(self):"""创建分析链 - 展示Runnable组合"""# 并行处理链parallel_chain = RunnableParallel({"basic_info": self.info_extraction_template | self._mock_llm | StrOutputParser(),"competition_analysis": self.competitor_analysis_template | self._mock_llm | StrOutputParser(),"original_input": RunnablePassthrough()})# 最终分析链final_chain = (parallel_chain| RunnablePassthrough.assign(structured_analysis=lambda x: self.structured_analysis_template.format(basic_info=x["basic_info"],competition_analysis=x["competition_analysis"]))| (lambda x: x["structured_analysis"])| self._mock_llm| self.analysis_parser)return final_chaindef _mock_llm(self, prompt):"""模拟LLM响应 - 实际使用时替换为真实的LLM"""if isinstance(prompt, str):if "JSON格式" in prompt:return json.dumps({"product_name": "智能手表 Pro","features": ["健康监测", "GPS定位", "长续航"],"price": "¥2999","target_users": "运动爱好者"}, ensure_ascii=False)elif "竞争优势" in prompt:return "该产品在健康监测功能上领先竞品,电池续航能力突出,但价格相对较高。"else:# 返回结构化分析结果return json.dumps({"product_name": "智能手表 Pro","strengths": ["先进的健康监测", "超长续航", "精准GPS"],"weaknesses": ["价格偏高", "应用生态有限"],"target_market": "高端运动市场","price_range": "2500-3500元","recommendation": "适合对健康监测有高要求的运动爱好者"}, ensure_ascii=False)return "模拟响应"def demonstrate_chain_usage(self):"""演示链的使用"""print("=== 高级Prompt链演示 ===\n")# 输入数据input_data = {"product_description": "智能手表 Pro,配备先进的健康监测传感器,支持GPS定位,续航可达7天,售价2999元,主要面向运动爱好者。","competitor_info": "市面上同类产品价格在2000-4000元之间,主要竞品包括Apple Watch、华为Watch等。"}# 创建并执行链chain = self.create_analysis_chain()try:# 模拟链执行(实际使用时会调用真实LLM)print("正在执行分析链...")# 手动模拟链的执行过程basic_info = self.info_extraction_template.format(product_description=input_data["product_description"])print(f"1. 信息提取模板:\n{basic_info}\n")competition_prompt = self.competitor_analysis_template.format_messages(product_info=input_data["product_description"],competitor_info=input_data["competitor_info"])print(f"2. 竞品分析模板消息数量:{len(competition_prompt)}")print(f" 系统消息:{competition_prompt[0].content}")print(f" 用户消息:{competition_prompt[1].content[:100]}...\n")# 模拟最终结构化输出mock_result = ProductAnalysis(product_name="智能手表 Pro",strengths=["先进的健康监测", "超长续航", "精准GPS"],weaknesses=["价格偏高", "应用生态有限"],target_market="高端运动市场",price_range="2500-3500元",recommendation="适合对健康监测有高要求的运动爱好者")print("3. 结构化分析结果:")print(f" 产品名称:{mock_result.product_name}")print(f" 优势:{', '.join(mock_result.strengths)}")print(f" 劣势:{', '.join(mock_result.weaknesses)}")print(f" 目标市场:{mock_result.target_market}")print(f" 价格区间:{mock_result.price_range}")print(f" 推荐建议:{mock_result.recommendation}")except Exception as e:print(f"执行过程中出现错误:{e}")def demonstrate_prompt_composition(self):"""演示Prompt组合技巧"""print("\n=== Prompt组合技巧演示 ===\n")# 基础模板base_template = PromptTemplate.from_template("分析产品:{product}")# 详细模板detailed_template = PromptTemplate.from_template("请从{aspect}角度详细分析:{analysis_target}")# 模板组合combined_template = base_template + detailed_templateresult = combined_template.format(product="智能手机",aspect="用户体验",analysis_target="界面设计和操作流程")print("组合模板结果:")print(result)# Chat模板组合chat_base = ChatPromptTemplate.from_messages([("system", "你是产品分析专家")])chat_extended = chat_base + [("human", "请分析{product}的{aspect}")]chat_messages = chat_extended.format_messages(product="智能手机",aspect="市场竞争力")print(f"\nChat模板组合结果({len(chat_messages)}条消息):")for i, msg in enumerate(chat_messages):print(f"{i+1}. [{msg.__class__.__name__}] {msg.content}")# 使用示例
if __name__ == "__main__":advanced_chain = AdvancedPromptChain()advanced_chain.demonstrate_chain_usage()advanced_chain.demonstrate_prompt_composition()
4. 设计考量
4.1 核心目标:解耦、复用、标准化
解耦设计:
# PromptTemplate将模板逻辑与业务逻辑分离
class BusinessLogic:def __init__(self):# 模板与业务逻辑解耦self.email_template = PromptTemplate.from_template("尊敬的{customer_name},\n\n""感谢您购买{product_name}。\n""订单号:{order_id}\n""预计{delivery_date}送达。\n\n""如有问题请联系客服。")def send_order_confirmation(self, order_data: dict):# 业务逻辑专注于数据处理email_content = self.email_template.format(**order_data)# 发送邮件的具体实现...return email_content
复用机制:
# 基础模板可以被多个场景复用
base_analysis_template = PromptTemplate.from_template("请分析{target}的{aspect},重点关注{focus_points}。"
)# 场景1:产品分析
product_analysis = base_analysis_template.partial(aspect="市场表现",focus_points="用户反馈和销售数据"
)# 场景2:竞品分析
competitor_analysis = base_analysis_template.partial(aspect="竞争优势",focus_points="功能对比和价格策略"
)
标准化接口:
# 所有Prompt模板都实现统一的接口
class UnifiedPromptInterface:def format_for_llm(self, template, **kwargs):"""统一的LLM格式化接口"""if isinstance(template, PromptTemplate):return template.format(**kwargs)elif isinstance(template, ChatPromptTemplate):messages = template.format_messages(**kwargs)return self._messages_to_string(messages)else:raise ValueError(f"不支持的模板类型: {type(template)}")
4.2 适配LLM生态:普通LLM vs Chat模型
普通LLM适配:
# PromptTemplate专为文本补全模型设计
class TextCompletionAdapter:def __init__(self, llm):self.llm = llmself.prompt_template = PromptTemplate.from_template("Context: {context}\n""Question: {question}\n""Answer:")def generate(self, context: str, question: str) -> str:# 直接输出字符串,适合文本补全prompt = self.prompt_template.format(context=context, question=question)return self.llm.invoke(prompt)
Chat模型适配:
# ChatPromptTemplate专为对话模型优化
class ChatModelAdapter:def __init__(self, chat_model):self.chat_model = chat_modelself.chat_template = ChatPromptTemplate.from_messages([("system", "你是一个专业的问答助手。"),("human", "背景信息:{context}"),("human", "问题:{question}")])def generate(self, context: str, question: str) -> str:# 输出消息列表,适合对话模型messages = self.chat_template.format_messages(context=context,question=question)return self.chat_model.invoke(messages)
4.3 与其他框架对比
LangChain vs LlamaIndex:
# LangChain方式 - 更灵活的模板系统
langchain_template = ChatPromptTemplate.from_messages([("system", "你是{role},专长是{expertise}。"),MessagesPlaceholder(variable_name="history"),("human", "{query}")
])# LlamaIndex方式 - 更简单但功能有限
# from llama_index.core import PromptTemplate as LlamaPromptTemplate
# llama_template = LlamaPromptTemplate(
# "你是{role},专长是{expertise}。\n用户问题:{query}"
# )# LangChain优势:
# 1. 支持复杂的消息结构
# 2. 内置历史消息管理
# 3. 更强的类型安全
# 4. 丰富的组合能力
LangChain vs Haystack:
# LangChain - 声明式模板定义
langchain_prompt = PromptTemplate.from_template("基于文档:{documents}\n回答问题:{question}"
)# Haystack - 更程序化的方式
# from haystack.nodes import PromptNode, PromptTemplate as HaystackPrompt
# haystack_prompt = HaystackPrompt(
# name="qa_prompt",
# prompt_text="基于文档:{documents}\n回答问题:{question}"
# )# LangChain优势:
# 1. 更直观的API设计
# 2. 更好的IDE支持
# 3. 更丰富的验证机制
# 4. 更强的可组合性
5. 替代方案与优化空间
5.1 替代方案
方案1:Pydantic增强参数校验
from pydantic import BaseModel, validator
from langchain_core.prompts import PromptTemplateclass EnhancedPromptTemplate(BaseModel):"""增强的Prompt模板,提供更严格的参数校验"""template: strrequired_params: dict[str, type]optional_params: dict[str, type] = {}@validator('template')def validate_template_syntax(cls, v):"""验证模板语法"""try:# 检查花括号匹配open_count = v.count('{')close_count = v.count('}')if open_count != close_count:raise ValueError("花括号不匹配")return vexcept Exception as e:raise ValueError(f"模板语法错误: {e}")def format_with_validation(self, **kwargs):"""带类型验证的格式化"""# 检查必需参数missing_params = set(self.required_params.keys()) - set(kwargs.keys())if missing_params:raise ValueError(f"缺少必需参数: {missing_params}")# 类型验证for param, expected_type in self.required_params.items():if param in kwargs and not isinstance(kwargs[param], expected_type):raise TypeError(f"参数{param}类型错误,期望{expected_type},实际{type(kwargs[param])}")# 使用LangChain的PromptTemplate进行格式化prompt_template = PromptTemplate.from_template(self.template)return prompt_template.format(**kwargs)# 使用示例
enhanced_template = EnhancedPromptTemplate(template="用户{user_name}的订单{order_id}状态为{status}",required_params={"user_name": str, "order_id": int, "status": str}
)# 正确使用
result = enhanced_template.format_with_validation(user_name="张三",order_id=12345,status="已发货"
)
print(result)# 错误使用会抛出异常
try:enhanced_template.format_with_validation(user_name="张三",order_id="12345", # 类型错误:应该是intstatus="已发货")
except TypeError as e:print(f"类型验证失败: {e}")
方案2:Jinja2复杂逻辑模板
from jinja2 import Environment, Template
from langchain_core.prompts import PromptTemplateclass Jinja2EnhancedTemplate:"""基于Jinja2的增强模板,支持复杂逻辑"""def __init__(self, template_str: str):self.env = Environment()self.template = self.env.from_string(template_str)# 添加自定义过滤器self.env.filters['currency'] = self._currency_filterself.env.filters['date_format'] = self._date_filterdef _currency_filter(self, value):"""货币格式化过滤器"""return f"¥{value:,.2f}"def _date_filter(self, value, format='%Y-%m-%d'):"""日期格式化过滤器"""if hasattr(value, 'strftime'):return value.strftime(format)return str(value)def render(self, **kwargs):"""渲染模板"""return self.template.render(**kwargs)# 复杂模板示例
complex_template = Jinja2EnhancedTemplate("""
尊敬的{{ customer.name }},您的订单详情:
{% for item in order.items %}
- {{ item.name }}: {{ item.price | currency }} x {{ item.quantity }}
{% endfor %}总计:{{ order.total | currency }}
下单时间:{{ order.created_at | date_format('%Y年%m月%d日 %H:%M') }}{% if order.total > 1000 %}
🎉 恭喜您享受免费配送!
{% else %}
配送费:{{ shipping_fee | currency }}
{% endif %}{% if customer.vip_level > 0 %}
VIP{{ customer.vip_level }}专享优惠已自动应用。
{% endif %}
""")# 使用示例
from datetime import datetimeorder_data = {"customer": {"name": "张三","vip_level": 2},"order": {"items": [{"name": "智能手机", "price": 2999.00, "quantity": 1},{"name": "手机壳", "price": 59.90, "quantity": 2}],"total": 3118.80,"created_at": datetime.now()},"shipping_fee": 15.00
}result = complex_template.render(**order_data)
print(result)
5.2 优化方向
实用性优化:
class SmartPromptTemplate:"""智能Prompt模板 - 提供更好的用户体验"""def __init__(self, template: str):self.base_template = PromptTemplate.from_template(template)self.usage_stats = {}self.error_history = []def format_with_suggestions(self, **kwargs):"""带建议的格式化"""try:result = self.base_template.format(**kwargs)self._record_success(kwargs)return resultexcept KeyError as e:missing_var = str(e).strip("'")suggestions = self._get_suggestions(missing_var, kwargs)error_msg = f"缺少变量 '{missing_var}'"if suggestions:error_msg += f",您是否想要使用: {', '.join(suggestions)}?"self._record_error(error_msg, kwargs)raise ValueError(error_msg)def _get_suggestions(self, missing_var: str, provided_vars: dict) -> list[str]:"""基于编辑距离提供变量名建议"""suggestions = []for var in provided_vars.keys():if self._edit_distance(missing_var, var) <= 2:suggestions.append(var)return suggestionsdef _edit_distance(self, s1: str, s2: str) -> int:"""计算编辑距离"""if len(s1) < len(s2):return self._edit_distance(s2, s1)if len(s2) == 0:return len(s1)previous_row = list(range(len(s2) + 1))for i, c1 in enumerate(s1):current_row = [i + 1]for j, c2 in enumerate(s2):insertions = previous_row[j + 1] + 1deletions = current_row[j] + 1substitutions = previous_row[j] + (c1 != c2)current_row.append(min(insertions, deletions, substitutions))previous_row = current_rowreturn previous_row[-1]def _record_success(self, kwargs: dict):"""记录成功使用"""for var in kwargs.keys():self.usage_stats[var] = self.usage_stats.get(var, 0) + 1def _record_error(self, error: str, kwargs: dict):"""记录错误"""self.error_history.append({"error": error,"provided_vars": list(kwargs.keys()),"timestamp": datetime.now()})def get_usage_report(self) -> dict:"""获取使用报告"""return {"most_used_vars": sorted(self.usage_stats.items(), key=lambda x: x[1], reverse=True),"recent_errors": self.error_history[-5:],"total_errors": len(self.error_history)}# 使用示例
smart_template = SmartPromptTemplate("用户{user_name}在{date}购买了{product_name},价格{price}"
)# 正确使用
result = smart_template.format_with_suggestions(user_name="李四",date="2024-01-15",product_name="笔记本电脑",price="¥5999"
)# 错误使用 - 会提供建议
try:smart_template.format_with_suggestions(username="李四", # 错误的变量名date="2024-01-15",product_name="笔记本电脑",price="¥5999")
except ValueError as e:print(f"智能提示: {e}")
性能优化:
import functools
import hashlib
from typing import Any, Dictclass CachedPromptTemplate:"""带缓存的Prompt模板 - 性能优化"""def __init__(self, template: str, cache_size: int = 1000):self.base_template = PromptTemplate.from_template(template)self.cache_size = cache_sizeself.cache = {}self.cache_hits = 0self.cache_misses = 0def format(self, **kwargs) -> str:"""带缓存的格式化"""# 生成缓存键cache_key = self._generate_cache_key(kwargs)# 检查缓存if cache_key in self.cache:self.cache_hits += 1return self.cache[cache_key]# 缓存未命中,执行格式化self.cache_misses += 1result = self.base_template.format(**kwargs)# 存储到缓存if len(self.cache) >= self.cache_size:# 简单的LRU:删除第一个元素first_key = next(iter(self.cache))del self.cache[first_key]self.cache[cache_key] = resultreturn resultdef _generate_cache_key(self, kwargs: Dict[str, Any]) -> str:"""生成缓存键"""# 将参数转换为可哈希的字符串sorted_items = sorted(kwargs.items())key_string = str(sorted_items)return hashlib.md5(key_string.encode()).hexdigest()def get_cache_stats(self) -> dict:"""获取缓存统计"""total_requests = self.cache_hits + self.cache_misseshit_rate = self.cache_hits / total_requests if total_requests > 0 else 0return {"cache_hits": self.cache_hits,"cache_misses": self.cache_misses,"hit_rate": f"{hit_rate:.2%}","cache_size": len(self.cache)}def clear_cache(self):"""清空缓存"""self.cache.clear()self.cache_hits = 0self.cache_misses = 0# 使用示例
cached_template = CachedPromptTemplate("处理{task_type}任务:{description},优先级:{priority}"
)# 多次使用相同参数 - 会命中缓存
for i in range(5):result = cached_template.format(task_type="数据分析",description="用户行为分析",priority="高")print("缓存统计:", cached_template.get_cache_stats())
扩展性优化:
from abc import ABC, abstractmethod
from typing import Protocolclass PromptProcessor(Protocol):"""Prompt处理器协议"""def process(self, template: str, **kwargs) -> str:...class BasePromptExtension(ABC):"""Prompt扩展基类"""@abstractmethoddef pre_process(self, template: str, **kwargs) -> tuple[str, dict]:"""预处理"""pass@abstractmethoddef post_process(self, result: str, **kwargs) -> str:"""后处理"""passclass SecurityExtension(BasePromptExtension):"""安全扩展 - 过滤敏感信息"""def __init__(self):self.sensitive_patterns = [r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b', # 信用卡号r'\b\d{11}\b', # 手机号r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b' # 邮箱]def pre_process(self, template: str, **kwargs) -> tuple[str, dict]:"""预处理 - 检查模板安全性"""import re# 检查模板中是否包含敏感信息占位符for pattern in self.sensitive_patterns:if re.search(pattern, template):raise ValueError("模板包含潜在的敏感信息模式")return template, kwargsdef post_process(self, result: str, **kwargs) -> str:"""后处理 - 脱敏处理"""import re# 脱敏处理for pattern in self.sensitive_patterns:result = re.sub(pattern, lambda m: '*' * len(m.group()), result)return resultclass ExtensiblePromptTemplate:"""可扩展的Prompt模板"""def __init__(self, template: str):self.base_template = PromptTemplate.from_template(template)self.extensions: list[BasePromptExtension] = []def add_extension(self, extension: BasePromptExtension):"""添加扩展"""self.extensions.append(extension)def format(self, **kwargs) -> str:"""扩展格式化"""template = self.base_template.templateprocessed_kwargs = kwargs.copy()# 预处理for extension in self.extensions:template, processed_kwargs = extension.pre_process(template, **processed_kwargs)# 格式化temp_template = PromptTemplate.from_template(template)result = temp_template.format(**processed_kwargs)# 后处理for extension in self.extensions:result = extension.post_process(result, **processed_kwargs)return result# 使用示例
extensible_template = ExtensiblePromptTemplate("用户{user_name}的联系方式是{contact},订单号{order_id}"
)# 添加安全扩展
extensible_template.add_extension(SecurityExtension())# 使用 - 敏感信息会被自动脱敏
result = extensible_template.format(user_name="张三",contact="13812345678",order_id="ORD123456"
)
print(result) # 输出:用户张三的联系方式是***********,订单号ORD123456
6. 总结
6.1 核心价值总结
PromptTemplate和ChatPromptTemplate作为LangChain的核心组件,解决了AI应用开发中的关键问题:
- 模板化管理:将Prompt从硬编码转向模板化,提高可维护性
- 参数验证:自动验证输入参数,减少运行时错误
- 类型适配:针对不同LLM类型提供专门优化
- 组合能力:支持复杂的模板组合和链式处理
- 生态集成:与LangChain生态系统无缝集成
6.2 最佳实践建议
- 选择合适的模板类型:简单文本用PromptTemplate,对话场景用ChatPromptTemplate
- 合理使用partial变量:对于固定参数使用partial预填充
- 重视参数验证:利用内置验证机制确保数据安全
- 模板组合优化:通过模板组合实现复杂逻辑
- 性能考虑:对于高频使用场景考虑缓存优化
6.3 发展趋势
随着AI应用的复杂化,Prompt管理将朝着更智能、更安全、更高效的方向发展:
- 智能化:自动优化Prompt效果,智能参数建议
- 安全性:更强的安全检查和脱敏能力
- 性能:更高效的缓存和批处理机制
- 可视化:图形化的Prompt设计和调试工具
- 多模态:支持文本、图像、音频等多模态内容
LangChain的PromptTemplate体系为现代AI应用提供了坚实的基础,是构建可靠、可维护AI系统的重要工具。