5.PromptTemplate #

PromptTemplate 是一个用于管理和格式化文本模板的类，特别适合于需要“占位变量”的自然语言生成任务（如智能对话、提示工程等场景）。它的核心作用是：

1. 变量提取：自动识别模板字符串中的 {变量名} 或 {变量名:值} 结构，并提取所有变量名。变量名通过正则表达式识别并去重，保证不会重复。

2. 自动校验：在填充模板前，format() 方法会检查所有模板中出现的变量名是否都被赋值，如果缺少任何变量会直接抛出异常，避免生成内容时遗漏关键信息。

3. 灵活格式化：根据用户传入的变量，对模板进行格式化，生成最终填充后的字符串，可用于继续传递给大模型或其他处理环节。

典型用法示例

比如有一个模板字符串 "你好,我叫{name},你是谁?"，可以写成：

prompt_template = PromptTemplate.from_template("你好,我叫{name},你是谁?")
filled_prompt = prompt_template.format(name="张三")
print(filled_prompt)
# 输出：你好,我叫张三,你是谁?

变量提取机制

模板内的变量解析由如下正则表达式实现：

r'\{([^}:]+)(?::[^}]+)?\}'

它可以识别 {变量} 及 {变量:默认值} 形式。例如：

• {name} 会提取到 name
• {city:北京} 会提取到 city
• {age:25} 会提取到 age

使用注意事项

• 所有模板变量都必须以 format 的参数方式传入，否则会报错提示缺少的变量名。
• 支持带“冒号默认值”格式的变量，但本类不处理默认值逻辑，只提取冒号前的 key 名。
• 主要适用于中文、英文等自然语言模板的变量填充需求。

5.1. 5.PromptTemplate.py #

5.PromptTemplate.py

#from langchain_core.prompts import PromptTemplate
#from langchain_openai import ChatOpenAI

# 从 smartchain.chat_models 导入 ChatOpenAI 类
from smartchain.chat_models import ChatOpenAI
# 从 smartchain.prompts 导入 PromptTemplate 类
from smartchain.prompts import PromptTemplate

# 通过模板字符串创建一个 PromptTemplate 实例
prompt_template = PromptTemplate.from_template("你好,我叫{name},你是谁?")

# 打印 prompt_template 对象，以及它的类型类型
print(prompt_template, type(prompt_template))

# 创建一个 ChatOpenAI 实例，指定所用的模型为 "gpt-4o"
llm = ChatOpenAI(model="gpt-4o")

# 使用 prompt_template 的 format 方法，将 name 变量替换为"张三"，并传递给 llm.invoke 方法生成回复
result = llm.invoke(prompt_template.format(name="张三"))

# 输出 AI 回复的内容
print(result.content)

5.2. prompts.py #

smartchain/prompts.py

# 导入正则表达式模块，用于变量提取
import re

# 定义提示词模板类
class PromptTemplate:
    # 类说明文档，描述用途
    """提示词模板类，用于格式化字符串模板"""

    # 构造方法，初始化模板实例
    def __init__(self, template: str):
        # 保存模板字符串到实例属性
        self.template = template
        # 调用内部方法提取模板中的变量名列表
        input_variables = self._extract_variables(template)
        # 将变量名列表分配给实例属性
        self.input_variables = input_variables

    # 类方法：从模板字符串生成 PromptTemplate 实例
    @classmethod
    def from_template(cls, template: str):
        # 返回用 template 实例化的 PromptTemplate 对象
        return cls(template=template)

    # 格式化填充模板中的变量
    def format(self, **kwargs):
        # 计算模板中缺失但未传入的变量名集合
        missing_vars = set(self.input_variables) - set(kwargs.keys())
        # 如果存在缺失变量则抛出异常，提示哪些变量缺失
        if missing_vars:
            raise ValueError(f"缺少必需的变量: {missing_vars}")
        # 使用传入参数填充模板并返回格式化后的字符串
        return self.template.format(**kwargs)

    # 内部方法：从模板字符串中提取变量名
    def _extract_variables(self, template: str):
        # 定义正则表达式，匹配花括号中的变量名（冒号前的部分）
        pattern = r'\{([^}:]+)(?::[^}]+)?\}'
        # 查找所有符合 pattern 的变量名，返回匹配结果列表
        matches = re.findall(pattern, template)
        # 利用 dict 去重并保持顺序，最后转为列表返回
        return list(dict.fromkeys(matches))

5.3. pattern #

5.3.1.整体结构 #

r'\{([^}:]+)(?::[^}]+)?\}' 匹配花括号 {} 中的内容，支持 {key} 或 {key:value} 两种格式。

5.3.2.组成部分 #

5.3.2.1. \{ 和 \} #

• \{：匹配左花括号 {
• \}：匹配右花括号 }
• 因为 { 和 } 在正则中有特殊含义，所以需要转义

5.3.2.2. ([^}:]+) - 核心捕获组 #

• ()：捕获组，会提取这部分内容
• [^}:]+：匹配 一个或多个 不是 } 也不是 : 的字符
• 作用：匹配冒号前的 key 部分

5.3.2.3. (?::[^}]+)? - 可选部分 #

• (?:...)：非捕获组，只匹配不提取
• :：匹配冒号
• [^}]+：匹配冒号后的 value（不能包含 }）
• ?：表示整个非捕获组是可选的（0次或1次）

5.3.3.匹配示例 #

1. {name}          → 匹配 "name"
   {([^}:]+)}      ← key="name"，没有冒号部分

2. {age:25}        → 匹配 "age:25"
   {([^}:]+):[^}]+} ← key="age"，冒号后的"25"匹配但不捕获

3. {city:Beijing}  → 匹配 "city:Beijing"
                     key="city"，value="Beijing"（不捕获）

5.3.4.关键特性 #

1. 必须包含花括号
2. key不能包含 : 或 }
3. 冒号和value是可选的
4. 只捕获key部分

5.4. 去除列表重复元素 #

list(dict.fromkeys(matches)) 是 Python 中一种 去除列表重复元素 的简洁方法，同时保持原始顺序。

5.4.1 fromkeys #

5.4.1.1. fromkeys #

• 从列表 matches 创建一个字典
• 字典的键来自列表元素，所有键的值都是 None
• 重要特性：字典键是唯一的，重复的元素会被自动去重
• 保持顺序：Python 3.7+ 保证字典保持插入顺序

5.4.1.2. list #

• 将字典的键（已去重的元素）转换回列表

5.4.2 示例 #

matches = ['a', 'b', 'a', 'c', 'b', 'd', 'a']

# 逐步执行：
dict_result = dict.fromkeys(matches)
# dict_result = {'a': None, 'b': None, 'c': None, 'd': None}

final_list = list(dict_result)
# final_list = ['a', 'b', 'c', 'd']

5.4.3 注意事项 #

1. Python 版本要求：需要 Python 3.6+（官方保证 3.7+）
2. 可哈希元素：只能处理可哈希的元素类型
3. 保留第一个：重复元素保留第一个出现的位置

5.5. 类图 #

5.5.1 类图 #

5.5.2 类关系图 #

5.5.3 调用关系图 #

5.5.4 数据流转过程 #

1. 模板创建阶段

   • 使用 PromptTemplate.from_template("你好,我叫{name},你是谁?") 创建模板实例
   • 自动调用 _extract_variables() 方法，使用正则表达式 r'\{([^}:]+)(?::[^}]+)?\}' 提取变量名
   • 提取到变量列表：["name"]
   • 模板对象包含：template 属性和 input_variables 属性

2. 模板格式化阶段

   • 调用 prompt_template.format(name="张三") 方法
   • 检查变量完整性：验证 input_variables 中的所有变量是否都在 kwargs 中提供
   • 使用 Python 内置的 str.format() 方法填充变量
   • 模板字符串 "你好,我叫{name},你是谁?" → 格式化后 "你好,我叫张三,你是谁?"
   • 返回格式化后的字符串

3. 模型调用阶段

   • 将格式化后的字符串传给 ChatOpenAI.invoke()
   • ChatOpenAI 内部调用 _convert_input() 方法
   • 字符串输入被转换为 OpenAI API 格式：[{"role": "user", "content": "你好,我叫张三,你是谁?"}]

4. API 请求阶段

   • 构建 API 请求参数
   • 调用 OpenAI API 的 chat.completions.create() 方法
   • 传入模型名称和消息列表

5. 响应处理阶段

   • 从 API 响应中提取内容：response.choices[0].message.content
   • 创建 AIMessage 对象包装回复内容
   • 返回 AIMessage 对象给主程序

6. 结果输出阶段

   • 通过 result.content 访问 AI 回复的内容
   • 使用 print() 输出回复内容

关键特性说明

• PromptTemplate 的变量提取：使用正则表达式自动识别模板中的变量，支持 {variable} 和 {variable:format} 格式
• 变量完整性检查：format() 方法会验证所有必需的变量是否都已提供，缺失变量会抛出 ValueError 异常
• 字符串到消息的转换：ChatOpenAI 会自动将字符串输入转换为标准的消息格式，无需手动创建消息对象

6.ChatPromptTemplate #

ChatPromptTemplate 是 LangChain 风格的“多轮对话提示工程”工具类。它支持将系统消息、人类消息、AI消息等按顺序组合成复杂的对话提示，每个消息片段都可以包含模板变量，整体支持灵活拼装动态上下文。相比单轮的 PromptTemplate，它能表达更丰富的多角色对话、历史回合与情境设定，是复杂对话应用和推理链式开发的关键构件。

核心作用：

• 按照“消息”片段组织对话上下文，每个消息带有角色（system/human/ai）与内容模板。
• 支持变量自动填充，实现动态上下文的多轮提示（如把姓名、提问等变量嵌入多个回合）。
• 组织生成标准的消息对象列表(to_messages())，无需手动拼接角色和内容，接口强大且自动化。
• 与 ChatOpenAI.invoke() 等 LLM 调用方法无缝衔接，无需自行处理对话格式转换。

主要API用法：

• ChatPromptTemplate(messages)：通过消息模板列表直接构造。
• ChatPromptTemplate.from_messages([...])：通过具体消息对象或模板构造，适合动态分步拼装。
• invoke(变量字典)：传入变量，批量完成所有消息模板的填充，返回 ChatPromptValue 对象。
• to_string()：将多轮消息合成对话文本（通常给人类看，调试也方便）。
• to_messages()：生成 LLM API 需要的结构化消息对象列表，便于直接传给模型。

常见用法示例场景：

1. 情境设定 + 多轮对话
   比如先用 system 描述 AI 角色，再输入历史人类发问/AI回答，最后再加最新输入，实现连续对话和“记忆”上下文。
2. 变量上下文灵活插入
   如一次生产“你叫什么名字？”场景时，可以同时插入用户名、对话目标等多个变量，模板灵活简明。
3. 与模型调用贯通
   模板格式化后直接传给 llm.invoke(prompt_value)，无需开发者自己转换为 OpenAI 消息格式，极大简化消息流开发。

适用场景

• 构建复杂的多轮Prompt、角色扮演、对话历史上下文管理
• 搭建 RAG、工具调用、推理链、咨询问答等高阶LLM应用

注意事项

• 所有变量必须传入，缺失变量会报错，确保模板内容与参数输入一一对应
• 通常建议搭配消息对象（SystemMessage/HumanMessage/AIMessage）和 tuple/list 格式混合灵活定义模板

通过 ChatPromptTemplate，开发者可以轻松搭建出结构清晰、内容动态、多轮智能的 AI 对话系统，提升提示工程的可维护性和扩展性。

6.1. 6.ChatPromptTemplate.py #

6.ChatPromptTemplate.py

# 导入 ChatOpenAI 类用于调用大模型
#from langchain_openai import ChatOpenAI
# 导入对话模板及消息模板
#from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate, AIMessagePromptTemplate


from smartchain.chat_models import ChatOpenAI
from smartchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate, AIMessagePromptTemplate
# 创建一个 ChatOpenAI 实例，指定所使用的模型为 "gpt-4o"
llm = ChatOpenAI(model="gpt-4o")

# 定义一个 ChatPromptTemplate，包含系统消息、人类消息和 AI 消息，实现多轮对话模板
template = ChatPromptTemplate(
    [
        # 系统消息，设定 AI 助手身份和名称
        ("system", "你是一个乐于助人的 AI 机器人。你的名字叫{name}。"),
        # 用户消息，模拟人类主动问候
        ("human", "你好，你最近怎么样？"),
        # AI 消息，AI 针对问候做出回应
        ("ai", "我很好，谢谢你的关心！"),
        # 用户消息，带有变量输入，提升对话灵活性
        ("human", "{user_input}"),
    ]
)

# 填入变量，生成 prompt_value，从而得到带上下文变量的对话提示
prompt_value = template.invoke(
    {
        "name": "小助",
        "user_input": "你叫什么名字？",
    }
)

# 打印 prompt_value 转换为字符串的内容（合成对话文本）
print(prompt_value.to_string())

# 打印 prompt_value 转换为消息对象（消息列表 ，包含角色和内容）
print(prompt_value.to_messages())

# 调用 llm.invoke 以 prompt_value 为输入，获取 AI 回复结果
result = llm.invoke(prompt_value)

# 打印 AI 回复内容（只输出回复的字符串）
print(result.content)

# 通过 from_messages 方法，基于具体的消息模板对象构建 ChatPromptTemplate
template = ChatPromptTemplate.from_messages([
    # 构建系统消息模板
    SystemMessagePromptTemplate.from_template("你是一个乐于助人的 AI 机器人。你的名字叫{name}。"),
    # 构建用户消息模板
    HumanMessagePromptTemplate.from_template("你好，你最近怎么样？"),
    # 构建AI消息模板
    AIMessagePromptTemplate.from_template("我很好，谢谢你的关心！"),
    # 构建用户消息模板，内容带有变量
    HumanMessagePromptTemplate.from_template("{user_input}"),
])

# 调用 format_messages 方法，将变量填充进所有消息模板，得到最终的消息对象列表
prompt_messages = template.format_messages(name="小助", user_input="你叫什么名字？")

# 打印 format_messages 返回的消息对象列表（包含角色与消息内容）
print("format_messages 返回的消息列表:")
print(prompt_messages)

# 以格式化后的消息对象列表为输入，调用 llm.invoke 获取 AI 回复
result = llm.invoke(prompt_messages)

# 输出 AI 回复的内容
print(result.content)

6.2. prompts.py #

smartchain/prompts.py

# 导入正则表达式模块，用于变量提取
import re
+from .messages import SystemMessage, HumanMessage, AIMessage
# 定义提示词模板类
class PromptTemplate:
    # 类说明文档，描述用途
    """提示词模板类，用于格式化字符串模板"""

    # 构造方法，初始化模板实例
    def __init__(self, template: str):
        # 保存模板字符串到实例属性
        self.template = template
        # 调用内部方法提取模板中的变量名列表
        input_variables = self._extract_variables(template)
        # 将变量名列表分配给实例属性
        self.input_variables = input_variables

    # 类方法：从模板字符串生成 PromptTemplate 实例
    @classmethod
    def from_template(cls, template: str):
        # 返回用 template 实例化的 PromptTemplate 对象
        return cls(template=template)

    # 格式化填充模板中的变量
    def format(self, **kwargs):
        # 计算模板中缺失但未传入的变量名集合
        missing_vars = set(self.input_variables) - set(kwargs.keys())
        # 如果存在缺失变量则抛出异常，提示哪些变量缺失
        if missing_vars:
            raise ValueError(f"缺少必需的变量: {missing_vars}")
        # 使用传入参数填充模板并返回格式化后的字符串
        return self.template.format(**kwargs)

    # 内部方法：从模板字符串中提取变量名
    def _extract_variables(self, template: str):
        # 定义正则表达式，匹配花括号中的变量名（冒号前的部分）
        pattern = r'\{([^}:]+)(?::[^}]+)?\}'
        # 查找所有符合 pattern 的变量名，返回匹配结果列表
        matches = re.findall(pattern, template)
        # 利用 dict 去重并保持顺序，最后转为列表返回
        return list(dict.fromkeys(matches))
+
+# 定义一个用于存放格式化后的消息的类
+class ChatPromptValue:
+   # 聊天提示词值类，包含格式化后的消息列表
+   """聊天提示词值类，包含格式化后的消息列表"""
+
+   # 构造函数，接收一个消息对象列表
+   def __init__(self, messages):
+       # 保存消息列表到实例变量
+       self.messages = messages
+
+   # 将消息对象列表转为字符串，方便展示
+   def to_string(self):
+       # 新建一个用于存放字符串的列表
+       parts = []
+       # 遍历每个消息对象
+       for msg in self.messages:
+           # 如果消息对象有type和content属性
+           if hasattr(msg, 'type') and hasattr(msg, 'content'):
+               # 定义消息角色的映射关系
+               role_map = {
+                   "system": "System",
+                   "human": "Human",
+                   "ai": "AI"
+               }
+               # 获取对应的角色字符串，没有则首字母大写
+               role = role_map.get(msg.type, msg.type.capitalize())
+               # 拼接角色和消息内容
+               parts.append(f"{role}: {msg.content}")
+           else:
+               # 如果不是标准消息对象，则直接转为字符串
+               parts.append(str(msg))
+       # 将所有消息用换行符拼接起来组成一个字符串
+       return "\n".join(parts)
+
+   # 返回消息对象列表本身
+   def to_messages(self):
+       # 直接返回消息列表
+       return self.messages
+
+# 定义用于处理多轮对话消息模板的类
+class ChatPromptTemplate:
+   # 聊天提示词模板类，用于创建多轮对话的提示词
+   """聊天提示词模板类，用于创建多轮对话的提示词"""
+
+   # 构造方法，接收一个消息模板/对象的列表
+   def __init__(self, messages):
+       # 保存消息模板/对象列表
+       self.messages = messages
+       # 提取所有输入变量并存入实例变量
+       self.input_variables = self._extract_input_variables()
+   # 定义一个类方法，用于通过消息对象列表创建 ChatPromptTemplate 实例
+   @classmethod
+   def from_messages(cls, messages):
+       # 使用传入的 messages 参数创建并返回 ChatPromptTemplate 实例
+       return cls(messages=messages)
+   # 使用提供的变量格式化模板，返回消息列表
+   def format_messages(self, **kwargs):
+       # 格式化所有消息并返回
+       return self._format_all_messages(kwargs)    
+   # 私有方法，提取所有模板中用到的输入变量名
+   def _extract_input_variables(self):
+       # 用集合保存变量名，防止重复
+       variables = set()
+       # 遍历所有消息模板/对象
+       for msg in self.messages:
+           # 如果元素是(role, template_str)元组
+           if isinstance(msg, tuple) and len(msg) == 2:
+               _, template_str = msg
+               # 用PromptTemplate对象提取变量
+               prompt = PromptTemplate.from_template(template_str)
+               # 合并到集合中
+               variables.update(prompt.input_variables)
+       # 返回所有变量名组成的列表
+       return list(variables)
+
+   # 根据输入变量格式化所有消息模板，返回ChatPromptValue对象
+   def invoke(self, input_variables):
+       # 对消息模板进行实际变量填充
+       formatted_messages = self._format_all_messages(input_variables)
+       # 封装成ChatPromptValue对象返回
+       return ChatPromptValue(messages=formatted_messages)
+
+   # 将所有消息模板格式化并转换为消息对象列表
+   def _format_all_messages(self, variables):
+       # 新建列表保存格式化好的消息
+       formatted_messages = []
+       # 遍历每一个消息模板/对象
+       for msg in self.messages:
+           # 若是(role, template_str)元组
+           if isinstance(msg, tuple) and len(msg) == 2:
+               role, template_str = msg
+               # 创建PromptTemplate模板并填充变量
+               prompt = PromptTemplate.from_template(template_str)
+               content = prompt.format(**variables)
+               # 根据角色字符串生成对应的消息对象
+               formatted_messages.append(self._create_message_from_role(role, content))
+           # 如果是BaseMessagePromptTemplate的实例
+           elif isinstance(msg, BaseMessagePromptTemplate):
+               # 调用BaseMessagePromptTemplate的format方法，返回消息对象
+               formatted_messages.append(msg.format(**variables))    
+           else:
+               # 如不是模板，直接加入
+               formatted_messages.append(msg)
+       # 返回所有格式化的消息对象列表
+       return formatted_messages
+
+   # 辅助方法：根据角色和内容生成对应的标准消息对象
+   def _create_message_from_role(self, role, content):
+       # 角色字符串转小写做归一化
+       normalized_role = role.lower()
+       # 如果是system角色，返回SystemMessage对象
+       if normalized_role == "system":
+           return SystemMessage(content=content)
+       # 如果是human或user角色，返回HumanMessage对象
+       if normalized_role in ("human", "user"):
+           return HumanMessage(content=content)
+       # 如果是ai或assistant角色，返回AIMessage对象
+       if normalized_role in ("ai", "assistant"):
+           return AIMessage(content=content)
+       # 如果角色未知，则抛出异常
+       raise ValueError(f"未知的消息角色: {role}")    
+
+# 定义基础消息提示词模板类
+class BaseMessagePromptTemplate:
+   # 基础消息提示词模板类声明
+   """基础消息提示词模板类"""
+   
+   # 构造函数，必须传入PromptTemplate实例
+   def __init__(self, prompt: PromptTemplate):
+       # 将PromptTemplate实例保存在self.prompt属性中
+       self.prompt = prompt
+
+   # 工厂方法，利用模板字符串创建类实例
+   @classmethod
+   def from_template(cls, template: str):
+       # 通过模板字符串创建PromptTemplate对象
+       prompt = PromptTemplate.from_template(template)
+       # 用生成的PromptTemplate创建本类实例并返回
+       return cls(prompt=prompt)
+
+   # 格式化当前模板，返回消息对象
+   def format(self, **kwargs):
+       # 使用PromptTemplate格式化内容，得到最终文本
+       content = self.prompt.format(**kwargs)
+       # 调用子类实现的方法将文本转换为对应类型消息对象
+       return self._create_message(content)
+
+   # 抽象方法，子类必须实现，用于生成特定类型的消息对象
+   def _create_message(self, content):
+       raise NotImplementedError
+
+# 系统消息提示词模板类，继承自BaseMessagePromptTemplate
+class SystemMessagePromptTemplate(BaseMessagePromptTemplate):
+   # 系统消息提示词模板说明
+   """系统消息提示词模板"""
+
+   # 实现父类的_create_message方法，返回系统消息对象
+   def _create_message(self, content):
+       # 创建并返回SystemMessage对象，内容为content
+       return SystemMessage(content=content)
+
+# 人类消息提示词模板类，继承自BaseMessagePromptTemplate
+class HumanMessagePromptTemplate(BaseMessagePromptTemplate):
+   # 人类消息提示词模板说明
+   """人类消息提示词模板"""
+
+   # 实现父类的_create_message方法，返回人类消息对象
+   def _create_message(self, content):
+       # 创建并返回HumanMessage对象，内容为content
+       return HumanMessage(content=content)
+
+# AI消息提示词模板类，继承自BaseMessagePromptTemplate
+class AIMessagePromptTemplate(BaseMessagePromptTemplate):
+   # AI消息提示词模板说明
+   """AI消息提示词模板"""
+
+   # 实现父类的_create_message方法，返回AI消息对象
+   def _create_message(self, content):
+       # 创建并返回AIMessage对象，内容为content
+       return AIMessage(content=content)

6.3. chat_models.py #

smartchain/chat_models.py

# 导入操作系统相关模块
import os

# 导入 openai 模块
import openai

# 从 .messages 模块导入 AIMessage、HumanMessage 和 SystemMessage 类
from .messages import AIMessage, HumanMessage, SystemMessage
+from .prompts import ChatPromptValue
# 定义与 OpenAI 聊天模型交互的类
class ChatOpenAI:

    # 初始化方法
    def __init__(self, model: str = "gpt-4o", **kwargs):
        # 初始化 ChatOpenAI 类
        """
        初始化 ChatOpenAI

        Args:
            model: 模型名称，如 "gpt-4o"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("OPENAI_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 OPENAI_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 创建 OpenAI 客户端实例
        self.client = openai.OpenAI(api_key=self.api_key)

    # 调用模型生成回复的方法
    def invoke(self, input, **kwargs):
        # 调用模型生成回复
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 流式调用模型生成回复的方法
    def stream(self, input, **kwargs):
        # 流式调用模型生成回复
        """
        流式调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Yields:
            AIMessage: AI 的回复消息块（每次产生部分内容）
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，启用流式输出
        params = {
            "model": self.model,
            "messages": messages,
            "stream": True,  # 启用流式输出
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 客户端发起流式调用
        stream = self.client.chat.completions.create(**params)
        # 迭代流式响应
        for chunk in stream:
            # 检查是否有内容增量
            if chunk.choices and len(chunk.choices) > 0:
                delta = chunk.choices[0].delta
                # 检查 delta 中是否有 content，如果有则发送
                if hasattr(delta, 'content') and delta.content:
                    # 产生包含部分内容的 AIMessage
                    yield AIMessage(content=delta.content)

    # 内部方法，将输入转换为 OpenAI API 需要的消息格式
    def _convert_input(self, input):
        # 将输入转换为 OpenAI API 需要的消息格式
        """
        将输入转换为 OpenAI API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: OpenAI API 格式的消息列表
        """
+       if isinstance(input, ChatPromptValue):
+           input = input.to_messages()
        # 输入为字符串时，直接封装为用户角色消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
        # 如果输入是列表类型
        elif isinstance(input, list):
            # 新建一个空的消息列表
            messages = []
            # 遍历输入列表中的每一个元素
            for msg in input:
                # 判断是否为字符串，是则作为用户消息加入
                if isinstance(msg, str):
                    messages.append({"role": "user", "content": msg})
                # 判断是否为 HumanMessage、AIMessage 或 SystemMessage 实例
                elif isinstance(msg, (HumanMessage, AIMessage, SystemMessage)):
                    # 如果是 HumanMessage，将角色设为 user
                    if isinstance(msg, HumanMessage):
                        role = "user"
                    # 如果是 AIMessage，将角色设为 assistant
                    elif isinstance(msg, AIMessage):
                        role = "assistant"
                    # 如果是 SystemMessage，将角色设为 system
                    elif isinstance(msg, SystemMessage):
                        role = "system"
                    # 获取消息内容（有 content 属性则取 content，否则转为字符串）
                    content = msg.content if hasattr(msg, "content") else str(msg)
                    # 将角色和内容添加到消息列表
                    messages.append({"role": role, "content": content})
                # 如果元素本身为字典，直接添加进消息列表
                elif isinstance(msg, dict):
                    # 直接添加字典类型的消息
                    messages.append(msg)
                # 如果元素为长度为 2 的元组，将其解包为 role 和 content
                elif isinstance(msg, tuple) and len(msg) == 2:
                    # 将元组解包为 role 和 content
                    role, content = msg
                    # 将角色和内容添加到消息列表
                    messages.append({"role": role, "content": content})
            # 返回构建好的消息列表
            return messages    
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]

# 定义与 DeepSeek 聊天模型交互的类
class ChatDeepSeek:
    # 初始化方法
    # model: 模型名称，默认为 "deepseek-chat"
    # **kwargs: 其他可选参数（如 temperature, max_tokens 等）
    def __init__(self, model: str = "deepseek-chat", **kwargs):
        """
        初始化 ChatDeepSeek

        Args:
            model: 模型名称，如 "deepseek-chat"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("DEEPSEEK_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 DEEPSEEK_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 获取 DeepSeek 的 base_url，默认为官方地址
        base_url = kwargs.get("base_url", "https://api.deepseek.com/v1")
        # 创建 OpenAI 兼容的客户端实例（DeepSeek 使用 OpenAI 兼容的 API）
        self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)

    # 调用模型生成回复的方法
    # input: 输入内容，可以是字符串或消息列表
    # **kwargs: 额外的 API 参数
    def invoke(self, input, **kwargs):
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个选项
        choice = response.choices[0]
        # 获取消息内容
        content = choice.message.content or ""
        # 返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 内部方法，将输入转换为 API 需要的消息格式
    # input: 字符串、消息列表或 ChatPromptValue
    def _convert_input(self, input):
        """
        将输入转换为 API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: API 格式的消息列表
        """
        # 如果输入是字符串，直接作为用户消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]    
        else:
            # 其他输入类型，转为字符串作为 user 消息
            return [{"role": "user", "content": str(input)}]

# 定义与通义千问（Tongyi）聊天模型交互的类
class ChatTongyi:

    # 初始化方法
    # 初始化方法，设置模型名称和 API 相关参数
    def __init__(self, model: str = "qwen-max", **kwargs):
        """
        初始化 ChatTongyi

        Args:
            model: 模型名称，如 "qwen-max"
            **kwargs: 其他参数（如 temperature, max_tokens 等）
        """
        # 设置模型名称
        self.model = model
        # 获取 api_key，优先从参数获取，否则从环境变量获取
        self.api_key = kwargs.get("api_key") or os.getenv("DASHSCOPE_API_KEY")
        # 如果没有提供 api_key，则抛出异常
        if not self.api_key:
            raise ValueError("需要提供 api_key 或设置 DASHSCOPE_API_KEY 环境变量")
        # 保存除 api_key 之外的其他参数，用于 API 调用
        self.model_kwargs = {k: v for k, v in kwargs.items() if k != "api_key"}
        # 获取通义千问的 API base URL（使用 OpenAI 兼容模式），如果未指定则使用默认值
        base_url = kwargs.get("base_url", "https://dashscope.aliyuncs.com/compatible-mode/v1")
        # 创建 OpenAI 兼容的客户端实例（通义千问使用 OpenAI 兼容的 API）
        self.client = openai.OpenAI(api_key=self.api_key, base_url=base_url)

    # 调用模型生成回复的方法
    # 调用模型生成回复，返回 AIMessage 对象
    def invoke(self, input, **kwargs):
        """
        调用模型生成回复

        Args:
            input: 输入内容，可以是字符串或消息列表
            **kwargs: 额外的 API 参数

        Returns:
            AIMessage: AI 的回复消息
        """
        # 将输入数据转换为消息格式
        messages = self._convert_input(input)
        # 构建 API 请求参数字典，包含模型名、消息内容和其他参数
        params = {
            "model": self.model,
            "messages": messages,
            **self.model_kwargs,
            **kwargs
        }
        # 使用 OpenAI 兼容的客户端发起 chat.completions.create 调用以获取回复
        response = self.client.chat.completions.create(**params)
        # 取出返回结果中的第一个回复选项
        choice = response.choices[0]
        # 获取回复的消息内容，如果内容不存在则返回空字符串
        content = choice.message.content or ""
        # 构建并返回一个 AIMessage 对象
        return AIMessage(content=content)

    # 内部方法，将输入转换为 API 需要的消息格式
    # 支持字符串、消息列表等输入，统一包装为 OpenAI API 格式
    def _convert_input(self, input):
        """
        将输入转换为 API 需要的消息格式

        Args:
            input: 字符串、消息列表或 ChatPromptValue

        Returns:
            list[dict]: API 格式的消息列表
        """
        # 如果输入是字符串，直接包装为“用户”角色的消息
        if isinstance(input, str):
            return [{"role": "user", "content": input}]
        else:
            # 其他输入类型，转换为字符串作为“用户”消息内容
            return [{"role": "user", "content": str(input)}]            

6.4. 类图 #

6.4.1 类说明 #

6.4.2 类关系图 #

6.4.3 调用关系图 #

6.4.3.1 方式一：使用元组 #

定义模板     → 提取变量     → 传入值   → 格式化     → 封装        → 调用模型 → 返回结果
    ↓            ↓            ↓         ↓            ↓             ↓         ↓
元组列表   input_variables  变量字典   消息对象  ChatPromptValue  API调用   AIMessage

6.4.3.2 方式二：使用消息模板对象 #

6.4.4 数据流转过程 #

1. 模板创建：使用元组或消息模板对象创建 ChatPromptTemplate
2. 变量提取：自动提取所有模板中的变量名
3. 格式化：使用 invoke() 或 format_messages() 填充变量
4. 消息生成：根据角色创建对应的消息对象（SystemMessage、HumanMessage、AIMessage）
5. 模型调用：将格式化后的消息传给 ChatOpenAI.invoke()
6. API 请求：转换为 API 格式并调用 OpenAI API
7. 结果返回：返回 AIMessage 对象，包含 AI 回复内容

7.MessagesPlaceholder #

在实际的多轮对话场景中，除了单轮问答外，模型往往需要结合用户历史的聊天上下文一起进行推理，这时我们就需要在提示词模板里能灵活地插入和动态填充一段“消息历史”。为此，MessagesPlaceholder 类应运而生：

什么是 MessagesPlaceholder？

MessagesPlaceholder 是聊天模板中的一种“占位符”类型，用于表示一段由多条消息（如人类和 AI 轮流发言）组成的历史消息列表。它和普通的文本变量不同，普通变量通常用于填充一句话，而消息占位符可以一次性插入多条结构化消息对象，并保持消息角色和内容的结构化属性。

典型用法举例

假设已有历史消息列表 history（其中每条为 HumanMessage 或 AIMessage），你可以这样构建模板：

from smartchain.prompts import ChatPromptTemplate, MessagesPlaceholder

template = ChatPromptTemplate([
    ("system", "你是一个乐于助人的 AI 助手。"),
    MessagesPlaceholder("history"),  # 此处为动态插入历史消息
    ("human", "{question}"),
])

当调用 template.format_messages(history=history, question="你的姓名是什么？") 时，占位符 "history" 会自动被 history 列表中的所有消息对象（依次填充，原始的消息类型和顺序均会保留）替换。模板处理后拼接出来的消息列表即为模型实际输入。

为什么要用消息占位符？

• 保持对话上下文
  许多大模型需要完整的对话历史才能按照多轮语境回答问题，通过 MessagesPlaceholder 可以完整插入并编排所有历史消息而无需手动拼接字符串。
• 结构化链路
  占位符支持直接传递消息对象而不是生拼字符串，保证系统、用户、AI 各角色消息的有序性和类型安全。
• 简单易用
  结合变量名调用，与传递历史变量类似，无需额外处理即可实现多轮上下文插入。

工作原理

• 在 ChatPromptTemplate 内，遇到 MessagesPlaceholder("变量名") 时，不会格式化为一条具体字符串，而是把变量名对应的消息对象列表批量插入到最终消息流中。
• 你可以在模板中插入任意数量或位置的 MessagesPlaceholder，甚至用于多个不同阶段的上下文。

注意事项

• 传入的“历史变量”一定要是消息对象列表（如 [HumanMessage(...), AIMessage(...), ...]），不能是字符串列表。
• 若模板未被正确填充历史信息，模型就无法获得之前的上下文。

这种机制让多轮对话的提示词模板高度结构化且可复用，是现代聊天机器人实现上下文记忆的标准利器。

7.1. 7.MessagesPlaceholder.py #

7.MessagesPlaceholder.py

# 从 langchain_core.prompts 导入对话模板类和消息占位符
#from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 从 langchain_openai 导入 ChatOpenAI，大语言模型接口
#from langchain_openai import ChatOpenAI
# 导入人类消息和 AI 消息类
#from langchain_core.messages import HumanMessage, AIMessage

from smartchain.chat_models import ChatOpenAI
from smartchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from smartchain.messages import HumanMessage, AIMessage

# 构建历史消息列表，模拟对话历史（人类与 AI 的多轮消息）
history = [
    HumanMessage(content="你好"),
    AIMessage(content="你好，很高兴见到你"),
]

# 创建 ChatPromptTemplate 模板，包含系统消息、历史消息占位符和用户新提问
template = ChatPromptTemplate(
    [
        # 系统消息，设定 AI 身份
        ("system", "你是一个乐于助人的 AI 助手。"),
        # 消息占位符，将填充历史对话
        MessagesPlaceholder("history"),
        # 用户输入，实际提问
        ("human", "{question}"),
    ]
)

# 实例化 OpenAI 聊天模型，选用 "gpt-4o" 模型
llm = ChatOpenAI(model="gpt-4o")

# 调用模板的 format_messages 方法，注入历史对话和本轮问题
prompt_messages = template.format_messages(
    history=history,
    question="请介绍一下你自己？",
)

# 打印 format_messages 返回的消息，显示内部分角色的消息对象
print("format_messages 输出的消息：")
for msg in prompt_messages:
    print(msg)

# 直接将构建好的消息列表传入模型进行推理，获得 AI 回复
response = llm.invoke(prompt_messages)

# 打印模型返回的内容（AI 对最新问题的回答）
print("\n模型回复：")
print(response.content)

7.2. prompts.py #

smartchain/prompts.py

# 导入正则表达式模块，用于变量提取
import re
from .messages import SystemMessage, HumanMessage, AIMessage
# 定义提示词模板类
class PromptTemplate:
    # 类说明文档，描述用途
    """提示词模板类，用于格式化字符串模板"""

    # 构造方法，初始化模板实例
    def __init__(self, template: str):
        # 保存模板字符串到实例属性
        self.template = template
        # 调用内部方法提取模板中的变量名列表
        input_variables = self._extract_variables(template)
        # 将变量名列表分配给实例属性
        self.input_variables = input_variables

    # 类方法：从模板字符串生成 PromptTemplate 实例
    @classmethod
    def from_template(cls, template: str):
        # 返回用 template 实例化的 PromptTemplate 对象
        return cls(template=template)

    # 格式化填充模板中的变量
    def format(self, **kwargs):
        # 计算模板中缺失但未传入的变量名集合
        missing_vars = set(self.input_variables) - set(kwargs.keys())
        # 如果存在缺失变量则抛出异常，提示哪些变量缺失
        if missing_vars:
            raise ValueError(f"缺少必需的变量: {missing_vars}")
        # 使用传入参数填充模板并返回格式化后的字符串
        return self.template.format(**kwargs)

    # 内部方法：从模板字符串中提取变量名
    def _extract_variables(self, template: str):
        # 定义正则表达式，匹配花括号中的变量名（冒号前的部分）
        pattern = r'\{([^}:]+)(?::[^}]+)?\}'
        # 查找所有符合 pattern 的变量名，返回匹配结果列表
        matches = re.findall(pattern, template)
        # 利用 dict 去重并保持顺序，最后转为列表返回
        return list(dict.fromkeys(matches))

# 定义一个用于存放格式化后的消息的类
class ChatPromptValue:
    # 聊天提示词值类，包含格式化后的消息列表
    """聊天提示词值类，包含格式化后的消息列表"""

    # 构造函数，接收一个消息对象列表
    def __init__(self, messages):
        # 保存消息列表到实例变量
        self.messages = messages

    # 将消息对象列表转为字符串，方便展示
    def to_string(self):
        # 新建一个用于存放字符串的列表
        parts = []
        # 遍历每个消息对象
        for msg in self.messages:
            # 如果消息对象有type和content属性
            if hasattr(msg, 'type') and hasattr(msg, 'content'):
                # 定义消息角色的映射关系
                role_map = {
                    "system": "System",
                    "human": "Human",
                    "ai": "AI"
                }
                # 获取对应的角色字符串，没有则首字母大写
                role = role_map.get(msg.type, msg.type.capitalize())
                # 拼接角色和消息内容
                parts.append(f"{role}: {msg.content}")
            else:
                # 如果不是标准消息对象，则直接转为字符串
                parts.append(str(msg))
        # 将所有消息用换行符拼接起来组成一个字符串
        return "\n".join(parts)

    # 返回消息对象列表本身
    def to_messages(self):
        # 直接返回消息列表
        return self.messages

# 定义用于处理多轮对话消息模板的类
class ChatPromptTemplate:
    # 聊天提示词模板类，用于创建多轮对话的提示词
    """聊天提示词模板类，用于创建多轮对话的提示词"""

    # 构造方法，接收一个消息模板/对象的列表
    def __init__(self, messages):
        # 保存消息模板/对象列表
        self.messages = messages
        # 提取所有输入变量并存入实例变量
        self.input_variables = self._extract_input_variables()
    # 定义一个类方法，用于通过消息对象列表创建 ChatPromptTemplate 实例
    @classmethod
    def from_messages(cls, messages):
        # 使用传入的 messages 参数创建并返回 ChatPromptTemplate 实例
        return cls(messages=messages)
    # 使用提供的变量格式化模板，返回消息列表
    def format_messages(self, **kwargs):
        # 格式化所有消息并返回
        return self._format_all_messages(kwargs)    
    # 私有方法，提取所有模板中用到的输入变量名
    def _extract_input_variables(self):
        # 用集合保存变量名，防止重复
        variables = set()
        # 遍历所有消息模板/对象
        for msg in self.messages:
            # 如果元素是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                _, template_str = msg
                # 用PromptTemplate对象提取变量
                prompt = PromptTemplate.from_template(template_str)
                # 合并到集合中
                variables.update(prompt.input_variables)
+           # 如果是BaseMessagePromptTemplate子类实例
+           elif isinstance(msg, BaseMessagePromptTemplate):
+               variables.update(msg.prompt.input_variables)
+           # 如果是占位符对象
+           elif isinstance(msg, MessagesPlaceholder):
+               variables.add(msg.variable_name)    
        # 返回所有变量名组成的列表
        return list(variables)

    # 根据输入变量格式化所有消息模板，返回ChatPromptValue对象
    def invoke(self, input_variables):
        # 对消息模板进行实际变量填充
        formatted_messages = self._format_all_messages(input_variables)
        # 封装成ChatPromptValue对象返回
        return ChatPromptValue(messages=formatted_messages)

    # 将所有消息模板格式化并转换为消息对象列表
    def _format_all_messages(self, variables):
        # 新建列表保存格式化好的消息
        formatted_messages = []
        # 遍历每一个消息模板/对象
        for msg in self.messages:
            # 若是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                role, template_str = msg
                # 创建PromptTemplate模板并填充变量
                prompt = PromptTemplate.from_template(template_str)
                content = prompt.format(**variables)
                # 根据角色字符串生成对应的消息对象
                formatted_messages.append(self._create_message_from_role(role, content))
            # 如果是BaseMessagePromptTemplate的实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                # 调用BaseMessagePromptTemplate的format方法，返回消息对象
                formatted_messages.append(msg.format(**variables))    
+           # 如果是占位符对象
+           elif isinstance(msg, MessagesPlaceholder):
+               placeholder_messages = self._coerce_placeholder_value(
+                   msg.variable_name, variables.get(msg.variable_name)
+               )
+               formatted_messages.extend(placeholder_messages)    
            else:
                # 如不是模板，直接加入
                formatted_messages.append(msg)
        # 返回所有格式化的消息对象列表
        return formatted_messages
+    # 处理占位符对象的值，返回消息对象列表
+   def _coerce_placeholder_value(self, variable_name, value):
+       # 如果未传入变量，抛出异常
+       if value is None:
+           raise ValueError(f"MessagesPlaceholder '{variable_name}' 对应变量缺失")
+       # 如果是ChatPromptValue实例，转换为消息列表
+       if isinstance(value, ChatPromptValue):
+           return value.to_messages()
+       # 如果已经是消息对象/结构列表，则依次转换
+       if isinstance(value, list):
+           return [self._coerce_single_message(item) for item in value]
+       # 其他情况尝试单个转换
+       return [self._coerce_single_message(value)]
+       # 单个原始值转换为消息对象
+   def _coerce_single_message(self, value):
+       # 已是有效消息类型，直接返回
+       if isinstance(value, (SystemMessage, HumanMessage, AIMessage)):
+           return value
+       # 有type和content属性，也当消息对象直接返回
+       if hasattr(value, "type") and hasattr(value, "content"):
+           return value
+       # 字符串变为人类消息
+       if isinstance(value, str):
+           return HumanMessage(content=value)
+       # (role, content)元组转为指定角色的消息
+       if isinstance(value, tuple) and len(value) == 2:
+           # 解包元组
+           role, content = value
+           # 根据role和content生成对应的消息对象
+           return self._create_message_from_role(role, content)
+       # 字典，默认user角色
+       if isinstance(value, dict):
+           # 获取role和content
+           role = value.get("role", "user")
+           # 获取content
+           content = value.get("content", "")
+           # 根据role和content生成对应的消息对象
+           return self._create_message_from_role(role, content)
+       # 其他无法识别类型，抛出异常
+       raise TypeError("无法将占位符内容转换为消息")    
    # 辅助方法：根据角色和内容生成对应的标准消息对象
    def _create_message_from_role(self, role, content):
        # 角色字符串转小写做归一化
        normalized_role = role.lower()
        # 如果是system角色，返回SystemMessage对象
        if normalized_role == "system":
            return SystemMessage(content=content)
        # 如果是human或user角色，返回HumanMessage对象
        if normalized_role in ("human", "user"):
            return HumanMessage(content=content)
        # 如果是ai或assistant角色，返回AIMessage对象
        if normalized_role in ("ai", "assistant"):
            return AIMessage(content=content)
        # 如果角色未知，则抛出异常
        raise ValueError(f"未知的消息角色: {role}")    

# 定义基础消息提示词模板类
class BaseMessagePromptTemplate:
    # 基础消息提示词模板类声明
    """基础消息提示词模板类"""

    # 构造函数，必须传入PromptTemplate实例
    def __init__(self, prompt: PromptTemplate):
        # 将PromptTemplate实例保存在self.prompt属性中
        self.prompt = prompt

    # 工厂方法，利用模板字符串创建类实例
    @classmethod
    def from_template(cls, template: str):
        # 通过模板字符串创建PromptTemplate对象
        prompt = PromptTemplate.from_template(template)
        # 用生成的PromptTemplate创建本类实例并返回
        return cls(prompt=prompt)

    # 格式化当前模板，返回消息对象
    def format(self, **kwargs):
        # 使用PromptTemplate格式化内容，得到最终文本
        content = self.prompt.format(**kwargs)
        # 调用子类实现的方法将文本转换为对应类型消息对象
        return self._create_message(content)

    # 抽象方法，子类必须实现，用于生成特定类型的消息对象
    def _create_message(self, content):
        raise NotImplementedError

# 系统消息提示词模板类，继承自BaseMessagePromptTemplate
class SystemMessagePromptTemplate(BaseMessagePromptTemplate):
    # 系统消息提示词模板说明
    """系统消息提示词模板"""

    # 实现父类的_create_message方法，返回系统消息对象
    def _create_message(self, content):
        # 创建并返回SystemMessage对象，内容为content
        return SystemMessage(content=content)

# 人类消息提示词模板类，继承自BaseMessagePromptTemplate
class HumanMessagePromptTemplate(BaseMessagePromptTemplate):
    # 人类消息提示词模板说明
    """人类消息提示词模板"""

    # 实现父类的_create_message方法，返回人类消息对象
    def _create_message(self, content):
        # 创建并返回HumanMessage对象，内容为content
        return HumanMessage(content=content)

# AI消息提示词模板类，继承自BaseMessagePromptTemplate
class AIMessagePromptTemplate(BaseMessagePromptTemplate):
    # AI消息提示词模板说明
    """AI消息提示词模板"""

    # 实现父类的_create_message方法，返回AI消息对象
    def _create_message(self, content):
        # 创建并返回AIMessage对象，内容为content
        return AIMessage(content=content)
+
+# 定义动态消息列表占位符类
+class MessagesPlaceholder:
+   # 在聊天模板中插入动态消息列表的占位符
+   """在聊天模板中插入动态消息列表的占位符"""
+
+   # 构造方法，存储变量名
+   def __init__(self, variable_name: str):
+       self.variable_name = variable_name

7.3 类图 #

7.3.1 类图 #

7.3.2 类关系图 #

7.3.3 调用关系图 #

7.3.4 数据流转过程 #

1. 历史消息构建

   • 创建 HumanMessage 和 AIMessage 对象
   • 组成历史消息列表：[HumanMessage("你好"), AIMessage("你好，很高兴见到你")]

2. 占位符创建

   • 创建 MessagesPlaceholder("history") 对象
   • 指定占位符变量名为 "history"

3. 模板创建

   • 使用 ChatPromptTemplate 创建模板，包含：
     • 系统消息元组：("system", "...")
     • 消息占位符：MessagesPlaceholder("history")
     • 用户消息元组：("human", "{question}")
   • 自动提取变量：["history", "question"]

4. 模板格式化

   • 调用 format_messages(history=history, question="...")
   • 处理系统消息：创建 SystemMessage 对象
   • 处理占位符：
     • 识别 MessagesPlaceholder("history")
     • 调用 _coerce_placeholder_value("history", history)
     • 遍历历史消息列表，每个消息通过 _coerce_single_message() 处理
     • 由于是 HumanMessage 和 AIMessage 对象，直接返回
     • 将消息列表扩展到格式化结果中
   • 处理用户消息：格式化 {question} 变量，创建 HumanMessage 对象

5. 最终消息列表

   • 结果：[SystemMessage, HumanMessage(历史), AIMessage(历史), HumanMessage(当前问题)]

6. 模型调用

   • 将消息列表传给 ChatOpenAI.invoke()
   • ChatOpenAI 内部转换消息格式并调用 API
   • 返回 AIMessage 对象

8.FewShotPromptTemplate #

• FewShotPromptTemplate 主要实现了通过指定若干个“示例问答”，配合用户输入和前后缀，自动构建符合 few-shot 学习范式的提示词字符串。
• 其流程可归纳为：

1. 定义示例集合
   • 用户提供若干格式一致的示例数据，每个为 dict，含输入和输出字段（如 question/answer）。
2. 定义单条示例格式模板
   • 使用 PromptTemplate 定义单个示例的展现格式，包含变量（如 {question}）。
3. 创建 FewShotPromptTemplate
   • 指定示例集合、单例模板、前缀、后缀、输入变量等，初始化模板。
4. 插入真实用户输入，调用 .format(kwargs)**
   • 对前缀和后缀自动变量替换，对每条示例应用模板格式化，最终按照分隔符拼接所有部分生成完整提示词。
5. 交互与推理
   • 将生成的完整 few-shot prompt 交给 ChatOpenAI.invoke，实现大模型自动问答。

流程图说明

数据结构与流转

• examples：list[dict]，如 [{"question": ..., "answer": ...}, ...]
• PromptTemplate: 给单条示例用，负责变量替换
• FewShotPromptTemplate: 组成完整提示词流程——前缀 + 多个示例格式化 + 后缀，均可插入变量
• 完整 prompt: 字符串，样例区块与实际问题合并
• ChatOpenAI: 接收字符串 prompt，转为聊天消息格式，以对话方式推理

类功能对比

典型应用场景

• 自动构建“少量示例”加“实际用户问题”的高质量提示语，以提升大模型回答的准确性
• 灵活引入不同风格的 few-shot 示例，适用于翻译、问答、代码推理等多类任务

这种模块化的 FewShotPromptTemplate 支持更复杂的示例组织与复用，是构建提示工程自动化系统的关键基石。

8.1. 8.FewShotPromptTemplate.py #

8.FewShotPromptTemplate.py

# 导入 PromptTemplate 和 FewShotPromptTemplate，用于构建 few-shot 提示词模板
#from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
# 导入 OpenAI 聊天模型接口
#from langchain_openai import ChatOpenAI

from smartchain.chat_models import ChatOpenAI
from smartchain.prompts import PromptTemplate, FewShotPromptTemplate

# 定义 few-shot 示例集合，每个示例包含一个问题和对应答案
examples = [
    {"question": "1 plus 1 等于多少？", "answer": "答案是 2。"},
    {"question": "2 plus 2 等于多少？", "answer": "答案是 4。"},
]

# 定义每个示例的格式模板
example_prompt = PromptTemplate.from_template(
    "示例问题：{question}\n示例回答：{answer}"
)

# 构建 few-shot 提示词模板，包含前缀、示例、后缀和输入变量
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,                              # 提供 few-shot 示例
    example_prompt=example_prompt,                  # 每个示例应用的格式
    prefix="你是一个擅长算术的 AI 助手。以下是一些示例：",     # 前缀说明
    suffix="请回答用户问题：{user_question}\nAI：",    # 后缀及用户真实问题
    input_variables=["user_question"],              # 所需输入变量名
)

# 注入用户问题，生成完整的 few-shot 提示词文本
formatted_prompt = few_shot_prompt.format(user_question="3 plus 5 等于多少？")
# 打印生成的 few-shot 提示词
print("few-shot 提示词：\n")
print(formatted_prompt)

# 实例化 ChatOpenAI，选择 "gpt-4o" 模型
llm = ChatOpenAI(model="gpt-4o")
# 将格式化后的提示词传给模型进行推理
response = llm.invoke(formatted_prompt)
# 打印模型返回的内容
print("\n模型回答：")
print(response.content)

8.2. prompts.py #

smartchain/prompts.py

# 导入正则表达式模块，用于变量提取
import re
from .messages import SystemMessage, HumanMessage, AIMessage
# 定义提示词模板类
class PromptTemplate:
    # 类说明文档，描述用途
    """提示词模板类，用于格式化字符串模板"""

    # 构造方法，初始化模板实例
    def __init__(self, template: str):
        # 保存模板字符串到实例属性
        self.template = template
        # 调用内部方法提取模板中的变量名列表
        input_variables = self._extract_variables(template)
        # 将变量名列表分配给实例属性
        self.input_variables = input_variables

    # 类方法：从模板字符串生成 PromptTemplate 实例
    @classmethod
    def from_template(cls, template: str):
        # 返回用 template 实例化的 PromptTemplate 对象
        return cls(template=template)

    # 格式化填充模板中的变量
    def format(self, **kwargs):
        # 计算模板中缺失但未传入的变量名集合
        missing_vars = set(self.input_variables) - set(kwargs.keys())
        # 如果存在缺失变量则抛出异常，提示哪些变量缺失
        if missing_vars:
            raise ValueError(f"缺少必需的变量: {missing_vars}")
        # 使用传入参数填充模板并返回格式化后的字符串
        return self.template.format(**kwargs)

    # 内部方法：从模板字符串中提取变量名
    def _extract_variables(self, template: str):
        # 定义正则表达式，匹配花括号中的变量名（冒号前的部分）
        pattern = r'\{([^}:]+)(?::[^}]+)?\}'
        # 查找所有符合 pattern 的变量名，返回匹配结果列表
        matches = re.findall(pattern, template)
        # 利用 dict 去重并保持顺序，最后转为列表返回
        return list(dict.fromkeys(matches))

# 定义一个用于存放格式化后的消息的类
class ChatPromptValue:
    # 聊天提示词值类，包含格式化后的消息列表
    """聊天提示词值类，包含格式化后的消息列表"""

    # 构造函数，接收一个消息对象列表
    def __init__(self, messages):
        # 保存消息列表到实例变量
        self.messages = messages

    # 将消息对象列表转为字符串，方便展示
    def to_string(self):
        # 新建一个用于存放字符串的列表
        parts = []
        # 遍历每个消息对象
        for msg in self.messages:
            # 如果消息对象有type和content属性
            if hasattr(msg, 'type') and hasattr(msg, 'content'):
                # 定义消息角色的映射关系
                role_map = {
                    "system": "System",
                    "human": "Human",
                    "ai": "AI"
                }
                # 获取对应的角色字符串，没有则首字母大写
                role = role_map.get(msg.type, msg.type.capitalize())
                # 拼接角色和消息内容
                parts.append(f"{role}: {msg.content}")
            else:
                # 如果不是标准消息对象，则直接转为字符串
                parts.append(str(msg))
        # 将所有消息用换行符拼接起来组成一个字符串
        return "\n".join(parts)

    # 返回消息对象列表本身
    def to_messages(self):
        # 直接返回消息列表
        return self.messages

# 定义用于处理多轮对话消息模板的类
class ChatPromptTemplate:
    # 聊天提示词模板类，用于创建多轮对话的提示词
    """聊天提示词模板类，用于创建多轮对话的提示词"""

    # 构造方法，接收一个消息模板/对象的列表
    def __init__(self, messages):
        # 保存消息模板/对象列表
        self.messages = messages
        # 提取所有输入变量并存入实例变量
        self.input_variables = self._extract_input_variables()
    # 定义一个类方法，用于通过消息对象列表创建 ChatPromptTemplate 实例
    @classmethod
    def from_messages(cls, messages):
        # 使用传入的 messages 参数创建并返回 ChatPromptTemplate 实例
        return cls(messages=messages)
    # 使用提供的变量格式化模板，返回消息列表
    def format_messages(self, **kwargs):
        # 格式化所有消息并返回
        return self._format_all_messages(kwargs)    
    # 私有方法，提取所有模板中用到的输入变量名
    def _extract_input_variables(self):
        # 用集合保存变量名，防止重复
        variables = set()
        # 遍历所有消息模板/对象
        for msg in self.messages:
            # 如果元素是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                _, template_str = msg
                # 用PromptTemplate对象提取变量
                prompt = PromptTemplate.from_template(template_str)
                # 合并到集合中
                variables.update(prompt.input_variables)
            # 如果是BaseMessagePromptTemplate子类实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                variables.update(msg.prompt.input_variables)
            # 如果是占位符对象
            elif isinstance(msg, MessagesPlaceholder):
                variables.add(msg.variable_name)    
        # 返回所有变量名组成的列表
        return list(variables)

    # 根据输入变量格式化所有消息模板，返回ChatPromptValue对象
    def invoke(self, input_variables):
        # 对消息模板进行实际变量填充
        formatted_messages = self._format_all_messages(input_variables)
        # 封装成ChatPromptValue对象返回
        return ChatPromptValue(messages=formatted_messages)

    # 将所有消息模板格式化并转换为消息对象列表
    def _format_all_messages(self, variables):
        # 新建列表保存格式化好的消息
        formatted_messages = []
        # 遍历每一个消息模板/对象
        for msg in self.messages:
            # 若是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                role, template_str = msg
                # 创建PromptTemplate模板并填充变量
                prompt = PromptTemplate.from_template(template_str)
                content = prompt.format(**variables)
                # 根据角色字符串生成对应的消息对象
                formatted_messages.append(self._create_message_from_role(role, content))
            # 如果是BaseMessagePromptTemplate的实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                # 调用BaseMessagePromptTemplate的format方法，返回消息对象
                formatted_messages.append(msg.format(**variables))    
            # 如果是占位符对象
            elif isinstance(msg, MessagesPlaceholder):
                placeholder_messages = self._coerce_placeholder_value(
                    msg.variable_name, variables.get(msg.variable_name)
                )
                formatted_messages.extend(placeholder_messages)    
            else:
                # 如不是模板，直接加入
                formatted_messages.append(msg)
        # 返回所有格式化的消息对象列表
        return formatted_messages
     # 处理占位符对象的值，返回消息对象列表
    def _coerce_placeholder_value(self, variable_name, value):
        # 如果未传入变量，抛出异常
        if value is None:
            raise ValueError(f"MessagesPlaceholder '{variable_name}' 对应变量缺失")
        # 如果是ChatPromptValue实例，转换为消息列表
        if isinstance(value, ChatPromptValue):
            return value.to_messages()
        # 如果已经是消息对象/结构列表，则依次转换
        if isinstance(value, list):
            return [self._coerce_single_message(item) for item in value]
        # 其他情况尝试单个转换
        return [self._coerce_single_message(value)]
        # 单个原始值转换为消息对象
    def _coerce_single_message(self, value):
        # 已是有效消息类型，直接返回
        if isinstance(value, (SystemMessage, HumanMessage, AIMessage)):
            return value
        # 有type和content属性，也当消息对象直接返回
        if hasattr(value, "type") and hasattr(value, "content"):
            return value
        # 字符串变为人类消息
        if isinstance(value, str):
            return HumanMessage(content=value)
        # (role, content)元组转为指定角色的消息
        if isinstance(value, tuple) and len(value) == 2:
            # 解包元组
            role, content = value
            # 根据role和content生成对应的消息对象
            return self._create_message_from_role(role, content)
        # 字典，默认user角色
        if isinstance(value, dict):
            # 获取role和content
            role = value.get("role", "user")
            # 获取content
            content = value.get("content", "")
            # 根据role和content生成对应的消息对象
            return self._create_message_from_role(role, content)
        # 其他无法识别类型，抛出异常
        raise TypeError("无法将占位符内容转换为消息")    
    # 辅助方法：根据角色和内容生成对应的标准消息对象
    def _create_message_from_role(self, role, content):
        # 角色字符串转小写做归一化
        normalized_role = role.lower()
        # 如果是system角色，返回SystemMessage对象
        if normalized_role == "system":
            return SystemMessage(content=content)
        # 如果是human或user角色，返回HumanMessage对象
        if normalized_role in ("human", "user"):
            return HumanMessage(content=content)
        # 如果是ai或assistant角色，返回AIMessage对象
        if normalized_role in ("ai", "assistant"):
            return AIMessage(content=content)
        # 如果角色未知，则抛出异常
        raise ValueError(f"未知的消息角色: {role}")    

# 定义基础消息提示词模板类
class BaseMessagePromptTemplate:
    # 基础消息提示词模板类声明
    """基础消息提示词模板类"""

    # 构造函数，必须传入PromptTemplate实例
    def __init__(self, prompt: PromptTemplate):
        # 将PromptTemplate实例保存在self.prompt属性中
        self.prompt = prompt

    # 工厂方法，利用模板字符串创建类实例
    @classmethod
    def from_template(cls, template: str):
        # 通过模板字符串创建PromptTemplate对象
        prompt = PromptTemplate.from_template(template)
        # 用生成的PromptTemplate创建本类实例并返回
        return cls(prompt=prompt)

    # 格式化当前模板，返回消息对象
    def format(self, **kwargs):
        # 使用PromptTemplate格式化内容，得到最终文本
        content = self.prompt.format(**kwargs)
        # 调用子类实现的方法将文本转换为对应类型消息对象
        return self._create_message(content)

    # 抽象方法，子类必须实现，用于生成特定类型的消息对象
    def _create_message(self, content):
        raise NotImplementedError

# 系统消息提示词模板类，继承自BaseMessagePromptTemplate
class SystemMessagePromptTemplate(BaseMessagePromptTemplate):
    # 系统消息提示词模板说明
    """系统消息提示词模板"""

    # 实现父类的_create_message方法，返回系统消息对象
    def _create_message(self, content):
        # 创建并返回SystemMessage对象，内容为content
        return SystemMessage(content=content)

# 人类消息提示词模板类，继承自BaseMessagePromptTemplate
class HumanMessagePromptTemplate(BaseMessagePromptTemplate):
    # 人类消息提示词模板说明
    """人类消息提示词模板"""

    # 实现父类的_create_message方法，返回人类消息对象
    def _create_message(self, content):
        # 创建并返回HumanMessage对象，内容为content
        return HumanMessage(content=content)

# AI消息提示词模板类，继承自BaseMessagePromptTemplate
class AIMessagePromptTemplate(BaseMessagePromptTemplate):
    # AI消息提示词模板说明
    """AI消息提示词模板"""

    # 实现父类的_create_message方法，返回AI消息对象
    def _create_message(self, content):
        # 创建并返回AIMessage对象，内容为content
        return AIMessage(content=content)

# 定义动态消息列表占位符类
class MessagesPlaceholder:
    # 在聊天模板中插入动态消息列表的占位符
    """在聊天模板中插入动态消息列表的占位符"""

    # 构造方法，存储变量名
    def __init__(self, variable_name: str):
        self.variable_name = variable_name
+# 定义 FewShotPromptTemplate 类，用于构建 few-shot 提示词模板
+class FewShotPromptTemplate:
+   # 文档字符串：说明该类用于构造 few-shot 提示词的模板
+   """用于构造 few-shot 提示词的模板"""
+
+   # 构造方法，初始化类的各种属性
+   def __init__(
+       self,
+       *,
+       examples: list[dict] = None,             # 示例列表，元素为字典类型
+       example_prompt: PromptTemplate | str,     # 示例模板，可以是 PromptTemplate 对象或字符串
+       prefix: str = "",                        # few-shot 提示词的前缀内容
+       suffix: str = "",                        # few-shot 提示词的后缀内容
+       example_separator: str = "\n\n",         # 每个示例之间用的分隔符
+       input_variables: list[str] | None = None,# 输入变量列表
+   ):
+       # 如果未传入 examples，默认使用空列表
+       self.examples = examples or []
+       # 判断 example_prompt 是否为 PromptTemplate 类型
+       if isinstance(example_prompt, PromptTemplate):
+           # 如果是 PromptTemplate，直接赋值
+           self.example_prompt = example_prompt
+       else:
+           # 如果是字符串，则先用 from_template 创建 PromptTemplate 再赋值
+           self.example_prompt = PromptTemplate.from_template(example_prompt)
+       # 保存前缀内容
+       self.prefix = prefix
+       # 保存后缀内容
+       self.suffix = suffix
+       # 保存示例分隔符
+       self.example_separator = example_separator
+       # 如果未指定输入变量，则自动根据前后缀推断变量名
+       self.input_variables = input_variables or self._infer_input_variables()
+
+   # 私有方法：推断前缀和后缀出现的模板变量名
+   def _infer_input_variables(self) -> list[str]:
+       # 新建一个集合用于保存变量名（去重）
+       variables = set()
+       # 提取 prefix 中引用的变量名
+       variables.update(self._extract_vars(self.prefix))
+       # 提取 suffix 中引用的变量名
+       variables.update(self._extract_vars(self.suffix))
+       # 转换为列表返回
+       return list(variables)
+
+   # 私有方法：提取文本中所有花括号包裹的模板变量名
+   def _extract_vars(self, text: str) -> list[str]:
+       # 如果输入为空字符串，直接返回空列表
+       if not text:
+           return []
+       # 定义正则表达式，匹配 {变量名} 或 {变量名:格式}
+       pattern = r"\{([^}:]+)(?::[^}]+)?\}"
+       # 使用 re.findall 提取所有变量名
+       matches = re.findall(pattern, text)
+       # 去重并保持顺序返回变量名列表
+       return list(dict.fromkeys(matches))
+
+   # 格式化 few-shot 提示词，返回完整字符串
+   def format(self, **kwargs) -> str:
+       """
+       根据传入的变量生成完整的 few-shot 提示词文本
+
+       Args:
+           **kwargs: 输入变量，可选，供示例选择
+       """
+       # 判断必需的变量是否全部传入，缺失时抛异常
+       missing = set(self.input_variables) - set(kwargs.keys())
+       if missing:
+           raise ValueError(f"缺少必需的变量: {missing}")
+
+       # 新建 parts 列表，用于拼接完整提示词的各部分内容
+       parts: list[str] = []
+       # 如果前缀不为空，格式化后加入 parts
+       if self.prefix:
+           parts.append(self._format_text(self.prefix, **kwargs))
+       # 调用 format_examples 得到所有示例的字符串，并用分隔符拼接在一起
+       example_block = self.example_separator.join(self.format_examples())
+       # 如果 example_block 不为空字符串，加入 parts
+       if example_block:
+           parts.append(example_block)
+       # 如果后缀不为空，格式化后加入 parts
+       if self.suffix:
+           parts.append(self._format_text(self.suffix, **kwargs))
+       # 用示例分隔符连接所有组成部分，过滤空字符串
+       return self.example_separator.join(part for part in parts if part)
+
+   # 格式化所有示例，返回字符串列表
+   def format_examples(self, input_variables: dict = None) -> list[str]:
+       """
+       返回格式化后的示例字符串列表
+
+       Args:
+           input_variables: 输入变量字典（预留，将来可以用于定制每个示例的选择）
+       """
+       # 新建存放格式化后示例的列表
+       formatted = []
+       # 遍历 every example 字典
+       for example in self.examples:
+           # 用 example_prompt 对当前示例格式化
+           formatted.append(self.example_prompt.format(**example))
+       # 返回格式化后的所有示例字符串列表
+       return formatted
+
+   # 私有方法：用 PromptTemplate 对 text 进行格式化
+   def _format_text(self, text: str, **kwargs) -> str:
+       # 先创建 PromptTemplate 实例
+       temp_prompt = PromptTemplate.from_template(text)
+       # 用传入参数格式化
+       return temp_prompt.format(**kwargs)

8.3 类 #

8.3.1 类说明 #

8.3.2 类图 #

8.3.3 调用关系图 #

8.3.4 数据流转过程 #

1. 示例数据定义

   • 创建示例字典列表，每个字典包含示例的输入和输出
   • 示例：[{"question": "1 plus 1 等于多少？", "answer": "答案是 2。"}, ...]

2. 示例模板创建

   • 使用 PromptTemplate.from_template() 创建示例格式模板
   • 模板定义每个示例的显示格式："示例问题：{question}\n示例回答：{answer}"
   • 自动提取变量：["question", "answer"]

3. Few-shot 模板创建

   • 使用 FewShotPromptTemplate 创建模板，包含：
     • examples：示例数据列表
     • example_prompt：示例格式模板（PromptTemplate 对象）
     • prefix：前缀文本（可选，可包含变量）
     • suffix：后缀文本（可选，可包含变量）
     • input_variables：输入变量列表（如果未指定，会从前后缀自动推断）
     • example_separator：示例分隔符（默认 "\n\n"）

4. 模板格式化

   • 调用 few_shot_prompt.format(user_question="...")
   • 处理流程：
     • 检查变量完整性
     • 格式化前缀：使用 _format_text() 创建临时 PromptTemplate 并格式化
     • 格式化示例：调用 format_examples()，遍历每个示例，使用 example_prompt.format(**example) 格式化
     • 格式化后缀：使用 _format_text() 创建临时 PromptTemplate 并格式化
     • 组合结果：使用 example_separator 连接所有部分

5. 最终提示词结构

   你是一个擅长算术的 AI 助手。以下是一些示例：
   
   示例问题：1 plus 1 等于多少？
   示例回答：答案是 2。
   
   示例问题：2 plus 2 等于多少？
   示例回答：答案是 4。
   
   请回答用户问题：3 plus 5 等于多少？
   AI：

6. 模型调用

   • 将格式化后的字符串传给 ChatOpenAI.invoke()
   • ChatOpenAI 内部将字符串转换为消息格式并调用 API
   • 返回 AIMessage 对象

9.load_prompt #

load_prompt 是一个用于从 JSON 文件中加载提示词模板的工具函数，能够让你将提示词编写、管理在外部文件，而不是硬编码在代码内，极大提升了 prompt 工程的灵活性与可维护性。

核心场景

• 将提示词设计为可配置文件，方便非开发人员修改和维护
• 支持“一键切换/复用”不同场景/风格的 prompt，无需更改主代码
• 适合团队协作、A/B 测试等场合

典型调用流程

1. 新建 prompt 配置文件（如prompt_template.json）：

   {
       "_type": "prompt",
       "template": "你是一个乐于助人的AI助手。用户问：{question}，请回答："
   }

2. 代码中加载模板

   from smartchain.prompts import load_prompt
   
   prompt_template = load_prompt("prompt_template.json", encoding="utf-8")

3. 填写变量，格式化提示词

   prompt = prompt_template.format(question="什么是人工智能？")
   print(prompt)  # 结果：你是一个乐于助人的AI助手。用户问：什么是人工智能？，请回答：

4. 与 ChatOpenAI 联动调用大模型，得到回复

load_prompt 主要机制

• 支持的文件类型：目前仅支持 .json 格式
• 配置规范：文件内需包含
  • _type 字段（值为 "prompt"）
  • template 具体的字符串模板内容
• 加载流程
  1. 检查文件是否存在、扩展名是否合法
  2. 读取 JSON，校验 _type 字段
  3. 读取并解析模板字符串
  4. 返回 PromptTemplate 实例对象

典型易错点

• ⋆ 文件名后缀须为 .json
• ⋆ JSON 文件需有 "template" 字段，且可用花括号变量（如 {question}）
• ⋆ 文件内容须为 UTF-8 编码（可自定义 encoding）

9.1. 9.load_prompt.py #

9.load_prompt.py

#from langchain_core.prompts import load_prompt
#from langchain_openai import ChatOpenAI

from smartchain.chat_models import ChatOpenAI
from smartchain.prompts import load_prompt

# 从 JSON 文件加载提示词模板
prompt_template = load_prompt("prompt_template.json",encoding="utf-8")

# 创建 ChatOpenAI 实例
llm = ChatOpenAI(model="gpt-4o")

# 使用加载的模板格式化提示词
formatted_prompt = prompt_template.format(question="什么是人工智能？")

print("格式化后的提示词：")
print(formatted_prompt)

# 调用模型生成回复
result = llm.invoke(formatted_prompt)
print("AI 回复：")
print(result.content)

9.2 prompt_template.json #

prompt_template.json

{
    "_type": "prompt",
    "template": "你是一个乐于助人的AI助手。用户问：{question}，请回答："
}

9.3. prompts.py #

smartchain/prompts.py

# 导入正则表达式模块，用于变量提取
import re
+import json
+from pathlib import Path
from .messages import SystemMessage, HumanMessage, AIMessage
# 定义提示词模板类
class PromptTemplate:
    # 类说明文档，描述用途
    """提示词模板类，用于格式化字符串模板"""

    # 构造方法，初始化模板实例
    def __init__(self, template: str):
        # 保存模板字符串到实例属性
        self.template = template
        # 调用内部方法提取模板中的变量名列表
        input_variables = self._extract_variables(template)
        # 将变量名列表分配给实例属性
        self.input_variables = input_variables

    # 类方法：从模板字符串生成 PromptTemplate 实例
    @classmethod
    def from_template(cls, template: str):
        # 返回用 template 实例化的 PromptTemplate 对象
        return cls(template=template)

    # 格式化填充模板中的变量
    def format(self, **kwargs):
        # 计算模板中缺失但未传入的变量名集合
        missing_vars = set(self.input_variables) - set(kwargs.keys())
        # 如果存在缺失变量则抛出异常，提示哪些变量缺失
        if missing_vars:
            raise ValueError(f"缺少必需的变量: {missing_vars}")
        # 使用传入参数填充模板并返回格式化后的字符串
        return self.template.format(**kwargs)

    # 内部方法：从模板字符串中提取变量名
    def _extract_variables(self, template: str):
        # 定义正则表达式，匹配花括号中的变量名（冒号前的部分）
        pattern = r'\{([^}:]+)(?::[^}]+)?\}'
        # 查找所有符合 pattern 的变量名，返回匹配结果列表
        matches = re.findall(pattern, template)
        # 利用 dict 去重并保持顺序，最后转为列表返回
        return list(dict.fromkeys(matches))

# 定义一个用于存放格式化后的消息的类
class ChatPromptValue:
    # 聊天提示词值类，包含格式化后的消息列表
    """聊天提示词值类，包含格式化后的消息列表"""

    # 构造函数，接收一个消息对象列表
    def __init__(self, messages):
        # 保存消息列表到实例变量
        self.messages = messages

    # 将消息对象列表转为字符串，方便展示
    def to_string(self):
        # 新建一个用于存放字符串的列表
        parts = []
        # 遍历每个消息对象
        for msg in self.messages:
            # 如果消息对象有type和content属性
            if hasattr(msg, 'type') and hasattr(msg, 'content'):
                # 定义消息角色的映射关系
                role_map = {
                    "system": "System",
                    "human": "Human",
                    "ai": "AI"
                }
                # 获取对应的角色字符串，没有则首字母大写
                role = role_map.get(msg.type, msg.type.capitalize())
                # 拼接角色和消息内容
                parts.append(f"{role}: {msg.content}")
            else:
                # 如果不是标准消息对象，则直接转为字符串
                parts.append(str(msg))
        # 将所有消息用换行符拼接起来组成一个字符串
        return "\n".join(parts)

    # 返回消息对象列表本身
    def to_messages(self):
        # 直接返回消息列表
        return self.messages

# 定义用于处理多轮对话消息模板的类
class ChatPromptTemplate:
    # 聊天提示词模板类，用于创建多轮对话的提示词
    """聊天提示词模板类，用于创建多轮对话的提示词"""

    # 构造方法，接收一个消息模板/对象的列表
    def __init__(self, messages):
        # 保存消息模板/对象列表
        self.messages = messages
        # 提取所有输入变量并存入实例变量
        self.input_variables = self._extract_input_variables()
    # 定义一个类方法，用于通过消息对象列表创建 ChatPromptTemplate 实例
    @classmethod
    def from_messages(cls, messages):
        # 使用传入的 messages 参数创建并返回 ChatPromptTemplate 实例
        return cls(messages=messages)
    # 使用提供的变量格式化模板，返回消息列表
    def format_messages(self, **kwargs):
        # 格式化所有消息并返回
        return self._format_all_messages(kwargs)    
    # 私有方法，提取所有模板中用到的输入变量名
    def _extract_input_variables(self):
        # 用集合保存变量名，防止重复
        variables = set()
        # 遍历所有消息模板/对象
        for msg in self.messages:
            # 如果元素是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                _, template_str = msg
                # 用PromptTemplate对象提取变量
                prompt = PromptTemplate.from_template(template_str)
                # 合并到集合中
                variables.update(prompt.input_variables)
            # 如果是BaseMessagePromptTemplate子类实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                variables.update(msg.prompt.input_variables)
            # 如果是占位符对象
            elif isinstance(msg, MessagesPlaceholder):
                variables.add(msg.variable_name)    
        # 返回所有变量名组成的列表
        return list(variables)

    # 根据输入变量格式化所有消息模板，返回ChatPromptValue对象
    def invoke(self, input_variables):
        # 对消息模板进行实际变量填充
        formatted_messages = self._format_all_messages(input_variables)
        # 封装成ChatPromptValue对象返回
        return ChatPromptValue(messages=formatted_messages)

    # 将所有消息模板格式化并转换为消息对象列表
    def _format_all_messages(self, variables):
        # 新建列表保存格式化好的消息
        formatted_messages = []
        # 遍历每一个消息模板/对象
        for msg in self.messages:
            # 若是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                role, template_str = msg
                # 创建PromptTemplate模板并填充变量
                prompt = PromptTemplate.from_template(template_str)
                content = prompt.format(**variables)
                # 根据角色字符串生成对应的消息对象
                formatted_messages.append(self._create_message_from_role(role, content))
            # 如果是BaseMessagePromptTemplate的实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                # 调用BaseMessagePromptTemplate的format方法，返回消息对象
                formatted_messages.append(msg.format(**variables))    
            # 如果是占位符对象
            elif isinstance(msg, MessagesPlaceholder):
                placeholder_messages = self._coerce_placeholder_value(
                    msg.variable_name, variables.get(msg.variable_name)
                )
                formatted_messages.extend(placeholder_messages)    
            else:
                # 如不是模板，直接加入
                formatted_messages.append(msg)
        # 返回所有格式化的消息对象列表
        return formatted_messages
     # 处理占位符对象的值，返回消息对象列表
    def _coerce_placeholder_value(self, variable_name, value):
        # 如果未传入变量，抛出异常
        if value is None:
            raise ValueError(f"MessagesPlaceholder '{variable_name}' 对应变量缺失")
        # 如果是ChatPromptValue实例，转换为消息列表
        if isinstance(value, ChatPromptValue):
            return value.to_messages()
        # 如果已经是消息对象/结构列表，则依次转换
        if isinstance(value, list):
            return [self._coerce_single_message(item) for item in value]
        # 其他情况尝试单个转换
        return [self._coerce_single_message(value)]
        # 单个原始值转换为消息对象
    def _coerce_single_message(self, value):
        # 已是有效消息类型，直接返回
        if isinstance(value, (SystemMessage, HumanMessage, AIMessage)):
            return value
        # 有type和content属性，也当消息对象直接返回
        if hasattr(value, "type") and hasattr(value, "content"):
            return value
        # 字符串变为人类消息
        if isinstance(value, str):
            return HumanMessage(content=value)
        # (role, content)元组转为指定角色的消息
        if isinstance(value, tuple) and len(value) == 2:
            # 解包元组
            role, content = value
            # 根据role和content生成对应的消息对象
            return self._create_message_from_role(role, content)
        # 字典，默认user角色
        if isinstance(value, dict):
            # 获取role和content
            role = value.get("role", "user")
            # 获取content
            content = value.get("content", "")
            # 根据role和content生成对应的消息对象
            return self._create_message_from_role(role, content)
        # 其他无法识别类型，抛出异常
        raise TypeError("无法将占位符内容转换为消息")    
    # 辅助方法：根据角色和内容生成对应的标准消息对象
    def _create_message_from_role(self, role, content):
        # 角色字符串转小写做归一化
        normalized_role = role.lower()
        # 如果是system角色，返回SystemMessage对象
        if normalized_role == "system":
            return SystemMessage(content=content)
        # 如果是human或user角色，返回HumanMessage对象
        if normalized_role in ("human", "user"):
            return HumanMessage(content=content)
        # 如果是ai或assistant角色，返回AIMessage对象
        if normalized_role in ("ai", "assistant"):
            return AIMessage(content=content)
        # 如果角色未知，则抛出异常
        raise ValueError(f"未知的消息角色: {role}")    

# 定义基础消息提示词模板类
class BaseMessagePromptTemplate:
    # 基础消息提示词模板类声明
    """基础消息提示词模板类"""

    # 构造函数，必须传入PromptTemplate实例
    def __init__(self, prompt: PromptTemplate):
        # 将PromptTemplate实例保存在self.prompt属性中
        self.prompt = prompt

    # 工厂方法，利用模板字符串创建类实例
    @classmethod
    def from_template(cls, template: str):
        # 通过模板字符串创建PromptTemplate对象
        prompt = PromptTemplate.from_template(template)
        # 用生成的PromptTemplate创建本类实例并返回
        return cls(prompt=prompt)

    # 格式化当前模板，返回消息对象
    def format(self, **kwargs):
        # 使用PromptTemplate格式化内容，得到最终文本
        content = self.prompt.format(**kwargs)
        # 调用子类实现的方法将文本转换为对应类型消息对象
        return self._create_message(content)

    # 抽象方法，子类必须实现，用于生成特定类型的消息对象
    def _create_message(self, content):
        raise NotImplementedError

# 系统消息提示词模板类，继承自BaseMessagePromptTemplate
class SystemMessagePromptTemplate(BaseMessagePromptTemplate):
    # 系统消息提示词模板说明
    """系统消息提示词模板"""

    # 实现父类的_create_message方法，返回系统消息对象
    def _create_message(self, content):
        # 创建并返回SystemMessage对象，内容为content
        return SystemMessage(content=content)

# 人类消息提示词模板类，继承自BaseMessagePromptTemplate
class HumanMessagePromptTemplate(BaseMessagePromptTemplate):
    # 人类消息提示词模板说明
    """人类消息提示词模板"""

    # 实现父类的_create_message方法，返回人类消息对象
    def _create_message(self, content):
        # 创建并返回HumanMessage对象，内容为content
        return HumanMessage(content=content)

# AI消息提示词模板类，继承自BaseMessagePromptTemplate
class AIMessagePromptTemplate(BaseMessagePromptTemplate):
    # AI消息提示词模板说明
    """AI消息提示词模板"""

    # 实现父类的_create_message方法，返回AI消息对象
    def _create_message(self, content):
        # 创建并返回AIMessage对象，内容为content
        return AIMessage(content=content)

# 定义动态消息列表占位符类
class MessagesPlaceholder:
    # 在聊天模板中插入动态消息列表的占位符
    """在聊天模板中插入动态消息列表的占位符"""

    # 构造方法，存储变量名
    def __init__(self, variable_name: str):
        self.variable_name = variable_name
# 定义 FewShotPromptTemplate 类，用于构建 few-shot 提示词模板
class FewShotPromptTemplate:
    # 文档字符串：说明该类用于构造 few-shot 提示词的模板
    """用于构造 few-shot 提示词的模板"""

    # 构造方法，初始化类的各种属性
    def __init__(
        self,
        *,
        examples: list[dict] = None,             # 示例列表，元素为字典类型
        example_prompt: PromptTemplate | str,     # 示例模板，可以是 PromptTemplate 对象或字符串
        prefix: str = "",                        # few-shot 提示词的前缀内容
        suffix: str = "",                        # few-shot 提示词的后缀内容
        example_separator: str = "\n\n",         # 每个示例之间用的分隔符
        input_variables: list[str] | None = None,# 输入变量列表
    ):
        # 如果未传入 examples，默认使用空列表
        self.examples = examples or []
        # 判断 example_prompt 是否为 PromptTemplate 类型
        if isinstance(example_prompt, PromptTemplate):
            # 如果是 PromptTemplate，直接赋值
            self.example_prompt = example_prompt
        else:
            # 如果是字符串，则先用 from_template 创建 PromptTemplate 再赋值
            self.example_prompt = PromptTemplate.from_template(example_prompt)
        # 保存前缀内容
        self.prefix = prefix
        # 保存后缀内容
        self.suffix = suffix
        # 保存示例分隔符
        self.example_separator = example_separator
        # 如果未指定输入变量，则自动根据前后缀推断变量名
        self.input_variables = input_variables or self._infer_input_variables()

    # 私有方法：推断前缀和后缀出现的模板变量名
    def _infer_input_variables(self) -> list[str]:
        # 新建一个集合用于保存变量名（去重）
        variables = set()
        # 提取 prefix 中引用的变量名
        variables.update(self._extract_vars(self.prefix))
        # 提取 suffix 中引用的变量名
        variables.update(self._extract_vars(self.suffix))
        # 转换为列表返回
        return list(variables)

    # 私有方法：提取文本中所有花括号包裹的模板变量名
    def _extract_vars(self, text: str) -> list[str]:
        # 如果输入为空字符串，直接返回空列表
        if not text:
            return []
        # 定义正则表达式，匹配 {变量名} 或 {变量名:格式}
        pattern = r"\{([^}:]+)(?::[^}]+)?\}"
        # 使用 re.findall 提取所有变量名
        matches = re.findall(pattern, text)
        # 去重并保持顺序返回变量名列表
        return list(dict.fromkeys(matches))

    # 格式化 few-shot 提示词，返回完整字符串
    def format(self, **kwargs) -> str:
        """
        根据传入的变量生成完整的 few-shot 提示词文本

        Args:
            **kwargs: 输入变量，可选，供示例选择
        """
        # 判断必需的变量是否全部传入，缺失时抛异常
        missing = set(self.input_variables) - set(kwargs.keys())
        if missing:
            raise ValueError(f"缺少必需的变量: {missing}")

        # 新建 parts 列表，用于拼接完整提示词的各部分内容
        parts: list[str] = []
        # 如果前缀不为空，格式化后加入 parts
        if self.prefix:
            parts.append(self._format_text(self.prefix, **kwargs))
        # 调用 format_examples 得到所有示例的字符串，并用分隔符拼接在一起
        example_block = self.example_separator.join(self.format_examples())
        # 如果 example_block 不为空字符串，加入 parts
        if example_block:
            parts.append(example_block)
        # 如果后缀不为空，格式化后加入 parts
        if self.suffix:
            parts.append(self._format_text(self.suffix, **kwargs))
        # 用示例分隔符连接所有组成部分，过滤空字符串
        return self.example_separator.join(part for part in parts if part)

    # 格式化所有示例，返回字符串列表
    def format_examples(self, input_variables: dict = None) -> list[str]:
        """
        返回格式化后的示例字符串列表

        Args:
            input_variables: 输入变量字典（预留，将来可以用于定制每个示例的选择）
        """
        # 新建存放格式化后示例的列表
        formatted = []
        # 遍历 every example 字典
        for example in self.examples:
            # 用 example_prompt 对当前示例格式化
            formatted.append(self.example_prompt.format(**example))
        # 返回格式化后的所有示例字符串列表
        return formatted

    # 私有方法：用 PromptTemplate 对 text 进行格式化
    def _format_text(self, text: str, **kwargs) -> str:
        # 先创建 PromptTemplate 实例
        temp_prompt = PromptTemplate.from_template(text)
        # 用传入参数格式化
        return temp_prompt.format(**kwargs)
+       
+# 定义一个从文件加载提示词模板的函数
+def load_prompt(path: str | Path, encoding: str | None = None) -> PromptTemplate:
+   # 将传入的路径参数转换为 Path 对象，方便后续进行文件操作
+   file_path = Path(path)
+   # 判断文件是否存在，如果不存在则抛出 FileNotFoundError 异常
+   if not file_path.exists():
+       raise FileNotFoundError(f"提示词文件不存在: {path}")
+   # 判断文件扩展名是否为 .json，如果不是则抛出 ValueError 异常
+   if file_path.suffix != ".json":
+       raise ValueError(f"只支持 .json 格式文件，当前文件: {file_path.suffix}")
+   # 打开文件，使用指定编码（一般为 utf-8），并读取 JSON 配置信息到 config 变量
+   with file_path.open(encoding=encoding) as f:
+       config = json.load(f)
+   # 从配置字典中获取 "_type" 字段，不存在则默认值为 "prompt"
+   config_type = config.get("_type", "prompt")
+   # 校验 _type 字段是否为 "prompt"，如果不是则抛出异常
+   if config_type != "prompt":
+       raise ValueError(f"不支持的提示词类型: {config_type}，当前只支持 'prompt'")
+   # 从配置中获取模板字符串 template，如果不存在则抛出异常
+   template = config.get("template")
+   if template is None:
+       raise ValueError("配置文件中缺少 'template' 字段")
+   # 使用读取到的模板字符串创建 PromptTemplate 实例并返回
+   return PromptTemplate.from_template(template)

10.partial #

• PromptTemplate 的 partial（部分填充）机制让你可以将模板中的部分变量提前锁定并赋值，后续只需填写剩余变量即可。
• 这对于多个问题格式类似、只变化个别人/参数的场景非常高效。

关键方法与用法说明：

• PromptTemplate.from_template(template_str)

  • 用于通过字符串模板快速创建一个 PromptTemplate 实例。
  • 自动检测并整理出所有待填充变量（如 {role}、{user_name}、{question}）。

• partial(**kwargs)

  • 可链式/批量指定部分变量（如提前将 role 固定为“AI助手”），返回带有剩余未填充变量的新模板对象。
  • .input_variables：查看当前剩余待填充的变量列表。
  • .partial_variables：查看已锁定/注入的变量及对应值。

• format(**kwargs)

  • 最终将所有变量补齐，得到完整的格式化字符串。

例如，先定义较通用的模板，然后通过 partial 锁定固定维度（如 AI 角色或特定用户），即可大大简化后续模板使用与变量输入的复杂度。这一范式非常适用于多重场景复用的 prompt 工程。

10.1. 10.partial.py #

10.partial.py

# 导入 PromptTemplate 类（用于提示词模板）和 ChatOpenAI（用于调用大模型）
#from langchain_core.prompts import PromptTemplate
#from langchain_openai import ChatOpenAI

from smartchain.chat_models import ChatOpenAI
from smartchain.prompts import PromptTemplate

# 使用 from_template 方法创建一个包含三个变量（role、user_name、question）的提示词模板
template = PromptTemplate.from_template(
    "你是一个{role}。用户{user_name}问：{question}，请回答："
)

# 打印模板中所有待填充的输入变量
print("原始模板的输入变量：", template.input_variables)

# 使用 partial 方法部分填充变量（相当于锁定一部分内容，减少后续格式化时需提供的内容）
# 这里预先填充 role 和 user_name，只剩下 question 需要后续提供
partial_template = template.partial(role="AI助手", user_name="张三")

# 打印部分填充后剩余的输入变量
print("部分填充后的输入变量：", partial_template.input_variables)
# 打印已经部分填充的变量及其对应值
print("部分填充的变量：", partial_template.partial_variables)

# 对剩余未填充的变量 question 进行格式化，获得完整的提示词字符串
formatted_prompt = partial_template.format(question="什么是人工智能？")
# 打印最终格式化好的提示词字符串
print("格式化后的提示词：")
print(formatted_prompt)

# 创建 ChatOpenAI 的实例（指定模型为 gpt-4o）
llm = ChatOpenAI(model="gpt-4o")
# 将格式化后的提示词作为输入，调用大模型生成回复
result = llm.invoke(formatted_prompt)
# 打印 AI 回复内容
print("AI 回复：")
print(result.content)

10.2. prompts.py #

smartchain/prompts.py

# 导入正则表达式模块，用于变量提取
import re
import json
from pathlib import Path
from .messages import SystemMessage, HumanMessage, AIMessage
# 定义提示词模板类
class PromptTemplate:
    # 类说明文档，描述用途
    """提示词模板类，用于格式化字符串模板"""

    # 构造方法，初始化模板实例
+   def __init__(self, template: str, partial_variables: dict = None):
        # 保存模板字符串到实例属性
        self.template = template
+       # 保存部分变量（已预填充的变量）
+       self.partial_variables = partial_variables or {}
        # 调用内部方法提取模板中的变量名列表
+       all_variables = self._extract_variables(template)
+       # 从所有变量中排除已部分填充的变量
+       self.input_variables = [v for v in all_variables if v not in self.partial_variables]

    # 类方法：从模板字符串生成 PromptTemplate 实例
    @classmethod
    def from_template(cls, template: str):
        # 返回用 template 实例化的 PromptTemplate 对象
        return cls(template=template)

    # 格式化填充模板中的变量
    def format(self, **kwargs):
+       # 合并部分变量和用户提供的变量
+       all_vars = {**self.partial_variables, **kwargs}
        # 计算模板中缺失但未传入的变量名集合
        missing_vars = set(self.input_variables) - set(kwargs.keys())
        # 如果存在缺失变量则抛出异常，提示哪些变量缺失
        if missing_vars:
            raise ValueError(f"缺少必需的变量: {missing_vars}")
        # 使用传入参数填充模板并返回格式化后的字符串
+       return self.template.format(**all_vars)

    # 内部方法：从模板字符串中提取变量名
    def _extract_variables(self, template: str):
        # 定义正则表达式，匹配花括号中的变量名（冒号前的部分）
        pattern = r'\{([^}:]+)(?::[^}]+)?\}'
        # 查找所有符合 pattern 的变量名，返回匹配结果列表
        matches = re.findall(pattern, template)
        # 利用 dict 去重并保持顺序，最后转为列表返回
        return list(dict.fromkeys(matches))
+       # 定义部分填充模板变量的方法，返回新的模板实例
+   def partial(self, **kwargs):
+       """
+       部分填充模板变量，返回一个新的 PromptTemplate 实例
+       
+       Args:
+           **kwargs: 要部分填充的变量及其值
+       
+       Returns:
+           新的 PromptTemplate 实例，其中指定的变量已被填充
+       
+       示例:
+           template = PromptTemplate.from_template("你好，我叫{name}，我来自{city}")
+           partial_template = template.partial(name="张三")
+           # 现在只需要提供 city 参数
+           result = partial_template.format(city="北京")
+       """
+       # 合并现有对象的部分变量（partial_variables）和本次要填充的新变量
+       new_partial_variables = {**self.partial_variables, **kwargs}
+       # 使用原模板字符串和更新后的部分变量，创建新的 PromptTemplate 实例
+       new_template = PromptTemplate(
+           template=self.template,
+           partial_variables=new_partial_variables
+       )
+       # 返回新的 PromptTemplate 实例
+       return new_template    

# 定义一个用于存放格式化后的消息的类
class ChatPromptValue:
    # 聊天提示词值类，包含格式化后的消息列表
    """聊天提示词值类，包含格式化后的消息列表"""

    # 构造函数，接收一个消息对象列表
    def __init__(self, messages):
        # 保存消息列表到实例变量
        self.messages = messages

    # 将消息对象列表转为字符串，方便展示
    def to_string(self):
        # 新建一个用于存放字符串的列表
        parts = []
        # 遍历每个消息对象
        for msg in self.messages:
            # 如果消息对象有type和content属性
            if hasattr(msg, 'type') and hasattr(msg, 'content'):
                # 定义消息角色的映射关系
                role_map = {
                    "system": "System",
                    "human": "Human",
                    "ai": "AI"
                }
                # 获取对应的角色字符串，没有则首字母大写
                role = role_map.get(msg.type, msg.type.capitalize())
                # 拼接角色和消息内容
                parts.append(f"{role}: {msg.content}")
            else:
                # 如果不是标准消息对象，则直接转为字符串
                parts.append(str(msg))
        # 将所有消息用换行符拼接起来组成一个字符串
        return "\n".join(parts)

    # 返回消息对象列表本身
    def to_messages(self):
        # 直接返回消息列表
        return self.messages

# 定义用于处理多轮对话消息模板的类
class ChatPromptTemplate:
    # 聊天提示词模板类，用于创建多轮对话的提示词
    """聊天提示词模板类，用于创建多轮对话的提示词"""

    # 构造方法，接收一个消息模板/对象的列表
    def __init__(self, messages):
        # 保存消息模板/对象列表
        self.messages = messages
        # 提取所有输入变量并存入实例变量
        self.input_variables = self._extract_input_variables()
    # 定义一个类方法，用于通过消息对象列表创建 ChatPromptTemplate 实例
    @classmethod
    def from_messages(cls, messages):
        # 使用传入的 messages 参数创建并返回 ChatPromptTemplate 实例
        return cls(messages=messages)
    # 使用提供的变量格式化模板，返回消息列表
    def format_messages(self, **kwargs):
        # 格式化所有消息并返回
        return self._format_all_messages(kwargs)    
    # 私有方法，提取所有模板中用到的输入变量名
    def _extract_input_variables(self):
        # 用集合保存变量名，防止重复
        variables = set()
        # 遍历所有消息模板/对象
        for msg in self.messages:
            # 如果元素是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                _, template_str = msg
                # 用PromptTemplate对象提取变量
                prompt = PromptTemplate.from_template(template_str)
                # 合并到集合中
                variables.update(prompt.input_variables)
            # 如果是BaseMessagePromptTemplate子类实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                variables.update(msg.prompt.input_variables)
            # 如果是占位符对象
            elif isinstance(msg, MessagesPlaceholder):
                variables.add(msg.variable_name)    
        # 返回所有变量名组成的列表
        return list(variables)

    # 根据输入变量格式化所有消息模板，返回ChatPromptValue对象
    def invoke(self, input_variables):
        # 对消息模板进行实际变量填充
        formatted_messages = self._format_all_messages(input_variables)
        # 封装成ChatPromptValue对象返回
        return ChatPromptValue(messages=formatted_messages)

    # 将所有消息模板格式化并转换为消息对象列表
    def _format_all_messages(self, variables):
        # 新建列表保存格式化好的消息
        formatted_messages = []
        # 遍历每一个消息模板/对象
        for msg in self.messages:
            # 若是(role, template_str)元组
            if isinstance(msg, tuple) and len(msg) == 2:
                role, template_str = msg
                # 创建PromptTemplate模板并填充变量
                prompt = PromptTemplate.from_template(template_str)
                content = prompt.format(**variables)
                # 根据角色字符串生成对应的消息对象
                formatted_messages.append(self._create_message_from_role(role, content))
            # 如果是BaseMessagePromptTemplate的实例
            elif isinstance(msg, BaseMessagePromptTemplate):
                # 调用BaseMessagePromptTemplate的format方法，返回消息对象
                formatted_messages.append(msg.format(**variables))    
            # 如果是占位符对象
            elif isinstance(msg, MessagesPlaceholder):
                placeholder_messages = self._coerce_placeholder_value(
                    msg.variable_name, variables.get(msg.variable_name)
                )
                formatted_messages.extend(placeholder_messages)    
            else:
                # 如不是模板，直接加入
                formatted_messages.append(msg)
        # 返回所有格式化的消息对象列表
        return formatted_messages
     # 处理占位符对象的值，返回消息对象列表
    def _coerce_placeholder_value(self, variable_name, value):
        # 如果未传入变量，抛出异常
        if value is None:
            raise ValueError(f"MessagesPlaceholder '{variable_name}' 对应变量缺失")
        # 如果是ChatPromptValue实例，转换为消息列表
        if isinstance(value, ChatPromptValue):
            return value.to_messages()
        # 如果已经是消息对象/结构列表，则依次转换
        if isinstance(value, list):
            return [self._coerce_single_message(item) for item in value]
        # 其他情况尝试单个转换
        return [self._coerce_single_message(value)]
        # 单个原始值转换为消息对象
    def _coerce_single_message(self, value):
        # 已是有效消息类型，直接返回
        if isinstance(value, (SystemMessage, HumanMessage, AIMessage)):
            return value
        # 有type和content属性，也当消息对象直接返回
        if hasattr(value, "type") and hasattr(value, "content"):
            return value
        # 字符串变为人类消息
        if isinstance(value, str):
            return HumanMessage(content=value)
        # (role, content)元组转为指定角色的消息
        if isinstance(value, tuple) and len(value) == 2:
            # 解包元组
            role, content = value
            # 根据role和content生成对应的消息对象
            return self._create_message_from_role(role, content)
        # 字典，默认user角色
        if isinstance(value, dict):
            # 获取role和content
            role = value.get("role", "user")
            # 获取content
            content = value.get("content", "")
            # 根据role和content生成对应的消息对象
            return self._create_message_from_role(role, content)
        # 其他无法识别类型，抛出异常
        raise TypeError("无法将占位符内容转换为消息")    
    # 辅助方法：根据角色和内容生成对应的标准消息对象
    def _create_message_from_role(self, role, content):
        # 角色字符串转小写做归一化
        normalized_role = role.lower()
        # 如果是system角色，返回SystemMessage对象
        if normalized_role == "system":
            return SystemMessage(content=content)
        # 如果是human或user角色，返回HumanMessage对象
        if normalized_role in ("human", "user"):
            return HumanMessage(content=content)
        # 如果是ai或assistant角色，返回AIMessage对象
        if normalized_role in ("ai", "assistant"):
            return AIMessage(content=content)
        # 如果角色未知，则抛出异常
        raise ValueError(f"未知的消息角色: {role}")    

# 定义基础消息提示词模板类
class BaseMessagePromptTemplate:
    # 基础消息提示词模板类声明
    """基础消息提示词模板类"""

    # 构造函数，必须传入PromptTemplate实例
    def __init__(self, prompt: PromptTemplate):
        # 将PromptTemplate实例保存在self.prompt属性中
        self.prompt = prompt

    # 工厂方法，利用模板字符串创建类实例
    @classmethod
    def from_template(cls, template: str):
        # 通过模板字符串创建PromptTemplate对象
        prompt = PromptTemplate.from_template(template)
        # 用生成的PromptTemplate创建本类实例并返回
        return cls(prompt=prompt)

    # 格式化当前模板，返回消息对象
    def format(self, **kwargs):
        # 使用PromptTemplate格式化内容，得到最终文本
        content = self.prompt.format(**kwargs)
        # 调用子类实现的方法将文本转换为对应类型消息对象
        return self._create_message(content)

    # 抽象方法，子类必须实现，用于生成特定类型的消息对象
    def _create_message(self, content):
        raise NotImplementedError

# 系统消息提示词模板类，继承自BaseMessagePromptTemplate
class SystemMessagePromptTemplate(BaseMessagePromptTemplate):
    # 系统消息提示词模板说明
    """系统消息提示词模板"""

    # 实现父类的_create_message方法，返回系统消息对象
    def _create_message(self, content):
        # 创建并返回SystemMessage对象，内容为content
        return SystemMessage(content=content)

# 人类消息提示词模板类，继承自BaseMessagePromptTemplate
class HumanMessagePromptTemplate(BaseMessagePromptTemplate):
    # 人类消息提示词模板说明
    """人类消息提示词模板"""

    # 实现父类的_create_message方法，返回人类消息对象
    def _create_message(self, content):
        # 创建并返回HumanMessage对象，内容为content
        return HumanMessage(content=content)

# AI消息提示词模板类，继承自BaseMessagePromptTemplate
class AIMessagePromptTemplate(BaseMessagePromptTemplate):
    # AI消息提示词模板说明
    """AI消息提示词模板"""

    # 实现父类的_create_message方法，返回AI消息对象
    def _create_message(self, content):
        # 创建并返回AIMessage对象，内容为content
        return AIMessage(content=content)

# 定义动态消息列表占位符类
class MessagesPlaceholder:
    # 在聊天模板中插入动态消息列表的占位符
    """在聊天模板中插入动态消息列表的占位符"""

    # 构造方法，存储变量名
    def __init__(self, variable_name: str):
        self.variable_name = variable_name
# 定义 FewShotPromptTemplate 类，用于构建 few-shot 提示词模板
class FewShotPromptTemplate:
    # 文档字符串：说明该类用于构造 few-shot 提示词的模板
    """用于构造 few-shot 提示词的模板"""

    # 构造方法，初始化类的各种属性
    def __init__(
        self,
        *,
        examples: list[dict] = None,             # 示例列表，元素为字典类型
        example_prompt: PromptTemplate | str,     # 示例模板，可以是 PromptTemplate 对象或字符串
        prefix: str = "",                        # few-shot 提示词的前缀内容
        suffix: str = "",                        # few-shot 提示词的后缀内容
        example_separator: str = "\n\n",         # 每个示例之间用的分隔符
        input_variables: list[str] | None = None,# 输入变量列表
    ):
        # 如果未传入 examples，默认使用空列表
        self.examples = examples or []
        # 判断 example_prompt 是否为 PromptTemplate 类型
        if isinstance(example_prompt, PromptTemplate):
            # 如果是 PromptTemplate，直接赋值
            self.example_prompt = example_prompt
        else:
            # 如果是字符串，则先用 from_template 创建 PromptTemplate 再赋值
            self.example_prompt = PromptTemplate.from_template(example_prompt)
        # 保存前缀内容
        self.prefix = prefix
        # 保存后缀内容
        self.suffix = suffix
        # 保存示例分隔符
        self.example_separator = example_separator
        # 如果未指定输入变量，则自动根据前后缀推断变量名
        self.input_variables = input_variables or self._infer_input_variables()

    # 私有方法：推断前缀和后缀出现的模板变量名
    def _infer_input_variables(self) -> list[str]:
        # 新建一个集合用于保存变量名（去重）
        variables = set()
        # 提取 prefix 中引用的变量名
        variables.update(self._extract_vars(self.prefix))
        # 提取 suffix 中引用的变量名
        variables.update(self._extract_vars(self.suffix))
        # 转换为列表返回
        return list(variables)

    # 私有方法：提取文本中所有花括号包裹的模板变量名
    def _extract_vars(self, text: str) -> list[str]:
        # 如果输入为空字符串，直接返回空列表
        if not text:
            return []
        # 定义正则表达式，匹配 {变量名} 或 {变量名:格式}
        pattern = r"\{([^}:]+)(?::[^}]+)?\}"
        # 使用 re.findall 提取所有变量名
        matches = re.findall(pattern, text)
        # 去重并保持顺序返回变量名列表
        return list(dict.fromkeys(matches))

    # 格式化 few-shot 提示词，返回完整字符串
    def format(self, **kwargs) -> str:
        """
        根据传入的变量生成完整的 few-shot 提示词文本

        Args:
            **kwargs: 输入变量，可选，供示例选择
        """
        # 判断必需的变量是否全部传入，缺失时抛异常
        missing = set(self.input_variables) - set(kwargs.keys())
        if missing:
            raise ValueError(f"缺少必需的变量: {missing}")

        # 新建 parts 列表，用于拼接完整提示词的各部分内容
        parts: list[str] = []
        # 如果前缀不为空，格式化后加入 parts
        if self.prefix:
            parts.append(self._format_text(self.prefix, **kwargs))
        # 调用 format_examples 得到所有示例的字符串，并用分隔符拼接在一起
        example_block = self.example_separator.join(self.format_examples())
        # 如果 example_block 不为空字符串，加入 parts
        if example_block:
            parts.append(example_block)
        # 如果后缀不为空，格式化后加入 parts
        if self.suffix:
            parts.append(self._format_text(self.suffix, **kwargs))
        # 用示例分隔符连接所有组成部分，过滤空字符串
        return self.example_separator.join(part for part in parts if part)

    # 格式化所有示例，返回字符串列表
    def format_examples(self, input_variables: dict = None) -> list[str]:
        """
        返回格式化后的示例字符串列表

        Args:
            input_variables: 输入变量字典（预留，将来可以用于定制每个示例的选择）
        """
        # 新建存放格式化后示例的列表
        formatted = []
        # 遍历 every example 字典
        for example in self.examples:
            # 用 example_prompt 对当前示例格式化
            formatted.append(self.example_prompt.format(**example))
        # 返回格式化后的所有示例字符串列表
        return formatted

    # 私有方法：用 PromptTemplate 对 text 进行格式化
    def _format_text(self, text: str, **kwargs) -> str:
        # 先创建 PromptTemplate 实例
        temp_prompt = PromptTemplate.from_template(text)
        # 用传入参数格式化
        return temp_prompt.format(**kwargs)
+
# 定义一个从文件加载提示词模板的函数
def load_prompt(path: str | Path, encoding: str | None = None) -> PromptTemplate:
    # 将传入的路径参数转换为 Path 对象，方便后续进行文件操作
    file_path = Path(path)
    # 判断文件是否存在，如果不存在则抛出 FileNotFoundError 异常
    if not file_path.exists():
        raise FileNotFoundError(f"提示词文件不存在: {path}")
    # 判断文件扩展名是否为 .json，如果不是则抛出 ValueError 异常
    if file_path.suffix != ".json":
        raise ValueError(f"只支持 .json 格式文件，当前文件: {file_path.suffix}")
    # 打开文件，使用指定编码（一般为 utf-8），并读取 JSON 配置信息到 config 变量
    with file_path.open(encoding=encoding) as f:
        config = json.load(f)
    # 从配置字典中获取 "_type" 字段，不存在则默认值为 "prompt"
    config_type = config.get("_type", "prompt")
    # 校验 _type 字段是否为 "prompt"，如果不是则抛出异常
    if config_type != "prompt":
        raise ValueError(f"不支持的提示词类型: {config_type}，当前只支持 'prompt'")
    # 从配置中获取模板字符串 template，如果不存在则抛出异常
    template = config.get("template")
    if template is None:
        raise ValueError("配置文件中缺少 'template' 字段")
    # 使用读取到的模板字符串创建 PromptTemplate 实例并返回
    return PromptTemplate.from_template(template)

11.组合Prompt #

可以将多个 PromptTemplate 模板片段进行组合，构建出结构化、可复用、可扩展的复杂提示词。这种做法的关键优势在于：

1. 复用性：将提示词拆分为多个片段，每个片段实现不同的功能，比如开头问候、主要问题、结尾语等，可在不同场景灵活复用。
2. 灵活性：各片段的插值变量可以通过partial方法提前给出部分默认值，简化最终组合时需要填充的变量数量。
3. 易维护与扩展：当需要调整某个环节（比如问题格式、问候内容），只需要修改对应的片段即可，而无需大面积修改整体提示词。

原理剖析：

• 每个片段（如hello、question、bye）本质上都是一个PromptTemplate对象。我们可以先分别渲染这些片段，将段落内容填入字典input_dict。
• 渲染后的中间结果（如hello实际渲染出的字符串）也可以作为后续片段的输入变量（如最终组合模板使用{hello}、{question}、{bye}）。
• 最终用一个“总模板”将所有片段拼接组合，即final_prompt = PromptTemplate.from_template("{hello}{question}{bye}")，并用前面获得的渲染结果再次填充，得到完整的提示词。

应用场景举例：

• 多轮对话场景下，结构化合成上下文信息、工具调用模板等；
• 同一模板的不同内容片段需要切换或定制时，仅调整某个片段即可；
• 多人协作拆分开发提示词系统模块。

使用要点：

• 变量命名需明确，避免不同片段间变量冲突；
• 渲染顺序要保证依赖关系（如先有内容变量，再填片段）；
• 可通过提前partial部分变量，使后续格式化更加方便，减轻调用者负担。

通过这种组合式的提示词模板方法，我们可以避免“巨型模板”的维护难题，实现大型、灵活的自然语言处理流程，尤其适用于复杂知识型或交互型智能体、链式推理等 LLM 场景。

11.1. 11.CombinePromptTemplate.py #

11.CombinePromptTemplate.py

# 导入智能链的 ChatOpenAI 类，用于调用大语言模型
from smartchain.chat_models import ChatOpenAI
# 导入智能链的 PromptTemplate 类，用于构建提示词模板
from smartchain.prompts import PromptTemplate

# 定义 prompts 字典，将提示词拆分为多个可以复用的模板组件
prompts = {
    # 定义 hello 片段，内容为"hello\n"，无需变量
    "hello": PromptTemplate.from_template('hello\n'),
    # 定义 question 片段，模板带有 {adj} 和 {content} 变量，默认预填 adj='冷'
    "question": PromptTemplate.from_template('给我讲个{adj}的关于{content}的笑话\n').partial(adj='冷'),
    # 定义 bye 片段，内容为"bye"，无需变量
    "bye": PromptTemplate.from_template('bye')
}

# 定义输入字典，初始化内容变量
input_dict = {
    "content": "秦始皇",
}

# 遍历 prompts 字典中的每个模板片段，依次进行渲染
for key, prompt in prompts.items():
    # 调用每个模板片段的 format 方法进行渲染，并将结果存入 input_dict
    input_dict[key] = prompt.format(**input_dict)
    # 每渲染完一个片段，input_dict 会新增一个键值对，供后续模板引用

# 打印各个模板组件渲染后的结果
print("各个组件的渲染结果：")
for key, value in input_dict.items():
    # 依次输出每个组件的变量名和值
    print(f"{key}: {repr(value)}")

# 创建最终的提示词模板，将所有片段拼接组合
final_prompt = PromptTemplate.from_template("{hello}{question}{bye}")
# 用 input_dict 对最终模板进行格式化，获取完整提示词
formatted_result = final_prompt.format(**input_dict)
# 打印组合后的完整提示词
print("\n组合后的完整提示词：")
print(formatted_result)

# 创建 ChatOpenAI 实例，指定模型名称
chat = ChatOpenAI(model="gpt-4o")

# 调用模型，将格式化后的提示词传递给大模型获取回复
result = chat.invoke(formatted_result)
# 打印大模型的回复内容
print("\n模型回复：")
print(result.content)