LangChain-05-Prompts模块
模块概览
职责与定位
Prompts模块是LangChain中构建和管理提示词模板的核心系统。该模块提供了灵活强大的模板化能力,使开发者能够将动态变量注入到提示词中,支持多种格式和组合方式,是构建LLM应用的基础组件。
核心职责:
- 提供多种提示词模板类型(字符串模板、聊天模板、Few-Shot模板等)
- 支持变量插值和格式化(f-string、jinja2、mustache)
- 管理消息序列和占位符
- 提供模板组合和复用机制
- 支持部分变量和默认值
- 提供模板验证和类型检查
- 支持Runnable接口,可与其他组件链式组合
输入输出
输入:
- 模板字符串或消息列表
- 变量字典(key-value对)
- 示例列表(Few-Shot场景)
- RunnableConfig(可选配置)
输出:
- PromptValue: 统一的提示值抽象
- StringPromptValue: 纯文本提示
- ChatPromptValue: 消息列表提示
- 可转换为字符串或消息列表,适配不同类型的模型
上下游依赖
依赖:
langchain_core.messages: 消息类型(BaseMessage、HumanMessage、AIMessage等)langchain_core.prompt_values: PromptValue封装langchain_core.runnables: Runnable接口,支持链式调用pydantic: 数据验证和模型定义jinja2: Jinja2模板引擎(可选)
被依赖:
langchain_core.language_models: 语言模型的输入langchain_core.chains: 链式调用的起点langchain_core.agents: Agent的提示构造- 所有需要构造提示词的组件
完整系统架构图
flowchart TB
subgraph User["用户层"]
APP[应用代码]
end
subgraph PromptsModule["Prompts模块"]
subgraph Creation["模板创建"]
FROM_TEMPLATE[from_template<br/>from_messages]
CONSTRUCTOR[构造函数]
end
subgraph Templates["模板层"]
PROMPT_TPL[PromptTemplate<br/>字符串模板]
CHAT_TPL[ChatPromptTemplate<br/>聊天模板]
FEW_SHOT[FewShotPromptTemplate<br/>Few-Shot模板]
end
subgraph Components["组件层"]
MSG_PLACEHOLDER[MessagesPlaceholder<br/>消息占位符]
MSG_TPL[MessagePromptTemplate<br/>消息模板]
IMAGE_TPL[ImagePromptTemplate<br/>图片模板]
end
subgraph Processing["处理层"]
VARIABLE_EXTRACT[变量提取<br/>get_template_variables]
FORMATTER[格式化器<br/>f-string/jinja2/mustache]
VALIDATOR[验证器<br/>check_valid_template]
end
subgraph Output["输出层"]
PROMPT_VALUE[PromptValue<br/>抽象基类]
STRING_VALUE[StringPromptValue<br/>文本输出]
CHAT_VALUE[ChatPromptValue<br/>消息列表输出]
end
end
subgraph Downstream["下游组件"]
LLM[LLM<br/>文本生成模型]
CHAT_MODEL[ChatModel<br/>对话模型]
CHAIN[Chain<br/>链式调用]
end
subgraph Core["核心基础设施"]
RUNNABLE[Runnable<br/>可运行接口]
CALLBACKS[Callbacks<br/>回调系统]
MESSAGES[Messages<br/>消息类型]
end
APP --> FROM_TEMPLATE
APP --> CONSTRUCTOR
FROM_TEMPLATE --> PROMPT_TPL
FROM_TEMPLATE --> CHAT_TPL
CONSTRUCTOR --> FEW_SHOT
PROMPT_TPL --> VARIABLE_EXTRACT
CHAT_TPL --> MSG_TPL
CHAT_TPL --> MSG_PLACEHOLDER
MSG_TPL --> IMAGE_TPL
VARIABLE_EXTRACT --> FORMATTER
FORMATTER --> VALIDATOR
VALIDATOR --> PROMPT_VALUE
PROMPT_TPL --> STRING_VALUE
CHAT_TPL --> CHAT_VALUE
STRING_VALUE --> PROMPT_VALUE
CHAT_VALUE --> PROMPT_VALUE
PROMPT_VALUE --> LLM
PROMPT_VALUE --> CHAT_MODEL
PROMPT_VALUE --> CHAIN
PROMPT_TPL -.继承.-> RUNNABLE
CHAT_TPL -.继承.-> RUNNABLE
PROMPT_TPL -.使用.-> CALLBACKS
CHAT_TPL -.使用.-> MESSAGES
style PromptsModule fill:#e1f5ff
style Templates fill:#fff4e1
style Output fill:#e8f5e9
style Downstream fill:#ffe1f5
系统架构说明
1. 图意概述
该架构图展示了Prompts模块的完整系统结构,从用户层的模板创建,经过模板层的处理和组合,到输出层生成PromptValue,最终传递给下游的语言模型。整个流程体现了"创建→处理→输出→传递"的完整链路。
2. 核心分层设计
创建层(Creation):
- 提供工厂方法(from_template、from_messages)
- 简化模板构造流程
- 自动推导输入变量
模板层(Templates):
- PromptTemplate: 基础字符串模板
- ChatPromptTemplate: 聊天模型专用模板
- FewShotPromptTemplate: Few-Shot学习模板
组件层(Components):
- MessagesPlaceholder: 动态消息插入
- MessagePromptTemplate: 单消息模板
- ImagePromptTemplate: 多模态支持
处理层(Processing):
- 变量提取: 从模板字符串中识别变量
- 格式化: 支持多种模板语法
- 验证: 确保模板和变量一致性
输出层(Output):
- PromptValue: 统一抽象
- StringPromptValue: 适配LLM
- ChatPromptValue: 适配ChatModel
3. 边界与约束
并发安全性:
- 模板实例不可变,线程安全
- 格式化操作无副作用,支持并发调用
超时控制:
- 模板格式化为同步操作,无超时机制
- 异步方法(aformat、ainvoke)支持取消
幂等性:
- 相同输入产生相同输出
- 模板实例可重复使用
顺序保证:
- ChatPromptTemplate保证消息顺序
- MessagesPlaceholder按位置展开
4. 异常与回退
模板语法错误:
- 构造时检查模板语法(如果validate_template=True)
- 格式化时抛出KeyError/ValueError
变量缺失:
- 格式化时抛出KeyError,提示缺失的变量名
- 可选变量(optional_variables)允许不传值
类型不匹配:
- MessagesPlaceholder要求list[BaseMessage]类型
- 自动转换支持的消息格式
5. 性能考量
变量提取缓存:
- 构造时提取一次,存储在input_variables
- 格式化时直接使用,无需重复解析
模板格式选择:
- f-string性能最优(Python原生)
- jinja2功能强大但性能较低
- mustache介于两者之间
内存占用:
- 模板字符串存储较小
- MessagesPlaceholder展开可能增加内存
批处理能力:
- batch方法支持批量格式化
- 减少函数调用开销
6. 版本兼容与演进
向后兼容:
- 保持公共API稳定
- 新增可选参数不破坏现有代码
废弃策略:
- 旧接口标记@deprecated
- 提供迁移指南
扩展点:
- 自定义格式化器(DEFAULT_FORMATTER_MAPPING)
- 自定义验证器(DEFAULT_VALIDATOR_MAPPING)
- 继承BasePromptTemplate实现新模板类型
模块类继承关系图
flowchart TB
subgraph Base["基类层"]
RUNNABLE[RunnableSerializable<br/>可运行基类]
BASE_PROMPT[BasePromptTemplate<br/>提示词模板基类]
STRING_PROMPT[StringPromptTemplate<br/>字符串模板基类]
BASE_CHAT[BaseChatPromptTemplate<br/>聊天模板基类]
BASE_MSG[BaseMessagePromptTemplate<br/>消息模板基类]
end
subgraph SimpleTemplates["简单模板"]
PROMPT_TEMPLATE[PromptTemplate<br/>基础字符串模板]
end
subgraph ChatTemplates["聊天模板"]
CHAT_PROMPT[ChatPromptTemplate<br/>聊天提示模板]
MSG_PLACEHOLDER[MessagesPlaceholder<br/>消息占位符]
HUMAN_TPL[HumanMessagePromptTemplate<br/>用户消息模板]
AI_TPL[AIMessagePromptTemplate<br/>AI消息模板]
SYSTEM_TPL[SystemMessagePromptTemplate<br/>系统消息模板]
end
subgraph FewShot["Few-Shot模板"]
FEW_SHOT[FewShotPromptTemplate<br/>示例提示模板]
FEW_SHOT_CHAT[FewShotChatMessagePromptTemplate<br/>聊天示例模板]
end
subgraph Specialized["特殊模板"]
IMAGE_TPL[ImagePromptTemplate<br/>图片模板]
DICT_TPL[DictPromptTemplate<br/>字典模板]
end
RUNNABLE <|-- BASE_PROMPT
BASE_PROMPT <|-- STRING_PROMPT
BASE_PROMPT <|-- BASE_CHAT
BASE_PROMPT <|-- IMAGE_TPL
BASE_PROMPT <|-- DICT_TPL
STRING_PROMPT <|-- PROMPT_TEMPLATE
STRING_PROMPT <|-- FEW_SHOT
BASE_CHAT <|-- CHAT_PROMPT
BASE_MSG <|-- MSG_PLACEHOLDER
BASE_MSG <|-- HUMAN_TPL
BASE_MSG <|-- AI_TPL
BASE_MSG <|-- SYSTEM_TPL
CHAT_PROMPT --> MSG_PLACEHOLDER
CHAT_PROMPT --> HUMAN_TPL
CHAT_PROMPT --> AI_TPL
CHAT_PROMPT --> SYSTEM_TPL
CHAT_PROMPT --> FEW_SHOT_CHAT
style BASE_PROMPT fill:#e1f5ff
style CHAT_PROMPT fill:#fff4e1
style PROMPT_TEMPLATE fill:#e8f5e9
类继承关系说明
基类层
RunnableSerializable:
- 所有Prompt模板的最顶层基类
- 提供invoke、ainvoke、batch等方法
- 支持管道操作符(|)
- 集成回调系统
BasePromptTemplate:
- 定义了所有模板的通用接口
- 核心方法: format、format_prompt、invoke
- 管理input_variables和partial_variables
- 提供验证和序列化功能
StringPromptTemplate:
- 字符串型模板的基类
- format返回str类型
- format_prompt返回StringPromptValue
BaseChatPromptTemplate:
- 聊天型模板的基类
- format_messages返回list[BaseMessage]
- format_prompt返回ChatPromptValue
模板类型
PromptTemplate: 最基础的字符串模板
- 支持f-string、jinja2、mustache语法
- 变量通过{}或{{}}插值
- 输出纯文本字符串
ChatPromptTemplate: 聊天模型专用模板
- 组合多个消息模板
- 支持MessagesPlaceholder动态插入对话历史
- 输出消息列表
FewShotPromptTemplate: Few-Shot学习模板
- 包含示例列表
- 自动格式化示例
- 适用于上下文学习
完整调用链路与时序图
场景1: PromptTemplate完整调用链路
从上游接口开始,展示用户调用PromptTemplate的完整流程。
sequenceDiagram
autonumber
participant User as 用户代码
participant PT as PromptTemplate
participant Extractor as get_template_variables
participant Validator as check_valid_template
participant Base as BasePromptTemplate
participant Formatter as DEFAULT_FORMATTER
participant PV as StringPromptValue
participant Model as ChatModel
Note over User,Model: 阶段1: 模板创建
User->>PT: from_template("Tell me a {adjective} joke")
activate PT
PT->>Extractor: 提取模板变量
activate Extractor
Extractor->>Extractor: 使用Formatter().parse()解析
Extractor-->>PT: ["adjective"]
deactivate Extractor
alt validate_template=True
PT->>Validator: 验证模板
activate Validator
Validator->>Validator: 检查变量一致性
Validator-->>PT: 验证通过
deactivate Validator
end
PT->>PT: 构造实例<br/>template="Tell me a {adjective} joke"<br/>input_variables=["adjective"]
PT-->>User: PromptTemplate实例
deactivate PT
Note over User,Model: 阶段2: 模板调用(invoke)
User->>PT: invoke({"adjective": "funny"})
activate PT
PT->>Base: invoke(input, config)
activate Base
Base->>Base: ensure_config(config)
Base->>Base: 注入metadata和tags
Base->>Base: _call_with_config(<br/> _format_prompt_with_error_handling,<br/> input, config, run_type="prompt"<br/>)
Note over Base: 回调系统启动
Base->>Base: _validate_input({"adjective": "funny"})
Base->>Base: 检查input_variables是否满足
Base->>PT: format_prompt(adjective="funny")
PT->>PT: format(adjective="funny")
activate PT
PT->>PT: _merge_partial_and_user_variables
PT->>PT: 合并partial_variables和kwargs
PT->>Formatter: DEFAULT_FORMATTER_MAPPING["f-string"]
activate Formatter
Formatter->>Formatter: template.format(adjective="funny")
Formatter-->>PT: "Tell me a funny joke"
deactivate Formatter
PT-->>Base: "Tell me a funny joke"
deactivate PT
Base->>PV: StringPromptValue(text="Tell me a funny joke")
activate PV
PV-->>Base: StringPromptValue实例
deactivate PV
Base-->>PT: PromptValue
deactivate Base
PT-->>User: PromptValue
deactivate PT
Note over User,Model: 阶段3: 传递给模型
User->>Model: prompt | model
activate Model
Model->>PV: to_messages()
activate PV
PV->>PV: 转换为[HumanMessage(content=text)]
PV-->>Model: [HumanMessage("Tell me a funny joke")]
deactivate PV
Model->>Model: _generate(messages)
Model-->>User: AIMessage(...)
deactivate Model
时序图详细说明
阶段1: 模板创建(步骤1-7)
- 用户调用from_template: 传入模板字符串
- 变量提取: 使用Formatter().parse()解析f-string,提取所有变量名
- 可选验证: 如果validate_template=True,检查模板语法和变量一致性
- 实例构造: 创建PromptTemplate实例,存储template和input_variables
关键代码解析:
@classmethod
def from_template(
cls,
template: str,
*,
template_format: str = "f-string",
partial_variables: dict[str, Any] | None = None,
**kwargs: Any,
) -> PromptTemplate:
# 1) 提取变量
input_variables = get_template_variables(template, template_format)
# get_template_variables内部:
# if template_format == "f-string":
# return {v for _, v, _, _ in Formatter().parse(template) if v is not None}
# 2) 移除partial_variables
if partial_variables:
input_variables = [
var for var in input_variables
if var not in partial_variables
]
# 3) 构造实例
return cls(
template=template,
input_variables=input_variables,
template_format=template_format,
partial_variables=partial_variables or {},
**kwargs,
)
边界条件:
- 模板语法错误会导致Formatter().parse()抛出ValueError
- 空模板字符串合法,input_variables为空列表
- 变量名不能为"stop"(内部保留)
阶段2: 模板调用(步骤8-20)
- 用户调用invoke: 传入变量字典 9-12. 配置处理: BasePromptTemplate.invoke确保config存在,注入metadata和tags 13-14. 输入验证: _validate_input检查所有必需变量是否提供
- 调用format_prompt: 委托给子类实现 16-19. 变量合并与格式化: 合并partial_variables和用户传入的kwargs,调用对应格式化器 20-22. 返回PromptValue: 包装为StringPromptValue
关键代码解析:
def invoke(
self, input: dict, config: RunnableConfig | None = None, **kwargs: Any
) -> PromptValue:
config = ensure_config(config)
# 注入元数据
if self.metadata:
config["metadata"] = {**config["metadata"], **self.metadata}
if self.tags:
config["tags"] += self.tags
# _call_with_config会:
# 1) 启动回调系统(on_prompt_start)
# 2) 调用_format_prompt_with_error_handling
# 3) 触发回调(on_prompt_end)
return self._call_with_config(
self._format_prompt_with_error_handling,
input,
config,
run_type="prompt",
serialized=self._serialized,
)
def _format_prompt_with_error_handling(self, inner_input: dict) -> PromptValue:
# 验证输入
inner_input_ = self._validate_input(inner_input)
# 调用format_prompt
return self.format_prompt(**inner_input_)
def format(self, **kwargs: Any) -> str:
# 合并变量
kwargs = self._merge_partial_and_user_variables(**kwargs)
# _merge_partial_and_user_variables内部:
# partial_kwargs = {
# k: v if not callable(v) else v()
# for k, v in self.partial_variables.items()
# }
# return {**partial_kwargs, **kwargs}
# 格式化
return DEFAULT_FORMATTER_MAPPING[self.template_format](self.template, **kwargs)
# 对于f-string: template.format(**kwargs)
边界条件:
- 缺少必需变量抛出KeyError,错误信息包含变量名
- partial_variables可以是callable,调用时才求值
- 格式化错误(如{var:.2f}传入非数字)抛出ValueError
性能考量:
- input_variables在构造时计算,invoke时不重复
- 回调系统有轻微开销(通常<1ms)
- f-string格式化极快(~µs级别)
阶段3: 传递给模型(步骤23-27)
23-24. 管道操作: prompt | model触发Runnable的__or__方法 25-27. 转换为消息: StringPromptValue.to_messages()转换为[HumanMessage]
关键代码解析:
# 管道操作符
class RunnableSerializable:
def __or__(self, other: Runnable) -> RunnableSequence:
# prompt | model 创建一个RunnableSequence
return RunnableSequence(first=self, last=other)
# StringPromptValue转换
class StringPromptValue(PromptValue):
def to_messages(self) -> list[BaseMessage]:
# 将文本包装为HumanMessage
return [HumanMessage(content=self.text)]
场景2: ChatPromptTemplate完整调用链路
展示带有MessagesPlaceholder的ChatPromptTemplate的完整流程。
sequenceDiagram
autonumber
participant User as 用户代码
participant CPT as ChatPromptTemplate
participant Conv as _convert_to_message_template
participant Validator as validate_input_variables
participant Base as BasePromptTemplate
participant Msg_TPL as MessagePromptTemplate
participant Placeholder as MessagesPlaceholder
participant Convert as convert_to_messages
participant PV as ChatPromptValue
participant Model as ChatModel
Note over User,Model: 阶段1: 模板创建
User->>CPT: from_messages([<br/> ("system", "You are {role}"),<br/> MessagesPlaceholder("history"),<br/> ("human", "{input}")<br/>])
activate CPT
loop 每个消息
CPT->>Conv: _convert_to_message_template(message)
activate Conv
alt message是tuple("system", "You are {role}")
Conv->>Conv: _create_template_from_message_type("system", template)
Conv->>Msg_TPL: SystemMessagePromptTemplate.from_template
activate Msg_TPL
Msg_TPL->>Msg_TPL: PromptTemplate.from_template("You are {role}")
Msg_TPL-->>Conv: SystemMessagePromptTemplate
deactivate Msg_TPL
else message是MessagesPlaceholder
Conv-->>CPT: MessagesPlaceholder(variable_name="history")
else message是tuple("human", "{input}")
Conv->>Msg_TPL: HumanMessagePromptTemplate.from_template
activate Msg_TPL
Msg_TPL->>Msg_TPL: PromptTemplate.from_template("{input}")
Msg_TPL-->>Conv: HumanMessagePromptTemplate
deactivate Msg_TPL
end
Conv-->>CPT: MessagePromptTemplate
deactivate Conv
end
CPT->>CPT: 收集所有input_variables:<br/>["role", "history", "input"]
CPT->>CPT: 识别optional_variables:<br/>如果MessagesPlaceholder.optional=True
CPT->>Validator: validate_input_variables
activate Validator
Validator->>Validator: 检查变量一致性
Validator-->>CPT: 验证通过
deactivate Validator
CPT->>CPT: 构造实例<br/>messages=[SystemMsgTpl, Placeholder, HumanMsgTpl]<br/>input_variables=["role", "history", "input"]
CPT-->>User: ChatPromptTemplate实例
deactivate CPT
Note over User,Model: 阶段2: 模板调用(invoke)
User->>CPT: invoke({<br/> "role": "assistant",<br/> "history": [("human", "Hi"), ("ai", "Hello!")],<br/> "input": "How are you?"<br/>})
activate CPT
CPT->>Base: invoke(input, config)
activate Base
Base->>Base: ensure_config(config)
Base->>Base: 注入metadata和tags
Base->>Base: _call_with_config(<br/> _format_prompt_with_error_handling,<br/> input, config, run_type="prompt"<br/>)
Note over Base: 回调系统启动
Base->>Base: _validate_input({<br/> "role": "assistant",<br/> "history": [...],<br/> "input": "How are you?"<br/>})
Base->>Base: 检查所有input_variables
Base->>CPT: format_prompt(**kwargs)
CPT->>CPT: format_messages(**kwargs)
activate CPT
CPT->>CPT: _merge_partial_and_user_variables
Note over CPT: 遍历messages列表
CPT->>Msg_TPL: SystemMessagePromptTemplate.format_messages(role="assistant")
activate Msg_TPL
Msg_TPL->>Msg_TPL: format(role="assistant")
Msg_TPL->>Msg_TPL: prompt.format(role="assistant")
Msg_TPL->>Msg_TPL: "You are assistant"
Msg_TPL-->>CPT: [SystemMessage("You are assistant")]
deactivate Msg_TPL
CPT->>Placeholder: MessagesPlaceholder.format_messages(history=[...])
activate Placeholder
Placeholder->>Placeholder: 获取kwargs["history"]
Placeholder->>Convert: convert_to_messages([("human", "Hi"), ("ai", "Hello!")])
activate Convert
Convert->>Convert: 转换tuple为BaseMessage
Convert-->>Placeholder: [HumanMessage("Hi"), AIMessage("Hello!")]
deactivate Convert
alt n_messages设置
Placeholder->>Placeholder: messages[-n_messages:]<br/>只保留最近N条
end
Placeholder-->>CPT: [HumanMessage("Hi"), AIMessage("Hello!")]
deactivate Placeholder
CPT->>Msg_TPL: HumanMessagePromptTemplate.format_messages(input="How are you?")
activate Msg_TPL
Msg_TPL->>Msg_TPL: format(input="How are you?")
Msg_TPL->>Msg_TPL: prompt.format(input="How are you?")
Msg_TPL->>Msg_TPL: "How are you?"
Msg_TPL-->>CPT: [HumanMessage("How are you?")]
deactivate Msg_TPL
CPT->>CPT: result.extend(messages)<br/>合并所有消息:<br/>[SystemMessage, HumanMessage, AIMessage, HumanMessage]
CPT-->>Base: messages列表
deactivate CPT
Base->>PV: ChatPromptValue(messages=[...])
activate PV
PV-->>Base: ChatPromptValue实例
deactivate PV
Base-->>CPT: PromptValue
deactivate Base
CPT-->>User: ChatPromptValue
deactivate CPT
Note over User,Model: 阶段3: 传递给模型
User->>Model: prompt | model
activate Model
Model->>PV: to_messages()
activate PV
PV->>PV: 直接返回messages列表
PV-->>Model: [SystemMessage, HumanMessage, AIMessage, HumanMessage]
deactivate PV
Model->>Model: _generate(messages)
Model-->>User: AIMessage(...)
deactivate Model
时序图详细说明
阶段1: 模板创建(步骤1-16)
1-2. 用户调用from_messages: 传入消息列表,支持多种格式 3-13. 消息转换循环: 将每个消息转换为MessagePromptTemplate
- tuple类型: _create_template_from_message_type根据角色创建对应模板
- MessagesPlaceholder: 直接使用
- BaseMessage: 包装为固定消息模板
- 变量收集: 遍历所有模板,收集input_variables
- 可选变量识别: 检查MessagesPlaceholder.optional
- 实例构造: 创建ChatPromptTemplate实例
关键代码解析:
def __init__(
self,
messages: Sequence[MessageLikeRepresentation],
*,
template_format: PromptTemplateFormat = "f-string",
**kwargs: Any,
) -> None:
# 1) 转换所有消息
messages_ = [
_convert_to_message_template(message, template_format)
for message in messages
]
# 2) 收集变量
input_vars: set[str] = set()
optional_variables: set[str] = set()
partial_vars: dict[str, Any] = {}
for _message in messages_:
# 可选占位符自动添加到partial_variables
if isinstance(_message, MessagesPlaceholder) and _message.optional:
partial_vars[_message.variable_name] = []
optional_variables.add(_message.variable_name)
elif isinstance(_message, (BaseChatPromptTemplate, BaseMessagePromptTemplate)):
input_vars.update(_message.input_variables)
# 3) 构造
kwargs = {
"input_variables": sorted(input_vars),
"optional_variables": sorted(optional_variables),
"partial_variables": partial_vars,
**kwargs,
}
super().__init__(messages=messages_, **kwargs)
def _convert_to_message_template(
message: MessageLikeRepresentation,
template_format: PromptTemplateFormat = "f-string",
) -> BaseMessage | BaseMessagePromptTemplate | BaseChatPromptTemplate:
# 支持的格式:
# 1. BaseMessagePromptTemplate: 直接使用
# 2. BaseMessage: 固定消息
# 3. tuple: ("role", "template")
# 4. str: 默认为human消息
# 5. dict: {"role": "...", "content": "..."}
if isinstance(message, (BaseMessagePromptTemplate, BaseChatPromptTemplate)):
return message
elif isinstance(message, BaseMessage):
return message
elif isinstance(message, str):
# 默认为HumanMessagePromptTemplate
return _create_template_from_message_type("human", message, template_format)
elif isinstance(message, (tuple, dict)):
if isinstance(message, dict):
message = (message["role"], message["content"])
message_type_str, template = message
return _create_template_from_message_type(
message_type_str, template, template_format
)
else:
raise NotImplementedError(f"Unsupported message type: {type(message)}")
def _create_template_from_message_type(
message_type: str,
template: str | list,
template_format: PromptTemplateFormat = "f-string",
) -> BaseMessagePromptTemplate:
# 根据消息类型创建对应模板
if message_type in {"human", "user"}:
return HumanMessagePromptTemplate.from_template(template, template_format)
elif message_type in {"ai", "assistant"}:
return AIMessagePromptTemplate.from_template(template, template_format)
elif message_type == "system":
return SystemMessagePromptTemplate.from_template(template, template_format)
elif message_type == "placeholder":
# 特殊处理: ("placeholder", "{variable_name}")
var_name = template[1:-1] # 去除{}
return MessagesPlaceholder(variable_name=var_name, optional=True)
else:
raise ValueError(f"Unexpected message type: {message_type}")
边界条件:
- 空消息列表合法,创建空模板
- 消息类型不区分大小写(human/Human/HUMAN)
- MessagesPlaceholder的variable_name必须存在于kwargs中(除非optional=True)
- 支持多个MessagesPlaceholder
阶段2: 模板调用(步骤17-47)
17-23. 配置与验证: 与PromptTemplate类似,确保配置和输入有效 24-30. 格式化第一个消息: SystemMessagePromptTemplate格式化 31-41. 格式化MessagesPlaceholder:
- 从kwargs获取history
- convert_to_messages转换为BaseMessage列表
- 可选的n_messages限制 42-47. 格式化剩余消息: HumanMessagePromptTemplate格式化
- 合并消息: extend合并所有消息到result列表 49-50. 包装为ChatPromptValue: 返回包含所有消息的PromptValue
关键代码解析:
def format_messages(self, **kwargs: Any) -> list[BaseMessage]:
# 1) 合并变量
kwargs = self._merge_partial_and_user_variables(**kwargs)
# 2) 遍历消息模板
result = []
for message_template in self.messages:
if isinstance(message_template, BaseMessage):
# 固定消息,直接添加
result.extend([message_template])
elif isinstance(message_template, (BaseMessagePromptTemplate, BaseChatPromptTemplate)):
# 模板消息,格式化后添加
message = message_template.format_messages(**kwargs)
result.extend(message)
else:
raise ValueError(f"Unexpected input: {message_template}")
return result
# MessagesPlaceholder实现
class MessagesPlaceholder(BaseMessagePromptTemplate):
def format_messages(self, **kwargs: Any) -> list[BaseMessage]:
# 1) 获取变量值
if self.variable_name in kwargs:
value = kwargs[self.variable_name]
elif self.optional:
# 可选占位符,返回空列表
return []
else:
raise KeyError(f"Missing required variable: {self.variable_name}")
# 2) 转换为消息列表
messages = convert_to_messages(value)
# convert_to_messages支持:
# - list[BaseMessage]: 直接返回
# - list[tuple]: [("human", "Hi"), ("ai", "Hello")]
# - list[dict]: [{"role": "human", "content": "Hi"}]
# 3) 限制消息数量
if self.n_messages is not None:
messages = messages[-self.n_messages:]
return messages
边界条件:
- MessagesPlaceholder的value必须是list类型
- convert_to_messages自动转换tuple/dict为BaseMessage
- n_messages从后向前截取(保留最新消息)
- optional=True时,缺失变量返回空列表
性能考量:
- 消息模板按顺序格式化,无法并行
- MessagesPlaceholder展开可能大幅增加消息数量
- ChatPromptValue不复制消息,使用引用
- 大量历史消息可能超过模型上下文长度
阶段3: 传递给模型(步骤48-53)
48-49. 管道操作: prompt | model创建RunnableSequence 50-53. 转换为消息: ChatPromptValue.to_messages()直接返回消息列表
关键代码解析:
class ChatPromptValue(PromptValue):
messages: Sequence[BaseMessage]
def to_messages(self) -> list[BaseMessage]:
# 直接返回消息列表,无需转换
return list(self.messages)
def to_string(self) -> str:
# 转换为字符串(用于LLM)
return get_buffer_string(self.messages)
# get_buffer_string格式:
# "System: You are assistant\nHuman: Hi\nAI: Hello!\nHuman: How are you?"
核心数据结构
classDiagram
class BasePromptTemplate {
<<abstract>>
+input_variables: list[str]
+partial_variables: dict
+metadata: dict|None
+tags: list[str]|None
+format(**kwargs) str
+format_prompt(**kwargs) PromptValue
+invoke(input) PromptValue
}
class PromptTemplate {
+template: str
+template_format: str
+validate_template: bool
+from_template(template: str) PromptTemplate
}
class ChatPromptTemplate {
+messages: list[MessageLike]
+from_messages(messages: list) ChatPromptTemplate
+from_template(template: str) ChatPromptTemplate
+format_messages(**kwargs) list[BaseMessage]
}
class MessagesPlaceholder {
+variable_name: str
+optional: bool
+n_messages: int|None
}
class FewShotPromptTemplate {
+examples: list[dict]
+example_prompt: PromptTemplate
+example_separator: str
+prefix: str
+suffix: str
}
BasePromptTemplate <|-- PromptTemplate
BasePromptTemplate <|-- ChatPromptTemplate
ChatPromptTemplate --> MessagesPlaceholder
BasePromptTemplate <|-- FewShotPromptTemplate
数据结构说明
BasePromptTemplate字段
| 字段 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
| input_variables | list[str] | 是 | [] | 模板中的变量名列表 |
| partial_variables | dict | 否 | {} | 部分变量,在构造时填充 |
| metadata | dict|None | 否 | None | 模板元数据 |
| tags | list[str]|None | 否 | None | 模板标签 |
PromptTemplate字段
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| template | str | - | 模板字符串 |
| template_format | str | “f-string” | 模板格式: “f-string"或"jinja2” |
| validate_template | bool | False | 是否验证模板语法 |
ChatPromptTemplate字段
| 字段 | 类型 | 说明 |
|---|---|---|
| messages | list | 消息模板列表 |
MessagesPlaceholder字段
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| variable_name | str | - | 占位符变量名 |
| optional | bool | False | 是否可选(允许不传值) |
| n_messages | int|None | None | 保留最近N条消息 |
核心API详解
API-1: PromptTemplate.from_template
基本信息
- 名称:
from_template - 方法签名:
@classmethod def from_template(cls, template: str, *, template_format: str = "f-string", partial_variables: dict | None = None, **kwargs) -> PromptTemplate - 幂等性: 幂等
功能说明
从模板字符串创建PromptTemplate实例,自动推导输入变量。
请求结构体
template: str # 模板字符串
template_format: str # 模板格式
partial_variables: dict | None # 部分变量
| 参数 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
| template | str | 是 | - | 包含变量的模板字符串 |
| template_format | str | 否 | “f-string” | “f-string"或"jinja2” |
| partial_variables | dict|None | 否 | None | 预填充的变量 |
响应结构体
PromptTemplate # 提示词模板实例
入口函数与关键代码
class PromptTemplate(StringPromptTemplate):
@classmethod
def from_template(
cls,
template: str,
*,
template_format: str = "f-string",
partial_variables: dict | None = None,
**kwargs: Any,
) -> PromptTemplate:
"""从模板字符串创建提示词模板
Args:
template: 模板字符串,如"Tell me a {adjective} joke about {content}"
template_format: 模板格式,"f-string"或"jinja2"
partial_variables: 部分变量字典
Returns:
PromptTemplate实例
"""
# 1) 提取模板中的变量
if template_format == "f-string":
input_variables = get_template_variables(template, template_format)
elif template_format == "jinja2":
input_variables = get_template_variables(template, template_format)
else:
msg = f"Unsupported template format: {template_format}"
raise ValueError(msg)
# 2) 从input_variables中移除partial_variables
if partial_variables:
input_variables = [
var for var in input_variables
if var not in partial_variables
]
# 3) 构造PromptTemplate实例
return cls(
template=template,
input_variables=input_variables,
template_format=template_format,
partial_variables=partial_variables or {},
**kwargs,
)
def format(self, **kwargs: Any) -> str:
"""格式化模板
Args:
**kwargs: 变量值
Returns:
格式化后的字符串
"""
# 1) 合并partial_variables和kwargs
all_kwargs = {**self.partial_variables, **kwargs}
# 2) 根据template_format格式化
if self.template_format == "f-string":
return self.template.format(**all_kwargs)
elif self.template_format == "jinja2":
return self._render_jinja2(self.template, all_kwargs)
代码说明:
- from_template自动从模板字符串提取变量名
- 支持f-string(默认)和jinja2两种格式
- partial_variables预填充部分变量
- format方法合并变量并格式化输出
调用链与上层函数
# 基础使用
prompt = PromptTemplate.from_template(
"Tell me a {adjective} joke about {content}"
)
result = prompt.format(adjective="funny", content="chickens")
# "Tell me a funny joke about chickens"
# 使用partial变量
prompt = PromptTemplate.from_template(
"Tell me a {adjective} joke about {content}",
partial_variables={"adjective": "funny"}
)
result = prompt.format(content="chickens")
# "Tell me a funny joke about chickens"
# Jinja2模板
prompt = PromptTemplate.from_template(
"Tell me a {{ adjective }} joke about {{ content }}",
template_format="jinja2"
)
# 在链中使用
chain = prompt | model | parser
result = chain.invoke({"adjective": "funny", "content": "chickens"})
时序图
sequenceDiagram
autonumber
participant User
participant PromptTemplate
participant Parser as get_template_variables
User->>PromptTemplate: from_template("Tell me a {adjective} joke")
PromptTemplate->>Parser: 提取变量
Parser-->>PromptTemplate: ["adjective"]
PromptTemplate->>PromptTemplate: 构造实例
PromptTemplate-->>User: PromptTemplate实例
User->>PromptTemplate: format(adjective="funny")
PromptTemplate->>PromptTemplate: 合并partial_variables
PromptTemplate->>PromptTemplate: template.format(**kwargs)
PromptTemplate-->>User: "Tell me a funny joke"
时序图说明:
图意: 展示从模板字符串创建PromptTemplate并格式化的过程。
边界条件:
- 模板语法错误会在构造或格式化时抛出异常
- 缺少必需变量时抛出KeyError
- partial_variables优先级低于format传入的kwargs
性能要点:
- 变量提取在构造时完成,format时不重复
- f-string格式化性能优于jinja2
- 模板实例可复用,避免重复构造
API-2: ChatPromptTemplate.from_messages
基本信息
- 名称:
from_messages - 方法签名:
@classmethod def from_messages(cls, messages: Sequence[MessageLikeRepresentation]) -> ChatPromptTemplate - 幂等性: 幂等
功能说明
从消息列表创建聊天提示模板。
请求结构体
messages: Sequence[MessageLikeRepresentation]
# MessageLikeRepresentation = tuple | str | BaseMessage | MessagePromptTemplate
支持的消息格式:
- 元组:
("system", "You are {role}") - 字符串:
"What is {topic}?"(默认HumanMessage) - BaseMessage实例
- MessagePromptTemplate实例
- MessagesPlaceholder: 动态插入对话历史
响应结构体
ChatPromptTemplate # 聊天提示模板实例
入口函数与关键代码
class ChatPromptTemplate(BaseChatPromptTemplate):
@classmethod
def from_messages(
cls,
messages: Sequence[MessageLikeRepresentation],
) -> ChatPromptTemplate:
"""从消息列表创建聊天提示模板
Args:
messages: 消息列表,支持多种格式
Returns:
ChatPromptTemplate实例
"""
# 1) 转换每个消息为MessagePromptTemplate
message_templates = []
for message in messages:
message_template = _convert_to_message_template(message)
message_templates.append(message_template)
# 2) 提取所有输入变量
input_variables = set()
for template in message_templates:
input_variables.update(template.input_variables)
# 3) 构造实例
return cls(
messages=message_templates,
input_variables=list(input_variables),
)
def format_messages(self, **kwargs: Any) -> list[BaseMessage]:
"""格式化为消息列表
Args:
**kwargs: 变量值
Returns:
BaseMessage列表
"""
# 1) 合并partial_variables和kwargs
all_kwargs = {**self.partial_variables, **kwargs}
# 2) 格式化每个消息模板
result = []
for message_template in self.messages:
messages = message_template.format_messages(**all_kwargs)
result.extend(messages)
return result
def _convert_to_message_template(message: MessageLikeRepresentation):
"""转换为MessagePromptTemplate"""
if isinstance(message, BaseMessagePromptTemplate):
return message
elif isinstance(message, BaseMessage):
# 固定消息,无变量
return FixedMessagePromptTemplate(message=message)
elif isinstance(message, tuple) and len(message) == 2:
# ("system", "You are {role}") → SystemMessagePromptTemplate
role, content = message
return _role_to_template(role, content)
elif isinstance(message, str):
# "What is {topic}?" → HumanMessagePromptTemplate
return HumanMessagePromptTemplate.from_template(message)
else:
msg = f"Unsupported message type: {type(message)}"
raise ValueError(msg)
def _role_to_template(role: str, content: str):
"""根据角色创建对应模板"""
role = role.lower()
if role in ("system",):
return SystemMessagePromptTemplate.from_template(content)
elif role in ("human", "user"):
return HumanMessagePromptTemplate.from_template(content)
elif role in ("ai", "assistant"):
return AIMessagePromptTemplate.from_template(content)
elif role == "placeholder":
return MessagesPlaceholder(variable_name=content)
else:
return ChatMessagePromptTemplate.from_template(content, role=role)
代码说明:
- from_messages支持多种消息格式
- 自动转换为MessagePromptTemplate
- 汇总所有模板的input_variables
- format_messages遍历模板并格式化
调用链与上层函数
# 基础使用
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
("human", "Hello, my name is {name}"),
("ai", "Hello {name}, nice to meet you!"),
("human", "{question}"),
])
messages = prompt.format_messages(name="Alice", question="What is AI?")
# [
# SystemMessage("You are a helpful assistant"),
# HumanMessage("Hello, my name is Alice"),
# AIMessage("Hello Alice, nice to meet you!"),
# HumanMessage("What is AI?"),
# ]
# 使用MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("history"),
("human", "{input}"),
])
messages = prompt.format_messages(
history=[
("human", "Hi"),
("ai", "Hello!"),
],
input="What is 2+2?"
)
# 在链中使用
chain = prompt | model | parser
result = chain.invoke({"name": "Bob", "question": "Explain quantum computing"})
时序图
sequenceDiagram
autonumber
participant User
participant ChatPrompt as ChatPromptTemplate
participant Convert as _convert_to_message_template
participant Model as ChatModel
User->>ChatPrompt: from_messages([<br/> ("system", "You are {role}"),<br/> ("human", "{input}")<br/>])
loop 每个消息
ChatPrompt->>Convert: 转换消息格式
Convert-->>ChatPrompt: MessagePromptTemplate
end
ChatPrompt-->>User: ChatPromptTemplate实例
User->>ChatPrompt: format_messages(role="assistant", input="Hi")
loop 每个模板
ChatPrompt->>ChatPrompt: 格式化模板
end
ChatPrompt-->>User: [SystemMessage(...), HumanMessage(...)]
User->>Model: invoke(messages)
Model-->>User: AIMessage(...)
时序图说明:
图意: 展示ChatPromptTemplate如何组合多个消息模板并格式化。
边界条件:
- MessagesPlaceholder要求对应变量是消息列表
- 可选的MessagesPlaceholder允许不传值
- 所有非可选变量必须提供
性能要点:
- 模板转换在构造时完成
- format_messages按顺序处理,不支持并行
- MessagesPlaceholder展开可能增加消息数量
API-3: MessagesPlaceholder
基本信息
- 名称:
MessagesPlaceholder - 构造签名:
def __init__(self, variable_name: str, *, optional: bool = False, n_messages: int | None = None) - 幂等性: 幂等
功能说明
在聊天模板中插入动态消息列表,常用于对话历史。
请求结构体
variable_name: str # 占位符变量名
optional: bool # 是否可选
n_messages: int | None # 保留最近N条消息
| 参数 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
| variable_name | str | 是 | - | 占位符对应的变量名 |
| optional | bool | 否 | False | True时允许不传该变量 |
| n_messages | int|None | 否 | None | 只保留最近N条消息 |
响应结构体
MessagesPlaceholder # 消息占位符实例
入口函数与关键代码
class MessagesPlaceholder(BaseMessagePromptTemplate):
def format_messages(self, **kwargs: Any) -> list[BaseMessage]:
"""格式化占位符为消息列表
Args:
**kwargs: 必须包含variable_name对应的消息列表
Returns:
消息列表
"""
# 1) 获取变量值
if self.variable_name in kwargs:
value = kwargs[self.variable_name]
elif self.optional:
# 可选占位符,返回空列表
return []
else:
msg = f"Missing required variable: {self.variable_name}"
raise KeyError(msg)
# 2) 转换为消息列表
messages = convert_to_messages(value)
# 3) 限制消息数量
if self.n_messages is not None:
messages = messages[-self.n_messages:]
return messages
代码说明:
- 从kwargs获取对应变量
- optional=True时允许变量缺失
- n_messages限制保留的消息数量
- 自动转换多种格式为消息列表
调用链与上层函数
# 基础使用
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
# 传入对话历史
messages = prompt.format_messages(
chat_history=[
("human", "What's 2+2?"),
("ai", "2+2 equals 4"),
("human", "What's 3+3?"),
("ai", "3+3 equals 6"),
],
input="Now what's 5+5?"
)
# 可选占位符
optional_prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
])
# 不传chat_history也不会报错
messages = optional_prompt.format_messages(input="Hello")
# 限制历史长度
limited_prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("chat_history", n_messages=2), # 只保留最近2条
("human", "{input}"),
])
API-4: FewShotPromptTemplate
基本信息
- 名称:
FewShotPromptTemplate - 构造签名: 包含examples、example_prompt等参数
- 幂等性: 幂等
功能说明
Few-Shot学习模板,自动格式化示例列表。
请求结构体
examples: list[dict] # 示例列表
example_prompt: PromptTemplate # 单个示例的格式模板
example_separator: str # 示例之间的分隔符
prefix: str # 前缀
suffix: str # 后缀
input_variables: list[str] # 输入变量
响应结构体
FewShotPromptTemplate # Few-Shot模板实例
入口函数与关键代码
class FewShotPromptTemplate(StringPromptTemplate):
examples: list[dict]
example_prompt: PromptTemplate
example_separator: str = "\n\n"
prefix: str = ""
suffix: str = ""
def format(self, **kwargs: Any) -> str:
"""格式化Few-Shot模板
Args:
**kwargs: 变量值
Returns:
格式化后的字符串
"""
# 1) 格式化前缀
result = [self.prefix.format(**kwargs) if self.prefix else ""]
# 2) 格式化每个示例
for example in self.examples:
example_str = self.example_prompt.format(**example)
result.append(example_str)
# 3) 格式化后缀
if self.suffix:
result.append(self.suffix.format(**kwargs))
# 4) 用分隔符连接
return self.example_separator.join([r for r in result if r])
代码说明:
- 遍历示例列表并格式化
- 使用example_prompt统一格式
- example_separator连接示例
- prefix和suffix包裹示例
调用链与上层函数
# 定义示例
examples = [
{"input": "What's 2+2?", "output": "4"},
{"input": "What's 3+3?", "output": "6"},
{"input": "What's 4+4?", "output": "8"},
]
# 定义示例格式
example_prompt = PromptTemplate.from_template(
"Input: {input}\nOutput: {output}"
)
# 创建Few-Shot模板
prompt = FewShotPromptTemplate(
examples=examples,
example_prompt=example_prompt,
prefix="Answer the following questions:",
suffix="Input: {input}\nOutput:",
input_variables=["input"],
)
# 使用
result = prompt.format(input="What's 5+5?")
# Answer the following questions:
#
# Input: What's 2+2?
# Output: 4
#
# Input: What's 3+3?
# Output: 6
#
# Input: What's 4+4?
# Output: 8
#
# Input: What's 5+5?
# Output:
典型使用场景
场景1: 角色扮演提示词
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的{domain}专家。
你的特点:
- 知识渊博,经验丰富
- 表达清晰,逻辑严谨
- 善于举例说明
请用专业但通俗易懂的方式回答问题。"""),
MessagesPlaceholder("examples", optional=True),
("human", "{question}"),
])
# 使用
response = (prompt | model).invoke({
"domain": "机器学习",
"question": "什么是梯度下降?"
})
场景2: 多轮对话管理
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
# 维护对话历史
chat_history = []
def chat(user_input: str):
messages = prompt.format_messages(
chat_history=chat_history,
input=user_input
)
response = model.invoke(messages)
# 更新历史
chat_history.append(HumanMessage(user_input))
chat_history.append(response)
return response.content
# 对话
chat("Hi, I'm Alice")
chat("What's my name?") # 记住前面的对话
场景3: 动态Few-Shot
from langchain_core.prompts import FewShotChatMessagePromptTemplate
# 动态选择示例
example_selector = SemanticSimilarityExampleSelector.from_examples(
examples,
OpenAIEmbeddings(),
Chroma,
k=2 # 选择最相关的2个示例
)
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_selector=example_selector,
example_prompt=ChatPromptTemplate.from_messages([
("human", "{input}"),
("ai", "{output}"),
]),
)
final_prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
few_shot_prompt,
("human", "{input}"),
])
最佳实践
1. 模板组织
# 推荐: 将模板定义为常量
SYSTEM_PROMPT = "You are a {role} expert"
USER_PROMPT = "Answer this question: {question}"
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
("human", USER_PROMPT),
])
# 推荐: 模板复用
base_system = "You are a helpful assistant"
qa_prompt = ChatPromptTemplate.from_messages([
("system", base_system),
("human", "Question: {question}"),
])
chat_prompt = ChatPromptTemplate.from_messages([
("system", base_system),
MessagesPlaceholder("history"),
("human", "{input}"),
])
2. 变量验证
# 使用partial固定某些变量
prompt = ChatPromptTemplate.from_messages([
("system", "You are a {role} expert. Today is {date}"),
("human", "{question}"),
])
# 固定date
from datetime import datetime
prompt_with_date = prompt.partial(date=datetime.now().strftime("%Y-%m-%d"))
# 只需传role和question
prompt_with_date.invoke({"role": "AI", "question": "..."})
3. 提示词工程
# 清晰的指令
prompt = ChatPromptTemplate.from_template("""
根据以下文本回答问题。
文本:
{context}
要求:
1. 只使用文本中的信息
2. 如果文本中没有答案,说"我不知道"
3. 引用原文支持你的答案
问题: {question}
答案:""")
# Few-Shot增强
few_shot_examples = [...] # 高质量示例
prompt = FewShotPromptTemplate(
examples=few_shot_examples,
example_prompt=example_template,
suffix="Now answer: {question}",
input_variables=["question"],
)
模块交互矩阵
Prompts模块与其他模块的交互关系
flowchart LR
subgraph Upstream["上游模块(调用方)"]
APP[应用代码]
CHAINS[Chains模块]
AGENTS[Agents模块]
end
subgraph PromptsCore["Prompts模块核心"]
PT[PromptTemplate]
CPT[ChatPromptTemplate]
FST[FewShotPromptTemplate]
end
subgraph Downstream["下游模块(被调用方)"]
LLM[LLM模型]
CHAT[ChatModel]
EMBEDDINGS[Embeddings]
end
subgraph Infrastructure["基础设施模块"]
RUNNABLE[Runnables]
MESSAGES[Messages]
CALLBACKS[Callbacks]
PV[PromptValues]
end
APP -->|创建模板| PT
APP -->|创建模板| CPT
CHAINS -->|链式调用| PT
CHAINS -->|链式调用| CPT
AGENTS -->|构造提示| CPT
AGENTS -->|格式化指令| FST
PT -->|继承| RUNNABLE
CPT -->|继承| RUNNABLE
FST -->|继承| RUNNABLE
PT -->|生成| PV
CPT -->|生成| PV
CPT -->|使用| MESSAGES
PT -->|通过Runnable| LLM
CPT -->|通过Runnable| CHAT
PT -->|格式化文档| EMBEDDINGS
PT -.触发.-> CALLBACKS
CPT -.触发.-> CALLBACKS
style PromptsCore fill:#e1f5ff
style Infrastructure fill:#fff4e1
交互矩阵详表
| 调用方 | Prompts模块 | 交互方式 | 数据格式 | 同步/异步 | 错误语义 | 一致性要求 |
|---|---|---|---|---|---|---|
| 应用代码 | PromptTemplate | from_template() | 模板字符串 | 同步 | ValueError(语法错误) | N/A |
| 应用代码 | ChatPromptTemplate | from_messages() | 消息列表 | 同步 | ValueError(格式错误) | N/A |
| Chains | BasePromptTemplate | invoke() | dict | 同步/异步 | KeyError(变量缺失) | 幂等 |
| Agents | ChatPromptTemplate | format_messages() | dict | 同步/异步 | KeyError(变量缺失) | 幂等 |
| Prompts | LLM | 通过管道(|) | PromptValue | 同步/异步 | ModelError | 无状态 |
| Prompts | ChatModel | 通过管道(|) | PromptValue | 同步/异步 | ModelError | 无状态 |
| Prompts | Callbacks | 自动触发 | 回调事件 | 同步/异步 | 忽略异常 | 无要求 |
| Prompts | Messages | 创建消息 | BaseMessage | 同步 | TypeError | 无状态 |
交互详解
1. 应用代码 → Prompts模块
场景: 创建提示词模板
调用链路:
# PromptTemplate
app_code → PromptTemplate.from_template()
→ get_template_variables() # 提取变量
→ __init__() # 构造实例
# ChatPromptTemplate
app_code → ChatPromptTemplate.from_messages()
→ _convert_to_message_template() # 转换消息
→ __init__() # 构造实例
数据流:
- 输入: 模板字符串或消息列表
- 输出: 模板实例
- 错误: ValueError(模板语法错误)
一致性: 无状态操作,线程安全
2. Chains/Agents → Prompts模块
场景: 链式调用中格式化提示词
调用链路:
chain/agent → prompt.invoke(input_dict)
→ BasePromptTemplate.invoke()
→ _call_with_config() # 启动回调
→ _format_prompt_with_error_handling()
→ _validate_input() # 验证输入
→ format_prompt() # 格式化
→ format() # 子类实现
→ 返回PromptValue
数据流:
- 输入: dict(变量字典)
- 输出: PromptValue
- 错误: KeyError(变量缺失)
一致性: 幂等操作,相同输入产生相同输出
3. Prompts模块 → 语言模型
场景: 通过管道传递提示词到模型
调用链路:
prompt | model
→ RunnableSequence.invoke()
→ prompt.invoke() # 生成PromptValue
→ prompt_value.to_messages() # 转换为消息
→ model.invoke(messages) # 调用模型
数据流:
- 输入: PromptValue
- 中间: list[BaseMessage]
- 输出: AIMessage
- 错误: ModelError(模型异常)
一致性: 无状态,每次调用独立
4. Prompts模块 → Callbacks系统
场景: 自动记录提示词格式化事件
调用链路:
prompt.invoke()
→ _call_with_config()
→ on_prompt_start(serialized, inputs) # 开始回调
→ _format_prompt_with_error_handling() # 执行格式化
→ on_prompt_end(result) # 结束回调
数据流:
- 事件: prompt_start, prompt_end
- 数据: serialized(模板序列化), inputs(输入), result(输出)
- 错误处理: 回调异常不影响主流程
一致性: 无要求,回调失败不影响格式化
性能分析与优化建议
性能瓶颈识别
1. 变量提取开销
场景: from_template构造时
瓶颈:
- f-string: Formatter().parse() ~10µs
- jinja2: Environment().parse() ~100µs
- mustache: tokenize() ~50µs
优化:
- 变量提取结果缓存在input_variables,避免重复解析
- 优先使用f-string格式
建议:
# 推荐: 使用f-string(默认)
prompt = PromptTemplate.from_template("Tell me a {adjective} joke")
# 避免: jinja2(除非需要复杂逻辑)
prompt = PromptTemplate.from_template(
"Tell me a {{ adjective }} joke",
template_format="jinja2"
)
2. 消息模板格式化
场景: ChatPromptTemplate.format_messages()
瓶颈:
- 顺序遍历所有消息模板 O(n)
- MessagesPlaceholder展开可能产生大量消息
- 无法并行格式化
优化:
- 减少消息模板数量
- 使用n_messages限制历史长度
- 批量格式化使用batch()方法
建议:
# 推荐: 限制历史长度
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("history", n_messages=10), # 只保留最近10条
("human", "{input}"),
])
# 避免: 无限制历史
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("history"), # 可能展开数千条消息
("human", "{input}"),
])
3. 回调系统开销
场景: invoke()时自动触发回调
瓶颈:
- 每次调用触发on_prompt_start/end
- 回调处理器序列化数据 ~1ms
- 多个回调处理器累积延迟
优化:
- 生产环境禁用不必要的回调
- 使用异步回调减少阻塞
建议:
# 生产环境: 禁用详细回调
from langchain_core.callbacks import StdOutCallbackHandler
# 避免: 过多回调处理器
config = {"callbacks": [StdOutCallbackHandler(), ...]} # 多个处理器
# 推荐: 最小化回调
config = {"callbacks": []} # 或只保留必要的
性能优化策略
1. 模板复用
原理: 模板实例不可变,可安全复用
示例:
# 推荐: 全局单例
SYSTEM_PROMPT = ChatPromptTemplate.from_messages([
("system", "You are a {role} expert"),
("human", "{question}"),
])
# 多次使用
result1 = SYSTEM_PROMPT.invoke({"role": "AI", "question": "..."})
result2 = SYSTEM_PROMPT.invoke({"role": "ML", "question": "..."})
# 避免: 重复创建
def get_prompt():
return ChatPromptTemplate.from_messages([...]) # 每次都创建
收益: 减少构造开销 ~100µs/次
2. 批量格式化
原理: batch()方法减少函数调用开销
示例:
# 推荐: 批量处理
inputs = [
{"adjective": "funny"},
{"adjective": "sad"},
{"adjective": "clever"},
]
results = prompt.batch(inputs) # 一次调用
# 避免: 循环调用
results = [prompt.invoke(inp) for inp in inputs] # 多次调用
收益: 减少回调开销和函数调用 ~30%
3. 部分变量优化
原理: partial()预填充固定变量
示例:
# 基础模板
base_prompt = PromptTemplate.from_template(
"You are a {role} expert. Today is {date}. {question}"
)
# 推荐: 固定不变的变量
from datetime import datetime
daily_prompt = base_prompt.partial(
date=datetime.now().strftime("%Y-%m-%d")
)
# 只需传question
result = daily_prompt.invoke({"role": "AI", "question": "..."})
# 避免: 每次都传所有变量
result = base_prompt.invoke({
"role": "AI",
"date": datetime.now().strftime("%Y-%m-%d"), # 重复计算
"question": "..."
})
收益: 减少参数传递和重复计算
4. 消息历史管理
策略: 智能截断和压缩
示例:
from langchain_core.messages import trim_messages
# 推荐: 智能截断
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
MessagesPlaceholder("history", n_messages=20), # 硬限制
("human", "{input}"),
])
# 或使用trim_messages
history = trim_messages(
full_history,
max_tokens=1000, # 基于token数截断
strategy="last", # 保留最近的
)
# 避免: 传入完整历史
prompt.invoke({"history": full_history}) # 可能数千条
收益: 减少格式化时间和模型成本
性能监控
关键指标
| 指标 | 正常范围 | 警告阈值 | 监控方法 |
|---|---|---|---|
| 模板创建时间 | <1ms | >10ms | 计时装饰器 |
| 格式化时间 | <5ms | >50ms | 回调统计 |
| 消息数量 | <50条 | >100条 | len(messages) |
| 输入变量数 | <10个 | >20个 | len(input_variables) |
| 模板实例数 | 全局<100 | >1000 | 弱引用计数 |
监控示例
import time
from langchain_core.callbacks import BaseCallbackHandler
class PerformanceMonitor(BaseCallbackHandler):
def on_prompt_start(self, serialized, inputs, **kwargs):
self.start_time = time.time()
def on_prompt_end(self, result, **kwargs):
duration = time.time() - self.start_time
if duration > 0.05: # 50ms警告
print(f"⚠️ Slow prompt formatting: {duration*1000:.2f}ms")
# 使用
config = {"callbacks": [PerformanceMonitor()]}
result = prompt.invoke(input_dict, config=config)
内存优化
1. 避免内存泄漏
问题: 模板实例引用导致内存泄漏
解决:
# 避免: 闭包捕获模板
def create_handler():
prompt = ChatPromptTemplate.from_messages([...])
def handler(input):
return prompt.invoke(input) # 闭包捕获prompt
return handler # prompt永不释放
# 推荐: 全局单例或参数传递
PROMPT = ChatPromptTemplate.from_messages([...])
def handler(input, prompt=PROMPT):
return prompt.invoke(input)
2. 消息对象复用
原理: BaseMessage不可变,可安全复用
示例:
# 推荐: 复用固定消息
SYSTEM_MSG = SystemMessage("You are a helpful assistant")
prompt = ChatPromptTemplate.from_messages([
SYSTEM_MSG, # 复用
("human", "{input}"),
])
# 避免: 每次创建新消息
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"), # 每次格式化都创建
("human", "{input}"),
])
常见问题与最佳实践
问题1: 模板变量冲突
症状: KeyError或输出错误
原因: 变量名重复或与保留字冲突
解决:
# 避免: 使用保留字"stop"
prompt = PromptTemplate.from_template("Stop at {stop}") # 报错!
# 推荐: 使用其他名称
prompt = PromptTemplate.from_template("Stop at {position}")
# 避免: 变量名重复
prompt = ChatPromptTemplate.from_messages([
("system", "Role: {role}"),
("human", "Your role is {role}"), # 重复,但合法
])
# 推荐: 使用描述性名称
prompt = ChatPromptTemplate.from_messages([
("system", "Role: {system_role}"),
("human", "Your role is {user_role}"),
])
问题2: MessagesPlaceholder展开失败
症状: ValueError或TypeError
原因: 传入的不是消息列表
解决:
# 避免: 传入非列表
prompt.invoke({"history": "some text"}) # 报错!
# 推荐: 确保是列表
prompt.invoke({"history": []}) # 空列表也可以
# 或使用optional
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder("history", optional=True), # 允许缺失
])
prompt.invoke({}) # 不传history也可以
问题3: 模板格式化性能差
症状: format_messages耗时>50ms
诊断步骤:
- 检查消息数量: len(prompt.messages)
- 检查历史长度: len(history)
- 检查模板复杂度: 是否使用jinja2
解决:
# 1. 减少消息模板数量
# 避免: 每个角色一个模板
for example in examples:
prompt.append(("human", example["input"]))
prompt.append(("ai", example["output"]))
# 推荐: 使用FewShotPromptTemplate
few_shot = FewShotPromptTemplate(examples=examples, ...)
# 2. 限制历史长度
MessagesPlaceholder("history", n_messages=10)
# 3. 使用f-string而非jinja2
# 避免: template_format="jinja2"
# 推荐: template_format="f-string"(默认)
最佳实践总结
1. 模板设计
✅ 推荐:
- 使用描述性变量名
- 模板单一职责
- 合理使用optional参数
- 限制历史长度(n_messages)
❌ 避免:
- 过长的模板字符串
- 过多的嵌套变量
- 未限制的MessagesPlaceholder
- 使用保留字作为变量名
2. 性能优化
✅ 推荐:
- 全局单例模板
- 使用batch()批量处理
- partial()固定不变的变量
- f-string优先于jinja2
❌ 避免:
- 重复创建模板实例
- 循环调用invoke()
- 传入完整历史消息
- 过多回调处理器
3. 错误处理
✅ 推荐:
- 使用optional参数
- 验证输入变量
- 捕获KeyError
- 提供默认值
❌ 避免:
- 忽略异常
- 假设变量存在
- 裸except语句
- 吞掉错误信息