LangChain源码-03提示模板
LangChain提示模板架构分析 LangChain的提示模板系统是一个多层级的语言处理引擎，负责将结构化数据转换为LLM可理解的指令。其核心架构包含： 输入层：接收用户输入的模板、变量和上下文 解析层：通过模板解析器、变量提取器和上下文构建器处理原始输入 模板处理层：支持多种模板类型（字符串、聊天、FewShot等）和渲染引擎（f-string/Jinja2） 输出层：包含多种输出解析器和缓存
·
03. Langchain提示模板的架构和源码分析
提示模板架构概览
提示模板系统是LangChain的语言引擎，它负责将结构化数据转换为LLM能够理解的文本指令。让我们先理解其整体架构：
# 提示模板架构图
prompt_architecture = '''
用户输入 → 模板解析 → 变量替换 → 格式验证 → 提示词生成 → LLM调用
↓ ↓ ↓ ↓ ↓ ↓
模板定义 ← 变量提取 ← 类型检查 ← 格式规范 ← 文本生成 ← 响应处理
'''
# 核心组件关系
component_relationships = '''
BasePromptTemplate (抽象基类)
↑
PromptTemplate (基础模板) ←→ ChatPromptTemplate (聊天模板)
↑ ↑
具体实现类 具体实现类
↑ ↑
FewShotPromptTemplate ←→ MessagesPlaceholder
↑ ↑
高级模板类 动态消息类
'''
提示模板系统架构图
提示模板架构图
查看大图：鼠标右键 → “在新标签页打开图片” → 浏览器自带放大
%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#4CAF50', 'primaryTextColor': '#fff', 'primaryBorderColor': '#2E7D32', 'lineColor': '#424242', 'secondaryColor': '#2196F3', 'tertiaryColor': '#FF9800', 'background': '#ffffff', 'mainBkg': '#4CAF50', 'secondBkg': '#2196F3', 'tertiaryBkg': '#FF9800'}}}%%
graph TD
%% 用户输入层
UserInput["用户输入<br/>User Input"] --> TemplateInput["模板输入<br/>Template Input"]
UserInput --> VariableInput["变量输入<br/>Variable Input"]
UserInput --> ContextInput["上下文输入<br/>Context Input"]
%% 模板解析层
TemplateInput --> TemplateParser["模板解析器<br/>Template Parser"]
VariableInput --> VariableExtractor["变量提取器<br/>Variable Extractor"]
ContextInput --> ContextBuilder["上下文构建器<br/>Context Builder"]
%% 模板处理核心层
TemplateParser --> TemplateValidator["模板验证器<br/>Template Validator"]
VariableExtractor --> VariableValidator["变量验证器<br/>Variable Validator"]
ContextBuilder --> ContextValidator["上下文验证器<br/>Context Validator"]
%% 模板类型层
TemplateValidator --> StringTemplate["字符串模板<br/>StringPromptTemplate"]
TemplateValidator --> ChatTemplate["聊天模板<br/>ChatPromptTemplate"]
TemplateValidator --> FewShotTemplate["FewShot模板<br/>FewShotPromptTemplate"]
TemplateValidator --> CustomTemplate["自定义模板<br/>CustomPromptTemplate"]
%% 格式化引擎层
StringTemplate --> FStringRenderer["f-string渲染器<br/>F-String Renderer"]
StringTemplate --> Jinja2Renderer["Jinja2渲染器<br/>Jinja2 Renderer"]
ChatTemplate --> MessageFormatter["消息格式化器<br/>Message Formatter"]
MessageFormatter --> HumanMessageTemplate["人类消息模板<br/>HumanMessagePromptTemplate"]
MessageFormatter --> AIMessageTemplate["AI消息模板<br/>AIMessagePromptTemplate"]
MessageFormatter --> SystemMessageTemplate["系统消息模板<br/>SystemMessagePromptTemplate"]
MessageFormatter --> MessagesPlaceholder["消息占位符<br/>MessagesPlaceholder"]
FewShotTemplate --> ExampleSelector["示例选择器<br/>Example Selector"]
ExampleSelector --> SemanticSelector["语义相似度选择器<br/>SemanticSimilarityExampleSelector"]
ExampleSelector --> LengthSelector["长度选择器<br/>LengthBasedExampleSelector"]
ExampleSelector --> CustomSelector["自定义选择器<br/>CustomExampleSelector"]
%% 渲染执行层
FStringRenderer --> TemplateRenderer["模板渲染器<br/>Template Renderer"]
Jinja2Renderer --> TemplateRenderer
MessageFormatter --> TemplateRenderer
ExampleSelector --> TemplateRenderer
%% 输出处理层
TemplateRenderer --> OutputParser["输出解析器<br/>Output Parser"]
OutputParser --> StrOutputParser["字符串解析器<br/>StrOutputParser"]
OutputParser --> JsonOutputParser["JSON解析器<br/>JsonOutputParser"]
OutputParser --> ListOutputParser["列表解析器<br/>ListOutputParser"]
OutputParser --> CustomOutputParser["自定义解析器<br/>CustomOutputParser"]
%% 缓存优化层
TemplateRenderer --> TemplateCache["模板缓存<br/>Template Cache"]
TemplateCache --> LRUCache["LRU缓存<br/>LRU Cache"]
TemplateCache --> AsyncCache["异步缓存<br/>Async Cache"]
TemplateCache --> BatchCache["批量缓存<br/>Batch Cache"]
%% 性能监控层
TemplateRenderer --> PerformanceMonitor["性能监控<br/>Performance Monitor"]
PerformanceMonitor --> RenderTimeTracker["渲染时间跟踪<br/>Render Time Tracker"]
PerformanceMonitor --> MemoryUsageMonitor["内存使用监控<br/>Memory Usage Monitor"]
PerformanceMonitor --> ErrorRateTracker["错误率跟踪<br/>Error Rate Tracker"]
%% 最终输出层
OutputParser --> FormattedOutput["格式化输出<br/>Formatted Output"]
StrOutputParser --> FormattedOutput
JsonOutputParser --> FormattedOutput
ListOutputParser --> FormattedOutput
CustomOutputParser --> FormattedOutput
FormattedOutput --> LLMInput["LLM输入<br/>LLM Input"]
LLMInput --> LLMResponse["LLM响应<br/>LLM Response"]
%% 样式定义
classDef inputLayer fill:#E3F2FD,stroke:#1976D2,stroke-width:2px,color:#000
classDef parserLayer fill:#E8F5E8,stroke:#388E3C,stroke-width:2px,color:#000
classDef templateLayer fill:#FFF3E0,stroke:#F57C00,stroke-width:2px,color:#000
classDef rendererLayer fill:#F3E5F5,stroke:#7B1FA2,stroke-width:2px,color:#000
classDef outputLayer fill:#E1F5FE,stroke:#0288D1,stroke-width:2px,color:#000
classDef cacheLayer fill:#FFEBEE,stroke:#D32F2F,stroke-width:2px,color:#000
classDef monitorLayer fill:#F1F8E9,stroke:#689F38,stroke-width:2px,color:#000
%% 应用样式
class UserInput,TemplateInput,VariableInput,ContextInput inputLayer
class TemplateParser,VariableExtractor,ContextBuilder,TemplateValidator,VariableValidator,ContextValidator parserLayer
class StringTemplate,ChatTemplate,FewShotTemplate,CustomTemplate,HumanMessageTemplate,AIMessageTemplate,SystemMessageTemplate,MessagesPlaceholder templateLayer
class FStringRenderer,Jinja2Renderer,MessageFormatter,ExampleSelector,SemanticSelector,LengthSelector,CustomSelector,TemplateRenderer rendererLayer
class OutputParser,StrOutputParser,JsonOutputParser,ListOutputParser,CustomOutputParser,FormattedOutput,LLMInput,LLMResponse outputLayer
class TemplateCache,LRUCache,AsyncCache,BatchCache cacheLayer
class PerformanceMonitor,RenderTimeTracker,MemoryUsageMonitor,ErrorRateTracker monitorLayer
%% 主要流程连接
UserInput --> TemplateParser
TemplateParser --> TemplateValidator
TemplateValidator --> StringTemplate
StringTemplate --> FStringRenderer
FStringRenderer --> TemplateRenderer
TemplateRenderer --> OutputParser
OutputParser --> FormattedOutput
%% FormattedOutput --> LLMInput
%% LLMInput --> LLMResponse
提示模板核心组件关系图
查看大图：鼠标右键 → “在新标签页打开图片” → 浏览器自带放大
%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#4CAF50', 'primaryTextColor': '#fff', 'primaryBorderColor': '#2E7D32', 'lineColor': '#424242', 'secondaryColor': '#2196F3', 'tertiaryColor': '#FF9800', 'background': '#ffffff', 'mainBkg': '#4CAF50', 'secondBkg': '#2196F3', 'tertiaryBkg': '#FF9800'}}}%%
graph TD
%% 抽象基类层
BasePromptTemplate["BasePromptTemplate<br/>提示模板抽象基类"]
BaseMessage["BaseMessage<br/>消息抽象基类"]
BaseOutputParser["BaseOutputParser<br/>输出解析器抽象基类"]
BaseExampleSelector["BaseExampleSelector<br/>示例选择器抽象基类"]
%% 字符串模板继承体系
BasePromptTemplate --> StringPromptTemplate["StringPromptTemplate<br/>字符串模板基类"]
StringPromptTemplate --> PromptTemplate["PromptTemplate<br/>基础提示模板"]
%% 聊天模板继承体系
BasePromptTemplate --> BaseChatPromptTemplate["BaseChatPromptTemplate<br/>聊天模板基类"]
BaseChatPromptTemplate --> ChatPromptTemplate["ChatPromptTemplate<br/>聊天提示模板"]
%% 高级模板类型
BasePromptTemplate --> FewShotPromptTemplate["FewShotPromptTemplate<br/>FewShot提示模板"]
%% 消息类型继承体系
BaseMessage --> HumanMessage["HumanMessage<br/>人类消息"]
BaseMessage --> AIMessage["AIMessage<br/>AI消息"]
BaseMessage --> SystemMessage["SystemMessage<br/>系统消息"]
BaseMessage --> FunctionMessage["FunctionMessage<br/>函数消息"]
BaseMessage --> ToolMessage["ToolMessage<br/>工具消息"]
BaseMessage --> ChatMessage["ChatMessage<br/>通用聊天消息"]
%% 消息提示模板体系
BaseMessagePromptTemplate["BaseMessagePromptTemplate<br/>消息提示模板基类"]
BaseMessagePromptTemplate --> HumanMessagePromptTemplate["HumanMessagePromptTemplate<br/>人类消息提示模板"]
BaseMessagePromptTemplate --> AIMessagePromptTemplate["AIMessagePromptTemplate<br/>AI消息提示模板"]
BaseMessagePromptTemplate --> SystemMessagePromptTemplate["SystemMessagePromptTemplate<br/>系统消息提示模板"]
BaseMessagePromptTemplate --> MessagesPlaceholder["MessagesPlaceholder<br/>消息占位符"]
BaseMessagePromptTemplate --> ChatMessagePromptTemplate["ChatMessagePromptTemplate<br/>通用聊天消息提示模板"]
%% 输出解析器体系
BaseOutputParser --> StrOutputParser["StrOutputParser<br/>字符串输出解析器"]
BaseOutputParser --> JsonOutputParser["JsonOutputParser<br/>JSON输出解析器"]
BaseOutputParser --> ListOutputParser["ListOutputParser<br/>列表输出解析器"]
BaseOutputParser --> PydanticOutputParser["PydanticOutputParser<br/>Pydantic输出解析器"]
%% 示例选择器体系
BaseExampleSelector --> LengthBasedExampleSelector["LengthBasedExampleSelector<br/>基于长度的示例选择器"]
BaseExampleSelector --> SemanticSimilarityExampleSelector["SemanticSimilarityExampleSelector<br/>语义相似度示例选择器"]
BaseExampleSelector --> CustomExampleSelector["CustomExampleSelector<br/>自定义示例选择器"]
%% 模板渲染器体系
TemplateRenderer["TemplateRenderer<br/>模板渲染器"]
TemplateRenderer --> FStringRenderer["FStringRenderer<br/>f-string渲染器"]
TemplateRenderer --> Jinja2Renderer["Jinja2Renderer<br/>Jinja2渲染器"]
TemplateRenderer --> AdvancedRenderer["AdvancedTemplateRenderer<br/>高级渲染器"]
%% 验证器体系
TemplateValidator["TemplateValidator<br/>模板验证器"]
TemplateValidator --> FStringValidator["F-string验证器"]
TemplateValidator --> Jinja2Validator["Jinja2验证器"]
TemplateValidator --> SafetyChecker["安全性检查器"]
%% 缓存系统
PromptTemplateCache["PromptTemplateCache<br/>提示模板缓存"]
PromptTemplateCache --> LRUCache["LRU缓存"]
PromptTemplateCache --> AsyncCache["异步缓存"]
PromptTemplateCache --> BatchCache["批量缓存"]
%% 异步处理器
AsyncPromptTemplateProcessor["AsyncPromptTemplateProcessor<br/>异步提示模板处理器"]
%% 核心依赖关系
PromptTemplate -.-> TemplateRenderer
PromptTemplate -.-> TemplateValidator
PromptTemplate -.-> PromptTemplateCache
ChatPromptTemplate -.-> HumanMessagePromptTemplate
ChatPromptTemplate -.-> AIMessagePromptTemplate
ChatPromptTemplate -.-> SystemMessagePromptTemplate
ChatPromptTemplate -.-> MessagesPlaceholder
HumanMessagePromptTemplate -.-> PromptTemplate
AIMessagePromptTemplate -.-> PromptTemplate
SystemMessagePromptTemplate -.-> PromptTemplate
FewShotPromptTemplate -.-> BaseExampleSelector
FewShotPromptTemplate -.-> PromptTemplate
SemanticSimilarityExampleSelector -.-> BaseMessage
SemanticSimilarityExampleSelector -.-> BaseOutputParser
TemplateRenderer -.-> BaseOutputParser
TemplateValidator -.-> BaseMessage
%% 实际应用场景连接
IntelligentCustomerServicePrompts["IntelligentCustomerServicePrompts<br/>智能客服提示系统"]
CodeGenerationPrompts["CodeGenerationPrompts<br/>代码生成提示系统"]
IntelligentCustomerServicePrompts -.-> ChatPromptTemplate
IntelligentCustomerServicePrompts -.-> FewShotPromptTemplate
CodeGenerationPrompts -.-> PromptTemplate
CodeGenerationPrompts -.-> ChatPromptTemplate
%% 核心配置和工具
ChainConfig["ChainConfig<br/>链配置"]
RunInfo["RunInfo<br/>运行信息"]
ChainPerformanceMonitor["ChainPerformanceMonitor<br/>链性能监控"]
PromptTemplate -.-> ChainConfig
ChatPromptTemplate -.-> RunInfo
AsyncPromptTemplateProcessor -.-> ChainPerformanceMonitor
%% 样式定义
classDef abstractClass fill:#E3F2FD,stroke:#1976D2,stroke-width:3px,color:#000
classDef stringTemplate fill:#E8F5E8,stroke:#388E3C,stroke-width:2px,color:#000
classDef chatTemplate fill:#FFF3E0,stroke:#F57C00,stroke-width:2px,color:#000
classDef messageClass fill:#F3E5F5,stroke:#7B1FA2,stroke-width:2px,color:#000
classDef outputParser fill:#E1F5FE,stroke:#0288D1,stroke-width:2px,color:#000
classDef selector fill:#FFEBEE,stroke:#D32F2F,stroke-width:2px,color:#000
classDef renderer fill:#F1F8E9,stroke:#689F38,stroke-width:2px,color:#000
classDef cache fill:#FFF8E1,stroke:#FFA000,stroke-width:2px,color:#000
classDef application fill:#E0F2F1,stroke:#00796B,stroke-width:2px,color:#000
%% 应用样式
class BasePromptTemplate,BaseMessage,BaseOutputParser,BaseExampleSelector abstractClass
class StringPromptTemplate,PromptTemplate stringTemplate
class BaseChatPromptTemplate,ChatPromptTemplate chatTemplate
class HumanMessage,AIMessage,SystemMessage,FunctionMessage,ToolMessage,ChatMessage,BaseMessagePromptTemplate,HumanMessagePromptTemplate,AIMessagePromptTemplate,SystemMessagePromptTemplate,MessagesPlaceholder,ChatMessagePromptTemplate messageClass
class StrOutputParser,JsonOutputParser,ListOutputParser,PydanticOutputParser outputParser
class LengthBasedExampleSelector,SemanticSimilarityExampleSelector,CustomExampleSelector,BaseExampleSelector selector
class TemplateRenderer,FStringRenderer,Jinja2Renderer,AdvancedRenderer,TemplateValidator renderer
class PromptTemplateCache,LRUCache,AsyncCache,BatchCache cache
class IntelligentCustomerServicePrompts,CodeGenerationPrompts,AsyncPromptTemplateProcessor,ChainConfig,RunInfo,ChainPerformanceMonitor application
%% 主要继承关系
BasePromptTemplate --> StringPromptTemplate
BasePromptTemplate --> BaseChatPromptTemplate
BasePromptTemplate --> FewShotPromptTemplate
StringPromptTemplate --> PromptTemplate
BaseChatPromptTemplate --> ChatPromptTemplate
BaseMessage --> HumanMessage
BaseMessage --> AIMessage
BaseMessage --> SystemMessage
BaseMessage --> FunctionMessage
BaseMessage --> ToolMessage
BaseMessage --> ChatMessage
BaseMessagePromptTemplate --> HumanMessagePromptTemplate
BaseMessagePromptTemplate --> AIMessagePromptTemplate
BaseMessagePromptTemplate --> SystemMessagePromptTemplate
BaseMessagePromptTemplate --> MessagesPlaceholder
BaseMessagePromptTemplate --> ChatMessagePromptTemplate
BaseOutputParser --> StrOutputParser
BaseOutputParser --> JsonOutputParser
BaseOutputParser --> ListOutputParser
BaseOutputParser --> PydanticOutputParser
BaseExampleSelector --> LengthBasedExampleSelector
BaseExampleSelector --> SemanticSimilarityExampleSelector
BaseExampleSelector --> CustomExampleSelector
核心抽象基类分析
1. BasePromptTemplate - 提示模板根接口
# 文件: libs/langchain-core/prompts/base.py
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, Field
class BasePromptTemplate(BaseModel, ABC):
"""所有提示模板的抽象基类"""
input_variables: List[str] = Field(description="模板中的输入变量列表")
output_parser: Optional[BaseOutputParser] = Field(default=None, description="输出解析器")
class Config:
"""Pydantic配置"""
arbitrary_types_allowed = True
@abstractmethod
def format(self, **kwargs: Any) -> str:
"""格式化模板为字符串 - 必须实现"""
pass
@abstractmethod
def format_messages(self, **kwargs: Any) -> List[BaseMessage]:
"""格式化模板为消息列表 - 必须实现"""
pass
def validate_input_variables(self, input_variables: List[str]) -> None:
"""验证输入变量"""
if len(input_variables) != len(set(input_variables)):
raise ValueError("输入变量列表中包含重复项")
def partial(self, **kwargs: Any) -> BasePromptTemplate:
"""创建部分填充的模板副本"""
# 创建新模板实例，部分变量已填充
from langchain_core.prompts import PromptTemplate
# 提取已填充的变量
partial_variables = {k: v for k, v in kwargs.items() if k in self.input_variables}
# 创建剩余变量的新模板
remaining_variables = [v for v in self.input_variables if v not in partial_variables]
return PromptTemplate(
template=self.template, # 需要子类提供
input_variables=remaining_variables,
partial_variables=partial_variables,
output_parser=self.output_parser
)
# 设计分析
base_design_analysis = '''
1. 抽象接口设计:
- 定义了format()和format_messages()两个核心方法
- 所有子类必须实现这两个方法
- 保证了API的一致性
2. 类型安全:
- 使用Pydantic进行数据验证
- 明确的类型注解
- 运行时类型检查
3. 可扩展性:
- 支持部分变量填充
- 支持输出解析器
- 支持自定义验证逻辑
4. 验证机制:
- 输入变量验证
- 重复变量检查
- 可扩展的验证框架
'''
2. StringPromptTemplate - 字符串模板基类
# 文件: libs/langchain-core/prompts/string.py
class StringPromptTemplate(BasePromptTemplate, ABC):
"""字符串提示模板的基类"""
template: str = Field(description="模板字符串")
def format(self, **kwargs: Any) -> str:
"""格式化模板为字符串"""
# 获取输入变量值
input_dict = self._get_input_dict(**kwargs)
# 调用子类实现的格式化方法
return self.format_template(**input_dict)
@abstractmethod
def format_template(self, **kwargs: Any) -> str:
"""子类必须实现的模板格式化方法"""
pass
def _get_input_dict(self, **kwargs: Any) -> Dict[str, Any]:
"""获取输入变量字典"""
input_dict = {}
# 处理部分变量
if hasattr(self, 'partial_variables') and self.partial_variables:
input_dict.update(self.partial_variables)
# 处理传入的变量
for variable_name in self.input_variables:
if variable_name in kwargs:
input_dict[variable_name] = kwargs[variable_name]
elif variable_name not in input_dict:
raise KeyError(f"输入变量 '{variable_name}' 未提供")
return input_dict
@property
def _prompt_type(self) -> str:
"""返回提示类型"""
return "string"
# 字符串模板特点
string_template_features = '''
1. 专注于字符串处理:
- 模板是字符串格式
- 输出也是字符串
- 适合简单的文本生成任务
2. 变量管理:
- 自动处理部分变量
- 严格的变量验证
- 清晰的错误提示
3. 扩展性设计:
- 抽象format_template方法
- 支持不同的字符串格式化策略
- 统一的变量处理逻辑
'''
PromptTemplate实现深度分析
1. 基础PromptTemplate实现
# 文件: libs/langchain-core/prompts/prompt.py
class PromptTemplate(StringPromptTemplate):
"""基础提示词模板实现"""
template: str = Field(description="模板字符串，支持{f}格式的变量")
validate_template: bool = Field(default=True, description="是否验证模板格式")
template_format: str = Field(default="f-string", description="模板格式")
def __init__(self, **kwargs: Any) -> None:
"""初始化提示词模板"""
super().__init__(**kwargs)
# 模板格式验证
if self.validate_template:
self._validate_template()
def _validate_template(self) -> None:
"""验证模板格式是否正确"""
if self.template_format == "f-string":
self._validate_f_string_template()
elif self.template_format == "jinja2":
self._validate_jinja2_template()
else:
raise ValueError(f"不支持的模板格式: {self.template_format}")
def _validate_f_string_template(self) -> None:
"""验证f-string格式模板"""
try:
# 尝试用空字典格式化模板
self.template.format(**{var: "" for var in self.input_variables})
except KeyError as e:
raise ValueError(f"模板中使用了未声明的变量: {e}")
except Exception as e:
raise ValueError(f"模板格式错误: {e}")
def format_template(self, **kwargs: Any) -> str:
"""使用f-string格式化模板"""
try:
return self.template.format(**kwargs)
except KeyError as e:
raise KeyError(f"缺少必需的输入变量: {e}")
except Exception as e:
raise ValueError(f"模板格式化错误: {e}")
@classmethod
def from_template(cls, template: str, **kwargs: Any) -> "PromptTemplate":
"""从模板字符串创建PromptTemplate"""
# 提取输入变量
input_variables = cls._extract_input_variables(template)
return cls(
template=template,
input_variables=input_variables,
**kwargs
)
@staticmethod
def _extract_input_variables(template: str) -> List[str]:
"""从模板字符串中提取输入变量"""
# 简单的f-string变量提取
import re
# 匹配{f}格式的变量
pattern = r'\{(\w+)\}'
matches = re.findall(pattern, template)
# 去重并保持顺序
seen = set()
variables = []
for match in matches:
if match not in seen:
seen.add(match)
variables.append(match)
return variables
@classmethod
def from_examples(
cls,
examples: List[Example],
suffix: str,
input_variables: List[str],
example_separator: str = "\n\n",
prefix: str = "",
**kwargs: Any,
) -> "PromptTemplate":
"""从示例创建FewShot样式的模板"""
# 构建模板字符串
template_parts = []
if prefix:
template_parts.append(prefix)
# 添加示例
for example in examples:
example_str = suffix.format(**example)
template_parts.append(example_str)
# 添加最终输入格式
template_parts.append(suffix)
template = example_separator.join(template_parts)
return cls(
template=template,
input_variables=input_variables,
**kwargs
)
# PromptTemplate使用示例
usage_examples = '''
# 基础使用
template = PromptTemplate.from_template("告诉我关于{topic}的信息")
result = template.format(topic="人工智能")
print(result) # 输出: 告诉我关于人工智能的信息
# 复杂模板
complex_template = PromptTemplate(
template="""
作为{role}，请用{style}的风格解释{concept}。
要求：
1. {requirement1}
2. {requirement2}
请确保内容{constraint}。
""",
input_variables=["role", "style", "concept", "requirement1", "requirement2", "constraint"]
)
# 从示例创建
examples = [
{"input": "happy", "output": "sad"},
{"input": "tall", "output": "short"}
]
few_shot_template = PromptTemplate.from_examples(
examples=examples,
suffix="输入: {input}\n输出: {output}",
input_variables=["input", "output"]
)
'''
2. 模板验证和错误处理
# 文件: libs/langchain-core/prompts/validation.py
class TemplateValidator:
"""模板验证器"""
@staticmethod
def validate_f_string_template(template: str, input_variables: List[str]) -> None:
"""验证f-string格式模板"""
try:
# 创建测试字典
test_kwargs = {var: f"TEST_{var}" for var in input_variables}
# 尝试格式化
formatted = template.format(**test_kwargs)
# 检查是否有未使用的变量
unused_vars = set(input_variables) - set(test_kwargs.keys())
if unused_vars:
logger.warning(f"模板中可能未使用这些变量: {unused_vars}")
except KeyError as e:
raise ValueError(f"模板中使用了未声明的变量: {e}")
except ValueError as e:
raise ValueError(f"模板格式错误: {e}")
except Exception as e:
raise ValueError(f"模板验证失败: {e}")
@staticmethod
def validate_jinja2_template(template: str, input_variables: List[str]) -> None:
"""验证Jinja2格式模板"""
try:
from jinja2 import Template, UndefinedError
jinja_template = Template(template)
# 测试渲染
test_context = {var: f"TEST_{var}" for var in input_variables}
try:
jinja_template.render(**test_context)
except UndefinedError as e:
raise ValueError(f"Jinja2模板中使用了未定义的变量: {e}")
except ImportError:
raise ValueError("Jinja2未安装，无法验证模板")
except Exception as e:
raise ValueError(f"Jinja2模板验证失败: {e}")
@staticmethod
def check_template_safety(template: str, template_format: str) -> bool:
"""检查模板安全性"""
if template_format == "f-string":
# 检查是否包含危险表达式
dangerous_patterns = [
r'\{.*__.*__.*\}', # 双下划线方法
r'\{.*import.*\}', # import语句
r'\{.*exec.*\}', # exec函数
r'\{.*eval.*\}', # eval函数
]
import re
for pattern in dangerous_patterns:
if re.search(pattern, template):
return False
elif template_format == "jinja2":
# Jinja2有自己的安全机制
return True
return True
# 验证系统特点
validation_features = '''
1. 多格式支持:
- F-string格式验证
- Jinja2格式验证
- 可扩展其他格式
2. 安全性检查:
- 危险表达式检测
- 代码注入防护
- 模板注入防护
3. 详细错误信息:
- 具体的错误位置
- 清晰的错误描述
- 修复建议提供
4. 性能优化:
- 高效的正则表达式
- 缓存验证结果
- 增量验证支持
'''
ChatPromptTemplate聊天模板分析
1. 聊天消息体系设计
# 文件: libs/langchain-core/messages.py
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class BaseMessage(BaseModel, ABC):
"""所有消息类型的基础类"""
content: str = Field(description="消息内容")
additional_kwargs: Dict[str, Any] = Field(default_factory=dict, description="额外参数")
@property
@abstractmethod
def type(self) -> str:
"""消息类型"""
pass
def dict(self, **kwargs: Any) -> Dict[str, Any]:
"""转换为字典"""
return {
"type": self.type,
"content": self.content,
"additional_kwargs": self.additional_kwargs,
}
class HumanMessage(BaseMessage):
"""用户消息"""
@property
def type(self) -> str:
return "human"
class AIMessage(BaseMessage):
"""AI回复消息"""
@property
def type(self) -> str:
return "ai"
class SystemMessage(BaseMessage):
"""系统消息"""
@property
def type(self) -> str:
return "system"
class FunctionMessage(BaseMessage):
"""函数调用消息"""
name: str = Field(description="函数名称")
@property
def type(self) -> str:
return "function"
class ToolMessage(BaseMessage):
"""工具调用消息"""
tool_call_id: str = Field(description="工具调用ID")
@property
def type(self) -> str:
return "tool"
# 消息体系设计分析
message_design_analysis = '''
1. 清晰的类型层次:
- BaseMessage定义通用接口
- 具体消息类型实现特定行为
- 类型属性用于标识和路由
2. 可扩展性设计:
- 支持additional_kwargs扩展
- 易于添加新消息类型
- 统一的序列化接口
3. 类型安全:
- 使用Pydantic进行验证
- 明确的类型注解
- 运行时类型检查
4. 与LLM接口兼容:
- 易于转换为不同LLM的格式
- 支持角色标记
- 支持额外元数据
'''
2. ChatPromptTemplate实现
# 文件: libs/langchain-core/prompts/chat.py
class BaseChatPromptTemplate(BasePromptTemplate, ABC):
"""聊天提示模板的抽象基类"""
@abstractmethod
def format_messages(self, **kwargs: Any) -> List[BaseMessage]:
"""格式化模板为消息列表"""
pass
def format(self, **kwargs: Any) -> str:
"""格式化模板为字符串（默认实现）"""
messages = self.format_messages(**kwargs)
return "\n".join([f"{msg.type}: {msg.content}" for msg in messages])
class ChatPromptTemplate(BaseChatPromptTemplate):
"""聊天提示模板实现"""
messages: List[Union[BaseMessagePromptTemplate, BaseMessage]] = Field(
description="消息模板列表"
)
def __init__(self, messages: List[Union[BaseMessagePromptTemplate, BaseMessage]], **kwargs: Any):
"""初始化聊天提示模板"""
# 提取输入变量
input_variables = self._extract_input_variables(messages)
super().__init__(messages=messages, input_variables=input_variables, **kwargs)
def _extract_input_variables(self, messages: List[Union[BaseMessagePromptTemplate, BaseMessage]]) -> List[str]:
"""从消息模板中提取输入变量"""
input_variables = []
for message in messages:
if isinstance(message, BaseMessagePromptTemplate):
# 从消息模板中提取变量
variables = message.input_variables
input_variables.extend(variables)
# BaseMessage不需要变量
# 去重并保持顺序
seen = set()
unique_variables = []
for var in input_variables:
if var not in seen:
seen.add(var)
unique_variables.append(var)
return unique_variables
def format_messages(self, **kwargs: Any) -> List[BaseMessage]:
"""格式化消息模板"""
formatted_messages = []
for message_template in self.messages:
if isinstance(message_template, BaseMessagePromptTemplate):
# 格式化消息模板
formatted_message = message_template.format(**kwargs)
formatted_messages.append(formatted_message)
elif isinstance(message_template, BaseMessage):
# 直接使用消息
formatted_messages.append(message_template)
else:
raise ValueError(f"不支持的消息类型: {type(message_template)}")
return formatted_messages
@classmethod
def from_messages(cls, messages: List[Union[BaseMessagePromptTemplate, BaseMessage]]) -> "ChatPromptTemplate":
"""从消息列表创建聊天提示模板"""
return cls(messages=messages)
@classmethod
def from_template(cls, template: str, **kwargs: Any) -> "ChatPromptTemplate":
"""从模板字符串创建聊天提示模板"""
# 创建人类消息模板
human_message = HumanMessagePromptTemplate.from_template(template)
return cls(messages=[human_message], **kwargs)
# 消息提示模板基类
class BaseMessagePromptTemplate(BaseModel, ABC):
"""消息提示模板的抽象基类"""
@abstractmethod
def format(self, **kwargs: Any) -> BaseMessage:
"""格式化消息模板"""
pass
@property
@abstractmethod
def input_variables(self) -> List[str]:
"""返回输入变量列表"""
pass
# 具体消息提示模板实现
class HumanMessagePromptTemplate(BaseMessagePromptTemplate):
"""人类消息提示模板"""
prompt: PromptTemplate
def format(self, **kwargs: Any) -> HumanMessage:
"""格式化为人类消息"""
content = self.prompt.format(**kwargs)
return HumanMessage(content=content)
@property
def input_variables(self) -> List[str]:
return self.prompt.input_variables
@classmethod
def from_template(cls, template: str, **kwargs: Any) -> "HumanMessagePromptTemplate":
"""从模板创建"""
prompt = PromptTemplate.from_template(template)
return cls(prompt=prompt)
class AIMessagePromptTemplate(BaseMessagePromptTemplate):
"""AI消息提示模板"""
prompt: PromptTemplate
def format(self, **kwargs: Any) -> AIMessage:
"""格式化为AI消息"""
content = self.prompt.format(**kwargs)
return AIMessage(content=content)
@property
def input_variables(self) -> List[str]:
return self.prompt.input_variables
class SystemMessagePromptTemplate(BaseMessagePromptTemplate):
"""系统消息提示模板"""
prompt: PromptTemplate
def format(self, **kwargs: Any) -> SystemMessage:
"""格式化为系统消息"""
content = self.prompt.format(**kwargs)
return SystemMessage(content=content)
@property
def input_variables(self) -> List[str]:
return self.prompt.input_variables
# 聊天模板使用示例
chat_template_examples = '''
# 基础聊天模板
chat_template = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个专业的AI助手"),
HumanMessagePromptTemplate.from_template("请解释{concept}"),
AIMessagePromptTemplate.from_template("我来为您解释{concept}")
])
# 复杂聊天模板
complex_chat = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template("你是一个{role}，擅长{expertise}"),
HumanMessagePromptTemplate.from_template("我的问题是：{question}"),
AIMessagePromptTemplate.from_template("让我来帮您解答：首先，{step1}，然后{step2}")
])
# 使用模板
messages = chat_template.format_messages(concept="机器学习")
for msg in messages:
print(f"{msg.type}: {msg.content}")
'''
3. 高级聊天模板功能
# 文件: libs/langchain-core/prompts/chat.py
class MessagesPlaceholder(BaseMessagePromptTemplate):
"""消息占位符 - 动态插入消息"""
variable_name: str = Field(description="变量名称")
optional: bool = Field(default=False, description="是否可选")
def format(self, **kwargs: Any) -> List[BaseMessage]:
"""格式化占位符"""
if self.variable_name in kwargs:
messages = kwargs[self.variable_name]
if isinstance(messages, list):
return messages
else:
return [messages]
elif self.optional:
return []
else:
raise KeyError(f"必需的变量 '{self.variable_name}' 未提供")
@property
def input_variables(self) -> List[str]:
return [self.variable_name] if not self.optional else []
class ChatMessagePromptTemplate(BaseMessagePromptTemplate):
"""通用聊天消息提示模板"""
prompt: PromptTemplate
role: str = Field(description="消息角色")
def format(self, **kwargs: Any) -> BaseMessage:
"""格式化为指定角色的消息"""
content = self.prompt.format(**kwargs)
# 根据角色创建相应类型的消息
if self.role == "human":
return HumanMessage(content=content)
elif self.role == "ai":
return AIMessage(content=content)
elif self.role == "system":
return SystemMessage(content=content)
else:
# 创建通用聊天消息
from langchain_core.messages import ChatMessage
return ChatMessage(content=content, role=self.role)
@property
def input_variables(self) -> List[str]:
return self.prompt.input_variables
# 高级功能示例
advanced_features_examples = '''
# MessagesPlaceholder使用
dynamic_chat = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个AI助手"),
MessagesPlaceholder(variable_name="history"), # 动态历史消息
HumanMessagePromptTemplate.from_template("{user_input}")
])
# 使用占位符
history = [
HumanMessage(content="你好"),
AIMessage(content="你好！有什么可以帮助您的吗？")
]
messages = dynamic_chat.format_messages(
history=history,
user_input="请解释机器学习"
)
# 通用角色模板
generic_chat = ChatPromptTemplate.from_messages([
ChatMessagePromptTemplate.from_template("system", "你是一个{role}"),
ChatMessagePromptTemplate.from_template("user", "{question}"),
ChatMessagePromptTemplate.from_template("assistant", "让我来帮您：{answer}")
])
'''
FewShotPromptTemplate高级模板分析
1. FewShot模板架构设计
# 文件: libs/langchain-core/prompts/few_shot.py
class FewShotPromptTemplate(BasePromptTemplate):
"""FewShot提示词模板实现"""
examples: Optional[List[dict]] = Field(default=None, description="示例列表")
example_selector: Optional[BaseExampleSelector] = Field(default=None, description="示例选择器")
example_prompt: PromptTemplate = Field(description="示例提示词模板")
suffix: str = Field(description="后缀模板")
input_variables: List[str] = Field(description="输入变量列表")
example_separator: str = Field(default="\n\n", description="示例分隔符")
prefix: str = Field(default="", description="前缀模板")
def __init__(self, **kwargs: Any) -> None:
"""初始化FewShot模板"""
super().__init__(**kwargs)
# 验证示例和选择器
if self.examples is not None and self.example_selector is not None:
raise ValueError("不能同时指定examples和example_selector")
if self.examples is None and self.example_selector is None:
raise ValueError("必须指定examples或example_selector之一")
def format(self, **kwargs: Any) -> str:
"""格式化FewShot模板"""
# 1. 获取示例
examples = self._get_examples(**kwargs)
# 2. 格式化示例
example_strings = []
for example in examples:
example_string = self.example_prompt.format(**example)
example_strings.append(example_string)
# 3. 构建最终提示词
parts = []
# 添加前缀
if self.prefix:
parts.append(self.prefix)
# 添加示例
if example_strings:
examples_string = self.example_separator.join(example_strings)
parts.append(examples_string)
# 添加后缀（包含实际输入）
suffix_string = self.suffix.format(**kwargs)
parts.append(suffix_string)
# 4. 组合所有部分
return self.example_separator.join(parts)
def _get_examples(self, **kwargs: Any) -> List[dict]:
"""获取示例"""
if self.example_selector is not None:
# 使用选择器动态选择示例
return self.example_selector.select_examples(kwargs)
elif self.examples is not None:
# 使用固定示例
return self.examples
else:
return []
def format_messages(self, **kwargs: Any) -> List[BaseMessage]:
"""格式化为消息列表"""
# 获取格式化后的字符串
formatted_string = self.format(**kwargs)
# 转换为人类消息
return [HumanMessage(content=formatted_string)]
# FewShot实现特点
few_shot_features = '''
1. 灵活的示例管理:
- 支持固定示例列表
- 支持动态示例选择
- 可扩展的示例选择器接口
2. 模板化示例:
- 每个示例使用独立模板
- 支持复杂的示例格式
- 一致的变量处理
3. 动态示例选择:
- 基于输入选择最相关示例
- 支持相似度匹配
- 可定制的选择策略
4. 格式化灵活性:
- 可配置的分隔符
- 支持前缀和后缀
- 多种输出格式
'''
2. 示例选择器实现
# 文件: libs/langchain-core/example_selectors/base.py
class BaseExampleSelector(ABC):
"""示例选择器的抽象基类"""
@abstractmethod
def select_examples(self, input_variables: Dict[str, Any]) -> List[dict]:
"""根据输入变量选择示例"""
pass
@abstractmethod
def add_example(self, example: dict) -> None:
"""添加示例到选择器"""
pass
# 文件: libs/langchain-core/example_selectors/semantic_similarity.py
class SemanticSimilarityExampleSelector(BaseExampleSelector):
"""基于语义相似度的示例选择器"""
def __init__(
self,
embeddings: Embeddings,
vectorstore_cls: Type[VectorStore],
k: int = 4,
input_keys: Optional[List[str]] = None,
**vectorstore_kwargs: Any
):
"""初始化语义相似度选择器"""
self.embeddings = embeddings
self.vectorstore_cls = vectorstore_cls
self.k = k
self.input_keys = input_keys
# 创建向量存储
self.vectorstore = vectorstore_cls(
embedding_function=embeddings,
**vectorstore_kwargs
)
self.examples: List[dict] = []
def add_example(self, example: dict) -> None:
"""添加示例"""
self.examples.append(example)
# 提取文本用于嵌入
if self.input_keys:
texts = [example[key] for key in self.input_keys if key in example]
else:
texts = list(example.values())
text = " ".join(texts)
# 添加到向量存储
self.vectorstore.add_texts(
texts=[text],
metadatas=[{"example": example}]
)
def select_examples(self, input_variables: Dict[str, Any]) -> List[dict]:
"""基于语义相似度选择示例"""
# 构建查询文本
if self.input_keys:
query_texts = [input_variables[key] for key in self.input_keys if key in input_variables]
else:
query_texts = list(input_variables.values())
query_text = " ".join(query_texts)
# 相似度搜索
results = self.vectorstore.similarity_search(
query=query_text,
k=self.k
)
# 提取示例
selected_examples = []
for result in results:
if result.metadata and "example" in result.metadata:
selected_examples.append(result.metadata["example"])
return selected_examples
# 使用示例
selector_usage_example = '''
# 创建嵌入模型
embeddings = OpenAIEmbeddings()
# 创建选择器
selector = SemanticSimilarityExampleSelector(
embeddings=embeddings,
vectorstore_cls=Chroma,
k=3
)
# 添加示例
examples = [
{"input": "happy", "output": "sad"},
{"input": "tall", "output": "short"},
{"input": "fast", "output": "slow"},
{"input": "hot", "output": "cold"}
]
for example in examples:
selector.add_example(example)
# 创建FewShot模板
few_shot_prompt = FewShotPromptTemplate(
example_selector=selector,
example_prompt=PromptTemplate.from_template("输入: {input}\n输出: {output}"),
suffix="输入: {input}\n输出:",
input_variables=["input"]
)
# 使用模板
result = few_shot_prompt.format(input="big")
print(result)
'''
模板渲染和格式化系统
1. 模板渲染引擎
# 文件: libs/langchain-core/prompts/string.py
class TemplateRenderer:
"""模板渲染引擎"""
@staticmethod
def render_f_string(template: str, variables: Dict[str, Any]) -> str:
"""渲染f-string格式模板"""
try:
return template.format(**variables)
except KeyError as e:
raise KeyError(f"缺少必需的变量: {e}")
except Exception as e:
raise ValueError(f"模板渲染失败: {e}")
@staticmethod
def render_jinja2(template: str, variables: Dict[str, Any]) -> str:
"""渲染Jinja2格式模板"""
try:
from jinja2 import Template
jinja_template = Template(template)
return jinja_template.render(**variables)
except ImportError:
raise ValueError("Jinja2未安装，无法渲染模板")
except Exception as e:
raise ValueError(f"Jinja2模板渲染失败: {e}")
@staticmethod
def render_custom(template: str, variables: Dict[str, Any], renderer_func) -> str:
"""使用自定义渲染函数"""
try:
return renderer_func(template, variables)
except Exception as e:
raise ValueError(f"自定义模板渲染失败: {e}")
# 高级渲染功能
class AdvancedTemplateRenderer:
"""高级模板渲染器"""
def __init__(self):
self.custom_functions = {}
self.filters = {}
def add_custom_function(self, name: str, func: Callable) -> None:
"""添加自定义函数"""
self.custom_functions[name] = func
def add_filter(self, name: str, filter_func: Callable) -> None:
"""添加过滤器"""
self.filters[name] = filter_func
def render_with_functions(self, template: str, variables: Dict[str, Any]) -> str:
"""使用自定义函数渲染模板"""
# 创建包含自定义函数的执行环境
execution_env = {
**variables,
**self.custom_functions,
**self.filters
}
try:
# 使用eval执行模板（注意安全风险）
return eval(f'f"""{template}"""', execution_env)
except Exception as e:
raise ValueError(f"带函数的模板渲染失败: {e}")
2. 格式化输出系统
# 文件: libs/langchain-core/output_parsers/base.py
class BaseOutputParser(BaseModel, ABC):
"""输出解析器的抽象基类"""
@abstractmethod
def parse(self, text: str) -> Any:
"""解析文本输出"""
pass
def parse_with_prompt(self, completion: str, prompt: PromptValue) -> Any:
"""结合提示词解析输出"""
return self.parse(completion)
# 文件: libs/langchain-core/output_parsers/string.py
class StrOutputParser(BaseOutputParser):
"""字符串输出解析器"""
def parse(self, text: str) -> str:
"""直接返回字符串"""
return text.strip()
# 文件: libs/langchain-core/output_parsers/json.py
class JsonOutputParser(BaseOutputParser):
"""JSON输出解析器"""
def parse(self, text: str) -> Any:
"""解析JSON格式的输出"""
try:
# 清理JSON字符串
cleaned_text = self._clean_json_text(text)
return json.loads(cleaned_text)
except json.JSONDecodeError as e:
raise ValueError(f"无效的JSON格式: {e}")
def _clean_json_text(self, text: str) -> str:
"""清理JSON文本"""
# 移除Markdown代码块标记
text = re.sub(r'```json\s*', '', text)
text = re.sub(r'\s*```', '', text)
# 提取JSON内容
json_pattern = r'\{.*\}|\[.*\]'
match = re.search(json_pattern, text, re.DOTALL)
if match:
return match.group(0)
else:
return text.strip()
# 格式化系统特点
formatting_system_features = '''
1. 多样化解析器:
- 字符串解析器
- JSON解析器
- 列表解析器
- 结构化解析器
2. 错误处理:
- 格式验证
- 错误恢复
- 详细错误信息
3. 可扩展性:
- 自定义解析器
- 插件式架构
- 组合解析器
4. 性能优化:
- 缓存解析结果
- 增量解析
- 并行处理
'''
模板系统性能优化
1. 缓存优化策略
# 文件: libs/langchain-core/prompts/cache.py
class PromptTemplateCache:
"""提示模板缓存系统"""
def __init__(self, max_size: int = 1000):
self.max_size = max_size
self.template_cache: Dict[str, Any] = {}
self.render_cache: Dict[str, str] = {}
self.access_count: Dict[str, int] = {}
def cache_template(self, template_key: str, template: Any) -> None:
"""缓存模板"""
if len(self.template_cache) >= self.max_size:
self._evict_lru()
self.template_cache[template_key] = template
self.access_count[template_key] = 0
def cache_render_result(self, render_key: str, result: str) -> None:
"""缓存渲染结果"""
if len(self.render_cache) >= self.max_size:
self._evict_lru_render()
self.render_cache[render_key] = result
self.access_count[render_key] = 0
def get_cached_template(self, template_key: str) -> Optional[Any]:
"""获取缓存的模板"""
if template_key in self.template_cache:
self.access_count[template_key] += 1
return self.template_cache[template_key]
return None
def get_cached_render_result(self, render_key: str) -> Optional[str]:
"""获取缓存的渲染结果"""
if render_key in self.render_cache:
self.access_count[render_key] += 1
return self.render_cache[render_key]
return None
def _evict_lru(self) -> None:
"""淘汰最近最少使用的缓存"""
if not self.access_count:
return
# 找到最少使用的项
lru_key = min(self.access_count.keys(), key=lambda k: self.access_count[k])
# 移除该项
if lru_key in self.template_cache:
del self.template_cache[lru_key]
if lru_key in self.render_cache:
del self.render_cache[lru_key]
del self.access_count[lru_key]
def clear_cache(self) -> None:
"""清空缓存"""
self.template_cache.clear()
self.render_cache.clear()
self.access_count.clear()
# 缓存优化示例
cached_template_example = '''
# 带缓存的提示模板
class CachedPromptTemplate(PromptTemplate):
"""带缓存功能的提示模板"""
_cache = PromptTemplateCache(max_size=100)
def format(self, **kwargs: Any) -> str:
"""格式化并缓存结果"""
# 生成缓存键
cache_key = f"{self.template}_{sorted(kwargs.items())}"
# 检查缓存
cached_result = self._cache.get_cached_render_result(cache_key)
if cached_result is not None:
return cached_result
# 执行格式化
result = super().format(**kwargs)
# 缓存结果
self._cache.cache_render_result(cache_key, result)
return result
'''
2. 异步处理优化
# 文件: libs/langchain-core/prompts/async_utils.py
import asyncio
from typing import List, Dict, Any, Optional
class AsyncPromptTemplateProcessor:
"""异步提示模板处理器"""
def __init__(self, max_concurrent: int = 10):
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
async def process_batch_async(
self,
templates: List[BasePromptTemplate],
inputs_list: List[Dict[str, Any]]
) -> List[str]:
"""异步批量处理模板"""
tasks = []
for template, inputs in zip(templates, inputs_list):
task = self._process_single_async(template, inputs)
tasks.append(task)
# 并发执行所有任务
results = await asyncio.gather(*tasks)
return results
async def _process_single_async(
self,
template: BasePromptTemplate,
inputs: Dict[str, Any]
) -> str:
"""异步处理单个模板"""
async with self.semaphore:
# 模拟异步处理（实际实现取决于具体模板）
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
template.format,
**inputs
)
async def stream_process_async(
self,
template: BasePromptTemplate,
inputs: Dict[str, Any],
chunk_size: int = 100
) -> AsyncIterator[str]:
"""异步流式处理"""
# 将大输入分块处理
input_chunks = self._chunk_inputs(inputs, chunk_size)
for chunk in input_chunks:
result = await self._process_single_async(template, chunk)
yield result
def _chunk_inputs(self, inputs: Dict[str, Any], chunk_size: int) -> List[Dict[str, Any]]]:
"""将输入分块"""
# 简单的分块逻辑（实际实现可能更复杂）
chunks = []
current_chunk = {}
for key, value in inputs.items():
if isinstance(value, str) and len(value) > chunk_size:
# 分割长字符串
for i in range(0, len(value), chunk_size):
chunk = {**current_chunk, key: value[i:i+chunk_size]}
chunks.append(chunk)
else:
current_chunk[key] = value
if current_chunk:
chunks.append(current_chunk)
return chunks
# 异步使用示例
async_usage_example = '''
# 异步批量处理
async def batch_process_templates():
processor = AsyncPromptTemplateProcessor(max_concurrent=5)
# 创建多个模板
templates = [
PromptTemplate.from_template("处理: {input1}"),
PromptTemplate.from_template("分析: {input2}"),
PromptTemplate.from_template("总结: {input3}")
]
# 准备输入数据
inputs_list = [
{"input1": "数据1"},
{"input2": "数据2"},
{"input3": "数据3"}
]
# 异步批量处理
results = await processor.process_batch_async(templates, inputs_list)
for i, result in enumerate(results):
print(f"结果{i+1}: {result}")
# 运行异步处理
# asyncio.run(batch_process_templates())
'''
模板系统最佳实践
1. 设计原则
模板设计原则 = '''
1. 清晰性原则:
- 模板变量命名清晰
- 模板结构易于理解
- 注释和文档完善
2. 可维护性原则:
- 模块化设计
- 避免过度复杂
- 支持版本控制
3. 性能原则:
- 缓存常用模板
- 避免重复渲染
- 支持异步处理
4. 安全性原则:
- 输入验证和清理
- 防止模板注入
- 敏感信息保护
5. 扩展性原则:
- 支持自定义格式
- 插件式架构
- 向后兼容性
'''
2. 性能优化技巧
性能优化技巧 = '''
1. 模板缓存:
- 缓存编译后的模板
- 缓存渲染结果
- 使用LRU淘汰策略
2. 变量预处理:
- 预验证变量格式
- 预转换变量类型
- 减少运行时计算
3. 批量处理:
- 批量渲染模板
- 并行处理多个模板
- 异步I/O操作
4. 内存优化:
- 及时释放大对象
- 使用生成器代替列表
- 避免内存泄漏
5. 监控和度量:
- 跟踪渲染时间
- 监控内存使用
- 记录错误率
'''
实际应用案例分析
1. 智能客服提示系统
class IntelligentCustomerServicePrompts:
"""智能客服提示词系统"""
def __init__(self):
self.setup_prompts()
def setup_prompts(self):
"""设置各种客服场景的提示词"""
# 意图识别提示词
self.intent_recognition_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个智能客服意图识别系统，请准确识别客户的意图。"
),
HumanMessagePromptTemplate.from_template(
"客户消息：{customer_message}\n\n"
"请识别客户的意图，并从以下类别中选择最匹配的："
"{intent_categories}\n\n"
"只返回意图类别名称，不要解释。"
)
])
# 情感分析提示词
self.sentiment_analysis_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个情感分析专家，请准确分析文本的情感倾向。"
),
HumanMessagePromptTemplate.from_template(
"文本内容：{text}\n\n"
"请分析这段文本的情感倾向，从以下选项中选择："
"积极、中性、消极、愤怒、焦虑\n\n"
"只返回情感类别，不要解释。"
)
])
# 回复生成提示词
self.response_generation_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个专业的客服代表，请根据以下信息为客户提供帮助。"
),
HumanMessagePromptTemplate.from_template(
"客户信息：\n"
"- 客户ID：{customer_id}\n"
"- 会员等级：{membership_level}\n"
"- 历史交互：{interaction_history}\n\n"
"当前问题：{current_issue}\n"
"检测到的意图：{detected_intent}\n"
"情感状态：{sentiment}\n\n"
"请提供专业、友好、个性化的回复。"
)
])
# 多语言支持
self.multilingual_prompts = {
'zh': self.create_chinese_prompts(),
'en': self.create_english_prompts(),
'fr': self.create_french_prompts()
}
def create_chinese_prompts(self):
"""创建中文提示词"""
return {
'greeting': ChatPromptTemplate.from_template(
"您好！我是AI客服助手，很高兴为您服务。请问有什么可以帮助您的吗？"
),
'apology': ChatPromptTemplate.from_template(
"非常抱歉给您带来了不便，我会立即为您处理这个问题。"
),
'escalation': ChatPromptTemplate.from_template(
"您的问题比较复杂，我需要为您转接到人工客服，请稍等片刻。"
)
}
def generate_intent_recognition(self, customer_message: str, intent_categories: List[str]) -> str:
"""生成意图识别提示词"""
result = self.intent_recognition_prompt.format(
customer_message=customer_message,
intent_categories=", ".join(intent_categories)
)
return result
def generate_sentiment_analysis(self, text: str) -> str:
"""生成情感分析提示词"""
result = self.sentiment_analysis_prompt.format(text=text)
return result
def generate_response(self, customer_info: dict, current_issue: str,
detected_intent: str, sentiment: str) -> str:
"""生成客服回复提示词"""
result = self.response_generation_prompt.format(
customer_id=customer_info.get('id', '未知'),
membership_level=customer_info.get('level', '普通'),
interaction_history=customer_info.get('history', '无'),
current_issue=current_issue,
detected_intent=detected_intent,
sentiment=sentiment
)
return result
# 使用示例
cs_prompts = IntelligentCustomerServicePrompts()
# 测试意图识别
intent_result = cs_prompts.generate_intent_recognition(
"我的订单什么时候能到？",
["订单查询", "产品咨询", "技术支持", "投诉建议"]
)
print("意图识别提示词:", intent_result)
# 测试情感分析
sentiment_result = cs_prompts.generate_sentiment_analysis(
"这个产品太差了，我很不满意！"
)
print("情感分析提示词:", sentiment_result)
2. 代码生成提示系统
class CodeGenerationPrompts:
"""代码生成提示词系统"""
def __init__(self):
self.setup_code_prompts()
def setup_code_prompts(self):
"""设置代码生成相关的提示词"""
# 代码解释提示词
self.code_explanation_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个专业的编程导师，擅长用通俗易懂的语言解释复杂的编程概念。"
),
HumanMessagePromptTemplate.from_template(
"请解释以下{language}代码的功能和实现原理：\n\n"
"```{language}\n{code}\n```\n\n"
"要求：\n"
"1. 用{explanation_style}的风格解释\n"
"2. 适合{audience_level}水平理解\n"
"3. 包含关键概念说明\n"
"4. 如果有优化建议请提出"
)
])
# 代码生成提示词
self.code_generation_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个专业的{language}开发者，能够编写高质量、可维护的代码。"
),
HumanMessagePromptTemplate.from_template(
"请用{language}编写代码实现以下功能：\n\n"
"功能描述：{functionality}\n\n"
"技术要求：\n"
"- 复杂度：{complexity}\n"
"- 性能要求：{performance_requirements}\n"
"- 代码风格：{code_style}\n"
"- 特殊约束：{constraints}\n\n"
"请提供完整的代码实现，包括必要的注释和错误处理。"
)
])
# 代码审查提示词
self.code_review_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个经验丰富的代码审查专家，能够从多个维度评估代码质量。"
),
HumanMessagePromptTemplate.from_template(
"请对以下{language}代码进行专业审查：\n\n"
"```{language}\n{code}\n```\n\n"
"审查维度：\n"
"1. 代码质量：{quality_focus}\n"
"2. 性能优化：{performance_focus}\n"
"3. 安全性：{security_focus}\n"
"4. 可维护性：{maintainability_focus}\n\n"
"请提供详细的审查报告，包括问题指出和改进建议。"
)
])
# 测试用例生成提示词
self.test_generation_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"你是一个专业的测试工程师，擅长编写全面的测试用例。"
),
HumanMessagePromptTemplate.from_template(
"请为以下{language}函数编写测试用例：\n\n"
"函数代码：\n```{language}\n{function_code}\n```\n\n"
"测试要求：\n"
"- 测试框架：{testing_framework}\n"
"- 覆盖率要求：{coverage_requirement}\n"
"- 边界情况：{boundary_cases}\n"
"- 异常处理：{exception_handling}\n\n"
"请提供完整的测试代码，包括测试数据和断言。"
)
])
def generate_code_explanation(self, language: str, code: str,
explanation_style: str = "通俗易懂",
audience_level: str = "中级") -> str:
"""生成代码解释提示词"""
result = self.code_explanation_prompt.format(
language=language,
code=code,
explanation_style=explanation_style,
audience_level=audience_level
)
return result
def generate_code(self, language: str, functionality: str,
complexity: str = "中等",
performance_requirements: str = "标准",
code_style: str = "PEP8",
constraints: str = "无特殊约束") -> str:
"""生成代码提示词"""
result = self.code_generation_prompt.format(
language=language,
functionality=functionality,
complexity=complexity,
performance_requirements=performance_requirements,
code_style=code_style,
constraints=constraints
)
return result
def generate_code_review(self, language: str, code: str,
quality_focus: str = "代码规范和可读性",
performance_focus: str = "算法复杂度和资源使用",
security_focus: str = "输入验证和异常处理",
maintainability_focus: str = "模块化和文档") -> str:
"""生成代码审查提示词"""
result = self.code_review_prompt.format(
language=language,
code=code,
quality_focus=quality_focus,
performance_focus=performance_focus,
security_focus=security_focus,
maintainability_focus=maintainability_focus
)
return result
# 使用示例
code_prompts = CodeGenerationPrompts()
# 测试代码解释
explanation_prompt = code_prompts.generate_code_explanation(
language="Python",
code="def quicksort(arr):\n if len(arr) <= 1:\n return arr\n pivot = arr[len(arr) // 2]\n left = [x for x in arr if x < pivot]\n middle = [x for x in arr if x == pivot]\n right = [x for x in arr if x > pivot]\n return quicksort(left) + middle + quicksort(right)",
explanation_style="循序渐进",
audience_level="初级"
)
print("代码解释提示词:", explanation_prompt)
源码学习要点总结
1. 核心设计模式
设计模式总结 = {
'模板方法模式': {
'应用': 'BasePromptTemplate定义框架，子类实现具体格式化',
'好处': '保证接口一致性，支持多种模板格式',
'实现': 'format()定义流程，子类实现format_template()'
},
'组合模式': {
'应用': 'ChatPromptTemplate组合多个消息模板',
'好处': '可以构建复杂的对话结构',
'实现': 'messages列表包含不同类型的消息模板'
},
'策略模式': {
'应用': '不同的模板格式策略（f-string, jinja2）',
'好处': '支持多种模板语法',
'实现': 'TemplateRenderer支持不同渲染策略'
},
'工厂模式': {
'应用': 'from_template(), from_messages()等工厂方法',
'好处': '简化模板创建过程',
'实现': '类方法创建模板实例'
}
}
2. 性能优化要点
性能优化要点 = '''
1. 模板缓存:
- 缓存编译后的模板
- 缓存渲染结果
- 使用LRU淘汰策略
2. 变量提取优化:
- 使用正则表达式快速提取变量
- 缓存变量提取结果
- 支持增量更新
3. 格式验证优化:
- 延迟验证（按需验证）
- 缓存验证结果
- 支持验证跳过
4. 内存管理:
- 及时释放大模板
- 使用弱引用
- 避免循环引用
5. 异步支持:
- 异步模板渲染
- 批量异步处理
- 流式输出支持
'''
小结
通过本章的深入学习，我们全面掌握了LangChain提示模板系统的源码实现：
核心架构理解
- 抽象层次设计：BasePromptTemplate → StringPromptTemplate/BaseChatPromptTemplate → 具体实现
- 模板类型体系：支持字符串模板、聊天模板、FewShot模板等多种格式
- 消息体系设计：BaseMessage及其子类的面向对象设计
源码实现细节
- PromptTemplate实现：f-string格式化、变量提取、模板验证的完整流程
- ChatPromptTemplate实现：消息模板组合、角色管理、动态格式化的实现机制
- FewShotPromptTemplate实现：示例管理、动态选择、模板组合的复杂逻辑
高级功能掌握
- 模板验证系统：多格式支持、安全性检查、详细错误处理
- 示例选择器：语义相似度选择、动态示例管理、可扩展架构
- 性能优化：缓存策略、异步处理、内存管理的综合优化
实际应用能力
- 模板设计：能够设计高效、可维护的提示模板系统
- 性能优化：具备模板系统性能调优和问题解决能力
- 定制开发：能够基于源码进行功能扩展和架构改进
提示模板系统是LangChain的灵魂，深入理解其源码让我们能够设计出更加智能、高效的AI交互界面，为构建强大的AI应用奠定坚实基础！ 🎯✨
在下一章中，我们将深入分析LangChain格式化输出的源码实现，探索如何让AI的输出更加结构化和易于处理！
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
18
0- 0
已为社区贡献44条内容
所有评论(0)