系列导航:[第一篇:LangChain 1.0 是什么] → [第二篇:ChatModel 与 LLM 的使用] → 第三篇(当前) → 第四篇:LCEL 核心设计思想 前置要求:已掌握第二篇内容,能用 ChatOpenAI 调通模型 本篇目标:彻底解决提示词的工程化管理问题。从"散落在代码各处的 f-string",升级到"可复用、可组合、可测试的提示词模块"。内容覆盖 PromptTemplate 的所有核心用法、消息体系、Few-shot 构建、多轮对话上下文注入,以及提示词的组织与管理最佳实践。
系列导航:[第一篇:LangChain 1.0 是什么] → [第二篇:ChatModel 与 LLM 的使用] → 第三篇(当前) → 第四篇:LCEL 核心设计思想
前置要求:已掌握第二篇内容,能用 ChatOpenAI 调通模型
ChatOpenAI
本篇目标:彻底解决提示词的工程化管理问题。从"散落在代码各处的 f-string",升级到"可复用、可组合、可测试的提示词模块"。内容覆盖 PromptTemplate 的所有核心用法、消息体系、Few-shot 构建、多轮对话上下文注入,以及提示词的组织与管理最佳实践。
先来看一个真实的开发场景。你正在写一个代码审查工具,随着功能迭代,代码变成了这样:
# 某个模块里 def review_python_code(code: str) -> str: prompt = f"你是一个 Python 专家,请审查以下代码,找出 Bug 和性能问题:\n\n{code}" return llm.invoke(prompt) # 另一个模块里 def review_javascript_code(code: str, framework: str) -> str: prompt = f"你是一个 {framework} 专家。代码如下:\n{code}\n请指出问题。" return llm.invoke(prompt) # 又一个模块里 def review_sql_query(sql: str, dialect: str = "MySQL") -> str: system = "你是数据库专家" user = f"请审查这个 {dialect} 查询的性能:{sql}" messages = [{"role": "system", "content": system}, {"role": "user", "content": user}] return llm.invoke(messages) # 测试文件里(复制粘贴了一份,略有不同) def test_review(): prompt = f"作为 Python 专家,审查代码:\n{test_code}" ...
这段代码有几个典型的"提示词工程债务":
1. 提示词散落、难以统一维护:同样是"代码审查"的 System Prompt,各个函数里各写一套,改一处漏十处。
2. 格式不一致:有的用 f-string 拼接,有的用字典,有的用消息对象,代码风格混乱。
3. 无法复用和组合:review_python_code 和 review_javascript_code 的结构几乎一样,却无法复用任何提示词逻辑。
review_python_code
review_javascript_code
4. 测试困难:提示词和业务逻辑耦合在一起,无法单独测试提示词的效果。
5. 版本管理缺失:提示词改了什么、为什么改、哪个版本效果更好,完全没有记录。
LangChain 的 PromptTemplate 体系就是专门解决这些问题的。
PromptTemplate
LangChain 提供了一组专门处理提示词的类,先了解它们的关系:
BasePromptTemplate(基类) ├── PromptTemplate # 简单字符串模板(用于 LLM 接口,现在很少用) ├── ChatPromptTemplate # 对话消息列表模板(主力,99% 场景用这个) │ └── from_messages() # 最常用的构建方法 ├── FewShotPromptTemplate # Few-shot 示例模板(字符串版) └── FewShotChatMessagePromptTemplate # Few-shot 示例模板(对话版) 消息占位符 └── MessagesPlaceholder # 在消息列表中插入动态消息块(多轮对话关键)
本篇以 ChatPromptTemplate 为核心展开,这是你日常开发中最重要的类。
ChatPromptTemplate
from langchain_core.prompts import ChatPromptTemplate # 方式一:from_messages()(推荐,最直观) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的 {role},擅长 {specialty}。"), ("human", "{question}"), ]) # 格式化:填入变量,得到消息列表 messages = prompt.invoke({ "role": "后端工程师", "specialty": "Python 和分布式系统", "question": "如何设计一个高并发的任务队列?" }) print(type(messages)) # <class 'langchain_core.prompt_values.ChatPromptValue'> print(messages.to_messages()) # [ # SystemMessage(content='你是一个专业的后端工程师,擅长 Python 和分布式系统。'), # HumanMessage(content='如何设计一个高并发的任务队列?') # ]
from_messages() 接收一个列表,每个元素是一个 (role, template_string) 的元组。role 支持三种值:
from_messages()
(role, template_string)
"system"
SystemMessage
SystemMessagePromptTemplate.from_template(...)
"human"
"user"
HumanMessage
HumanMessagePromptTemplate.from_template(...)
"ai"
"assistant"
AIMessage
AIMessagePromptTemplate.from_template(...)
当某些消息内容是固定的、不需要变量替换时,可以直接传入消息对象:
from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import SystemMessage, HumanMessage prompt = ChatPromptTemplate.from_messages([ # 固定的系统消息(直接传对象) SystemMessage(content="你是一个严格的代码审查专家,只输出问题列表,不做解释。"), # 动态的用户消息(用元组,支持变量替换) ("human", "编程语言:{language}\n\n代码:\n```\n{code}\n```"), ]) messages = prompt.invoke({"language": "Python", "code": "def f(x): return x/0"})
在组合复杂提示词时,经常需要确认一个模板需要哪些变量:
prompt = ChatPromptTemplate.from_messages([ ("system", "你是 {role} 专家,工作在 {company}。"), ("human", "请回答:{question}"), ]) print(prompt.input_variables) # ['role', 'company', 'question'] # 也可以用 input_schema 查看完整 Schema print(prompt.input_schema.schema()) # {'title': 'PromptInput', 'type': 'object', 'properties': {'role': ..., 'company': ..., 'question': ...}}
当某些变量是"半固定"的(比如角色信息在整个应用里相同,只有问题不同),可以用 partial 预填充:
partial
from langchain_core.prompts import ChatPromptTemplate # 一个通用的专家问答模板 expert_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一名专业的 {domain} 专家,有 {years} 年从业经验。"), ("human", "{question}"), ]) # 为不同领域创建"半具体化"的模板 python_expert_prompt = expert_prompt.partial(domain="Python 编程", years="10") db_expert_prompt = expert_prompt.partial(domain="数据库优化", years="8") # 使用时只需传入剩余变量 messages1 = python_expert_prompt.invoke({"question": "什么时候用生成器?"}) messages2 = db_expert_prompt.invoke({"question": "如何优化慢查询?"}) print(python_expert_prompt.input_variables) # ['question'](domain 和 years 已预填充)
这个特性在工厂函数模式下非常有用:用一个通用模板衍生出多个具体模板,避免重复定义。
System Prompt 是控制模型行为最关键的部分。一个结构良好的 System Prompt 通常包含:
from langchain_core.prompts import ChatPromptTemplate # 结构化 System Prompt 的最佳实践 system_template = """你是一个专业的技术文档翻译助手。 ## 角色定义 你专门负责将英文技术文档翻译成地道的中文,目标读者是中国的软件工程师。 ## 翻译原则 - 技术术语保留英文原文,首次出现时在括号内给出中文释义 - 代码块、变量名、函数名不翻译 - 语言风格:专业、简洁,符合技术文档习惯 - 遇到无法确定的内容,在译文后用【注:...】标注 ## 输出格式 直接输出翻译结果,不要添加任何前缀说明(如"以下是翻译:")。 当前任务语言方向:{source_lang} → {target_lang} """ prompt = ChatPromptTemplate.from_messages([ ("system", system_template), ("human", "请翻译以下内容:\n\n{content}"), ])
System Prompt 的几个最佳实践:
##
在某些场景(比如 Few-shot 学习、引导对话方向),你需要在提示词里硬编码一段"对话历史":
from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import AIMessage, HumanMessage # 通过预置对话历史来引导模型的行为风格 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个编程助手,回答要简洁。"), # 硬编码的示例对话(引导模型的回答风格) HumanMessage(content="什么是闭包?"), AIMessage(content="闭包是一个函数,它记住了自己被定义时所在作用域的变量,即使该作用域已经执行完毕。"), HumanMessage(content="什么是装饰器?"), AIMessage(content="装饰器是一个函数,它接收另一个函数作为参数,并返回一个新函数,用于在不修改原函数的情况下增加功能。"), # 实际的用户问题 ("human", "{question}"), ]) # 此时模型已经"看到"了两个问答示例,会倾向于用同样简洁的风格回答 messages = prompt.invoke({"question": "什么是生成器?"})
MessagesPlaceholder 是处理多轮对话历史的核心工具。它在提示词模板中占一个"槽位",调用时可以把任意长度的消息列表插入这个位置:
MessagesPlaceholder
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder # 带历史记录的对话模板 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有帮助的助手。"), MessagesPlaceholder(variable_name="chat_history"), # 历史消息插入点 ("human", "{input}"), ]) # 模拟多轮对话 from langchain_core.messages import HumanMessage, AIMessage # 第一轮(没有历史) messages_round1 = prompt.invoke({ "chat_history": [], "input": "我叫张伟" }) # 第二轮(包含第一轮历史) messages_round2 = prompt.invoke({ "chat_history": [ HumanMessage(content="我叫张伟"), AIMessage(content="你好,张伟!很高兴认识你。有什么我可以帮助你的吗?"), ], "input": "你还记得我的名字吗?" }) # 打印第二轮的完整消息列表 for msg in messages_round2.to_messages(): print(f"[{type(msg).__name__}]: {msg.content}") # [SystemMessage]: 你是一个有帮助的助手。 # [HumanMessage]: 我叫张伟 # [AIMessage]: 你好,张伟!很高兴认识你。有什么我可以帮助你的吗? # [HumanMessage]: 你还记得我的名字吗?
MessagesPlaceholder 还支持一个很实用的参数 optional,当对应变量不存在时不报错:
optional
prompt = ChatPromptTemplate.from_messages([ ("system", "你是助手。"), MessagesPlaceholder(variable_name="chat_history", optional=True), # 可选 ("human", "{input}"), ]) # 不传 chat_history 也不会报错,该占位符直接被忽略 messages = prompt.invoke({"input": "你好"})
Few-shot(少样本)提示是提升模型输出质量最有效的技巧之一。通过提供几个输入/输出的示例,让模型学习你期望的输出格式和风格,效果往往远比在 System Prompt 里详细描述要好。
对于示例不多(2~5 个)的情况,最直接的方式是硬编码在消息列表里:
from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import HumanMessage, AIMessage # 情感分析 Few-shot 示例 prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个情感分析专家。 请分析用户评论的情感,严格按照以下 JSON 格式输出: {{"sentiment": "positive/negative/neutral", "score": 0.0-1.0, "reason": "简短说明"}}"""), # Few-shot 示例 HumanMessage(content="这个产品真的很棒,用了之后完全超出预期!"), AIMessage(content='{"sentiment": "positive", "score": 0.95, "reason": "使用了强烈正面词汇且表达了超预期满意"}'), HumanMessage(content="东西收到了,就这样吧。"), AIMessage(content='{"sentiment": "neutral", "score": 0.5, "reason": "语气平淡,既无明显正面也无负面评价"}'), HumanMessage(content="质量太差了,买了就后悔,差评!"), AIMessage(content='{"sentiment": "negative", "score": 0.05, "reason": "明确表达后悔购买,使用了强烈负面词汇"}'), # 实际输入 ("human", "{review}"), ]) from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser chain = prompt | ChatOpenAI(model="gpt-4o") | JsonOutputParser() result = chain.invoke({"review": "包装还不错,但是功能有点少,性价比一般。"}) print(result) # {'sentiment': 'neutral', 'score': 0.45, 'reason': '正面评价包装,但对功能和性价比持保留态度'}
当示例很多(10+),或者需要根据用户输入动态选择最相关的示例时,要用专门的 Few-shot 模板类:
from langchain_core.prompts import ( ChatPromptTemplate, FewShotChatMessagePromptTemplate, ) # 准备示例库 examples = [ { "input": "将列表中所有元素乘以 2", "output": "```python\nresult = [x * 2 for x in lst]\n```\n**说明**:使用列表推导式,简洁高效。" }, { "input": "统计字符串中每个字符出现的次数", "output": "```python\nfrom collections import Counter\ncount = Counter(s)\n```\n**说明**:`Counter` 是专为计数设计的数据结构,比手动遍历更清晰。" }, { "input": "读取 JSON 文件并解析", "output": "```python\nimport json\nwith open('file.json', 'r', encoding='utf-8') as f:\n data = json.load(f)\n```\n**说明**:使用 `with` 语句确保文件正确关闭,指定编码避免乱码。" }, { "input": "过滤列表中的偶数", "output": "```python\nevens = [x for x in lst if x % 2 == 0]\n# 或者使用 filter\nevens = list(filter(lambda x: x % 2 == 0, lst))\n```\n**说明**:两种方式均可,列表推导式更 Pythonic。" }, ] # 定义单个示例的格式 example_prompt = ChatPromptTemplate.from_messages([ ("human", "{input}"), ("ai", "{output}"), ]) # 构建 Few-shot 模板 few_shot_prompt = FewShotChatMessagePromptTemplate( examples=examples, example_prompt=example_prompt, ) # 组合成完整的对话模板 final_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个 Python 编程助手,每次回答都提供代码示例和简短说明。"), few_shot_prompt, # 自动展开所有示例 ("human", "{input}"), ]) # 查看格式化后的消息数量 messages = final_prompt.invoke({"input": "如何对字典按值排序?"}) print(f"消息总数:{len(messages.to_messages())}") # 消息总数:1(system) + 4*2(示例) + 1(用户) = 10
示例库很大时,把所有示例都塞进 Prompt 会浪费大量 Token。更好的方式是根据用户输入,语义检索最相关的几个示例:
from langchain_core.prompts import ( ChatPromptTemplate, FewShotChatMessagePromptTemplate, ) from langchain_core.example_selectors import SemanticSimilarityExampleSelector from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS # 准备大量示例(实际场景可能有几十上百个) examples = [ {"input": "列表去重", "output": "使用 `list(set(lst))` 或保留顺序用 `dict.fromkeys(lst)`"}, {"input": "字典合并", "output": "Python 3.9+ 用 `d1 | d2`,低版本用 `{**d1, **d2}`"}, {"input": "列表排序", "output": "原地排序 `lst.sort()`,生成新列表 `sorted(lst)`"}, {"input": "字符串分割", "output": "按分隔符 `s.split(',')` 或按行 `s.splitlines()`"}, {"input": "异常处理", "output": "用 `try/except/else/finally`,只捕获具体异常类型"}, {"input": "文件写入", "output": "用 `with open('f', 'w') as f: f.write(content)` 确保关闭"}, {"input": "正则匹配", "output": "用 `re.search()` 搜索,`re.findall()` 找所有匹配"}, # ... 更多示例 ] # 创建语义相似度示例选择器 selector = SemanticSimilarityExampleSelector.from_examples( examples=examples, embeddings=OpenAIEmbeddings(), vectorstore_cls=FAISS, k=3, # 每次选取最相关的 3 个示例 ) # 构建动态 Few-shot 模板 dynamic_few_shot_prompt = FewShotChatMessagePromptTemplate( example_selector=selector, # 使用动态选择器代替静态示例列表 example_prompt=ChatPromptTemplate.from_messages([ ("human", "{input}"), ("ai", "{output}"), ]), ) final_prompt = ChatPromptTemplate.from_messages([ ("system", "你是 Python 编程专家,给出简洁的代码建议。"), dynamic_few_shot_prompt, ("human", "{input}"), ]) # 输入"列表操作"类问题,会自动选出"列表去重"、"列表排序"等相关示例 messages = final_prompt.invoke({"input": "如何反转列表?"}) # 系统会自动选择最相关的 3 个示例插入 Prompt
虽然本篇重点是 ChatPromptTemplate,但 PromptTemplate(纯文本模板)在某些特定场景下依然有用,值得了解。
PromptTemplate 输出的是一个格式化后的纯字符串,而非消息列表。它主要用在两类场景:
场景一:作为链中的文本处理步骤(不直接传给模型,而是先加工文本再传)
from langchain_core.prompts import PromptTemplate # 用于格式化搜索查询 search_query_template = PromptTemplate.from_template( "请将以下用户问题改写为适合搜索引擎的关键词查询(只输出关键词,不超过 10 个词):\n\n用户问题:{user_question}" ) # 在 LCEL 链中作为预处理步骤 from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") # 先生成搜索查询,再用查询结果回答 query_rewrite_chain = search_query_template | llm | StrOutputParser() query = query_rewrite_chain.invoke({"user_question": "怎么让 Python 程序跑得更快?"}) print(query) # 输出类似:Python 性能优化 多进程 多线程 异步 Cython NumPy
场景二:作为 FewShotPromptTemplate 的 example_prompt
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate # 定义单个示例的展示格式 example_prompt = PromptTemplate.from_template( "输入:{input}\n输出:{output}" ) examples = [ {"input": "苹果", "output": "水果"}, {"input": "狗", "output": "动物"}, {"input": "Python", "output": "编程语言"}, ] # 构建 Few-shot 纯文本模板(用于非 Chat 场景) few_shot_prompt = FewShotPromptTemplate( examples=examples, example_prompt=example_prompt, prefix="将以下词语分类:\n", suffix="\n输入:{word}\n输出:", input_variables=["word"], ) print(few_shot_prompt.format(word="玫瑰")) # 将以下词语分类: # # 输入:苹果 # 输出:水果 # # 输入:狗 # 输出:动物 # # 输入:Python # 输出:编程语言 # # 输入:玫瑰 # 输出:
StringPromptValue
ChatPromptValue
/completions
/chat/completions
记住一条原则:只要你用的是现代 Chat 模型(GPT-4、DeepSeek、Claude 等),永远用 ChatPromptTemplate。
LangChain 默认使用 {variable} 作为变量占位符,但对于复杂的条件逻辑,可以切换到 Jinja2 模板引擎:
{variable}
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate # 使用 Jinja2 语法(template_format="jinja2") system_template = PromptTemplate.from_template( template="""你是一个代码审查专家。 审查要求: {% if strict_mode %} - 严格模式:任何潜在问题都必须指出,包括代码风格 - 必须给出修复建议 {% else %} - 普通模式:只指出明显的 Bug 和安全问题 {% endif %} 编程语言:{{ language }} {% if context %} 项目背景:{{ context }} {% endif %} """, template_format="jinja2", input_variables=["language", "strict_mode", "context"] ) # Jinja2 支持条件判断、循环等逻辑 prompt = ChatPromptTemplate.from_messages([ system_template, ("human", "代码:\n```{{ language }}\n{{ code }}\n```"), ])
何时使用 Jinja2: - 需要在提示词中做条件判断(根据参数决定是否包含某段内容) - 需要循环渲染列表(比如动态生成多条规则) - 提示词逻辑复杂,花括号默认语法不够用
注意事项: Jinja2 模板中变量用 {{ variable }}(双花括号),而非默认的 {variable}(单花括号)。
{{ variable }}
提示词模板本身是一个 Runnable,可以直接用 | 管道符与其他组件串联。理解这一点能帮助你写出更清晰、更灵活的代码。
Runnable
|
from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_messages([ ("system", "你是 {domain} 专家。"), ("human", "{question}"), ]) llm = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() # 标准三件套:prompt → llm → parser chain = prompt | llm | parser # 数据流向(每步的输入输出类型): # dict # ↓ ChatPromptTemplate # ChatPromptValue(格式化后的消息列表) # ↓ ChatOpenAI # AIMessage(模型回复) # ↓ StrOutputParser # str(纯文本) result = chain.invoke({"domain": "Python", "question": "什么是 GIL?"})
理解每一步的输入输出类型非常重要。很多初学者遇到的报错,根本原因都是"某个组件收到了它不认识的输入类型"。
当需要同时用不同视角分析同一内容时,用 RunnableParallel 并行执行多个链,总耗时约等于最慢的那个(而非串行的总和):
RunnableParallel
from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() # 三个维度的代码分析提示词 security_prompt = ChatPromptTemplate.from_messages([ ("system", "你是安全专家,只关注安全漏洞,输出漏洞列表。"), ("human", "分析代码安全性:\n```python\n{code}\n```"), ]) performance_prompt = ChatPromptTemplate.from_messages([ ("system", "你是性能专家,只关注性能瓶颈,输出优化建议。"), ("human", "分析代码性能:\n```python\n{code}\n```"), ]) readability_prompt = ChatPromptTemplate.from_messages([ ("system", "你是代码规范专家,只关注可读性,对照 PEP 8 给出改进意见。"), ("human", "分析代码规范性:\n```python\n{code}\n```"), ]) # 并行执行三个链 parallel_review = RunnableParallel( security=security_prompt | llm | parser, performance=performance_prompt | llm | parser, readability=readability_prompt | llm | parser, ) code = """ import pickle def process(data): obj = pickle.loads(data) result = [] for i in range(len(obj)): result.append(obj[i] * 2) return result """ results = parallel_review.invoke({"code": code}) for aspect, analysis in results.items(): print(f"\n=== {aspect.upper()} ===") print(analysis)
根据输入内容动态选择不同的提示词路径,实现智能分流:
from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() # 不同领域的专门提示词 code_prompt = ChatPromptTemplate.from_messages([ ("system", "你是编程专家,专注于代码相关问题,优先给出示例代码。"), ("human", "{question}"), ]) math_prompt = ChatPromptTemplate.from_messages([ ("system", "你是数学专家,用公式和推导过程回答,确保逻辑严谨。"), ("human", "{question}"), ]) general_prompt = ChatPromptTemplate.from_messages([ ("system", "你是通用助手,全面、准确地回答问题。"), ("human", "{question}"), ]) # 路由函数:根据问题内容返回对应的链 def select_chain(input_dict: dict): question = input_dict["question"].lower() if any(kw in question for kw in ["代码", "python", "函数", "bug", "算法", "编程"]): return code_prompt | llm | parser elif any(kw in question for kw in ["计算", "数学", "公式", "证明", "微积分"]): return math_prompt | llm | parser else: return general_prompt | llm | parser router_chain = RunnableLambda(select_chain) # 自动路由到最合适的提示词 r1 = router_chain.invoke({"question": "Python 里怎么实现单例模式?"}) # → code_prompt r2 = router_chain.invoke({"question": "什么是量子计算?"}) # → general_prompt
对于大型项目,可以把 System Prompt 拆成多个独立片段,按需组合:
from langchain_core.prompts import ChatPromptTemplate # 拆分成独立的 Prompt 片段 ROLE_TEMPLATE = "你是一个专业的 {role}。\n\n" CONSTRAINTS_TEMPLATE = """回答约束: - 语言:{response_language} - 长度:不超过 {max_words} 字 - 格式:{output_format} """ TASK_TEMPLATE = "当前任务:{task_description}\n\n" # 用字符串拼接组合片段(适合简单场景) combined_system = ROLE_TEMPLATE + CONSTRAINTS_TEMPLATE + TASK_TEMPLATE prompt = ChatPromptTemplate.from_messages([ ("system", combined_system), ("human", "{question}"), ]) # 格式化 messages = prompt.invoke({ "role": "技术写作专家", "response_language": "中文", "max_words": "300", "output_format": "Markdown,使用标题和列表", "task_description": "为技术博客撰写文章摘要", "question": "请为一篇关于 LangChain 的文章写一个摘要" })
当一个应用有多个相关的提示词时,用工厂函数统一管理:
from langchain_core.prompts import ChatPromptTemplate # 提示词工厂 class CodeAssistantPrompts: """代码助手的所有提示词集中管理""" BASE_SYSTEM = """你是一个专业的 {language} 代码助手。 代码风格遵循 {style_guide} 规范。 """ @classmethod def review_prompt(cls) -> ChatPromptTemplate: """代码审查提示词""" return ChatPromptTemplate.from_messages([ ("system", cls.BASE_SYSTEM + "请找出代码中的 Bug、安全问题和性能瓶颈。"), ("human", "请审查以下代码:\n\n```{language}\n{code}\n```"), ]) @classmethod def explain_prompt(cls) -> ChatPromptTemplate: """代码解释提示词""" return ChatPromptTemplate.from_messages([ ("system", cls.BASE_SYSTEM + "请用清晰易懂的语言解释代码逻辑,面向 {audience} 读者。"), ("human", "请解释这段代码:\n\n```{language}\n{code}\n```"), ]) @classmethod def refactor_prompt(cls) -> ChatPromptTemplate: """代码重构提示词""" return ChatPromptTemplate.from_messages([ ("system", cls.BASE_SYSTEM + "请在保持功能不变的前提下,重构代码以提升可读性和性能。"), ("human", "重构目标:{refactor_goal}\n\n原始代码:\n```{language}\n{code}\n```"), ]) @classmethod def test_prompt(cls) -> ChatPromptTemplate: """测试用例生成提示词""" return ChatPromptTemplate.from_messages([ ("system", cls.BASE_SYSTEM + "请为函数生成全面的单元测试,使用 {test_framework} 框架。"), ("human", "为以下函数生成测试用例:\n\n```{language}\n{code}\n```"), ]) # 使用 from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(model="gpt-4o") parser = StrOutputParser() # 审查链 review_chain = CodeAssistantPrompts.review_prompt() | llm | parser result = review_chain.invoke({ "language": "Python", "style_guide": "PEP 8", "code": "def calc(x,y): return x/y" }) # 解释链(复用同一个 BASE_SYSTEM) explain_chain = CodeAssistantPrompts.explain_prompt() | llm | parser result = explain_chain.invoke({ "language": "Python", "style_guide": "PEP 8", "audience": "初级开发者", "code": "lambda x: x if x > 0 else -x" })
在生产环境中,提示词的每一次修改都可能影响模型输出的质量。建议把提示词配置化,便于版本追踪:
# prompts/config.py # 把提示词文本集中存放,与业务代码分离 PROMPTS = { "v1.0": { "system": "你是一个助手,请回答用户的问题。", "description": "初始版本,简单通用" }, "v1.1": { "system": "你是一个专业的技术助手。回答要准确、简洁,不确定的内容请明确说明。", "description": "v1.0 基础上增加了准确性约束" }, "v2.0": { "system": """你是一个专业的技术助手,有以下行为准则: 1. 回答要准确,不确定的内容明确说"不确定" 2. 优先给出可运行的代码示例 3. 复杂概念用类比方式解释 4. 回答长度与问题复杂度匹配""", "description": "v2.0 大幅增强了指令的具体性" }, } ACTIVE_VERSION = "v2.0" # 在业务代码中使用配置 from langchain_core.prompts import ChatPromptTemplate from prompts.config import PROMPTS, ACTIVE_VERSION def create_qa_prompt(version: str = ACTIVE_VERSION) -> ChatPromptTemplate: config = PROMPTS[version] return ChatPromptTemplate.from_messages([ ("system", config["system"]), ("human", "{question}"), ])
把本篇所有知识点串联起来,构建一个支持多轮对话、角色可配置、输出格式规范的完整对话系统:
import os from typing import Optional from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage, BaseMessage from langchain_core.output_parsers import StrOutputParser load_dotenv() # ========== 提示词定义 ========== ASSISTANT_SYSTEM_TEMPLATE = """你是 {assistant_name},一个专业的 {domain} 助手。 ## 性格特点 {personality} ## 回答规范 - 语言:使用 {language} 回答 - 风格:{tone} - 如果问题超出你的专业范围({domain}),请礼貌说明并引导用户 - 不确定的内容请明确表示不确定,不要编造 ## 特别说明 {special_instructions} """ def create_chat_prompt( assistant_name: str = "小助手", domain: str = "通用知识", personality: str = "友善、耐心、专业", language: str = "中文", tone: str = "专业而不失亲切", special_instructions: str = "无", ) -> ChatPromptTemplate: """创建可配置的多轮对话提示词""" return ChatPromptTemplate.from_messages([ ("system", ASSISTANT_SYSTEM_TEMPLATE), MessagesPlaceholder(variable_name="chat_history", optional=True), ("human", "{input}"), ]).partial( assistant_name=assistant_name, domain=domain, personality=personality, language=language, tone=tone, special_instructions=special_instructions, ) # ========== 对话管理器 ========== class ConversationManager: """管理多轮对话的上下文历史""" def __init__( self, prompt: ChatPromptTemplate, llm: ChatOpenAI, max_history_turns: int = 10, ): self.chain = prompt | llm | StrOutputParser() self.history: list[BaseMessage] = [] self.max_history_turns = max_history_turns def _trim_history(self): """保留最近 N 轮对话,避免上下文过长""" max_messages = self.max_history_turns * 2 # 每轮 = 1 HumanMessage + 1 AIMessage if len(self.history) > max_messages: self.history = self.history[-max_messages:] def chat(self, user_input: str, stream: bool = False) -> str: """发送消息并获取回复""" self._trim_history() if stream: # 流式输出 full_response = "" for chunk in self.chain.stream({ "chat_history": self.history, "input": user_input, }): print(chunk, end="", flush=True) full_response += chunk print() # 换行 else: full_response = self.chain.invoke({ "chat_history": self.history, "input": user_input, }) # 更新历史 self.history.append(HumanMessage(content=user_input)) self.history.append(AIMessage(content=full_response)) return full_response def clear_history(self): """清空对话历史""" self.history = [] print("对话历史已清空") def show_history(self): """展示当前对话历史""" print(f"\n=== 对话历史(共 {len(self.history) // 2} 轮)===") for i, msg in enumerate(self.history): role = "用户" if isinstance(msg, HumanMessage) else "助手" print(f"[{role}]: {msg.content[:100]}{'...' if len(msg.content) > 100 else ''}") print() # ========== 使用示例 ========== def main(): llm = ChatOpenAI(model="gpt-4o", temperature=0.7) # 创建一个 Python 技术助手 python_prompt = create_chat_prompt( assistant_name="PyHelper", domain="Python 编程", personality="专业、精确,喜欢举例说明,对初学者有耐心", tone="技术性,但避免过于学术化", special_instructions="遇到代码问题,优先给出可运行的示例代码", ) bot = ConversationManager( prompt=python_prompt, llm=llm, max_history_turns=8, ) print("Python 助手已启动,输入 'quit' 退出,'clear' 清空历史,'history' 查看历史\n") while True: user_input = input("你:").strip() if not user_input: continue if user_input.lower() == "quit": break elif user_input.lower() == "clear": bot.clear_history() continue elif user_input.lower() == "history": bot.show_history() continue print("助手:", end="") bot.chat(user_input, stream=True) if __name__ == "__main__": main()
开发时经常需要确认提示词格式化后的实际内容,特别是变量替换是否正确:
from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages([ ("system", "你是 {role} 专家。"), ("human", "{question}"), ]) # 方式一:invoke 后查看 to_messages() messages = prompt.invoke({"role": "Python", "question": "什么是 GIL?"}) for msg in messages.to_messages(): print(f"[{msg.__class__.__name__}]") print(msg.content) print("-" * 40) # 方式二:format_messages() 直接格式化 messages_list = prompt.format_messages(role="Python", question="什么是 GIL?")
提示词中如果需要输出字面量的花括号(比如 JSON 示例),需要用双花括号转义:
from langchain_core.prompts import ChatPromptTemplate # ❌ 错误:{{"key": "value"}} 中的 key 和 value 会被当成变量 prompt = ChatPromptTemplate.from_messages([ ("system", "请输出 JSON 格式:{\"key\": \"value\"}"), # 这样也不行 ]) # ✅ 正确:用双花括号表示字面量花括号 prompt = ChatPromptTemplate.from_messages([ ("system", """请严格按以下 JSON 格式输出(注意格式): {{"name": "示例", "value": 123}} 用户输入的内容是:{user_input}"""), ]) # 格式化后,{{}} 会变成 {},{user_input} 会被替换 messages = prompt.invoke({"user_input": "测试数据"}) print(messages.to_messages()[0].content) # 请严格按以下 JSON 格式输出(注意格式): # {"name": "示例", "value": 123} # 用户输入的内容是:测试数据
当历史消息过多,接近模型上下文窗口限制时,需要裁剪:
from langchain_core.messages import trim_messages from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") # trim_messages 按 Token 数裁剪历史消息 trimmer = trim_messages( max_tokens=4000, # 历史消息最多占用 4000 Token strategy="last", # 保留最近的消息("first" 则保留最早的) token_counter=llm, # 用模型的 tokenizer 计算 Token 数 include_system=True, # 保留 SystemMessage allow_partial=False, # 不允许截断单条消息 start_on="human", # 确保第一条消息是用户消息 ) # 在链中使用 from langchain_core.runnables import RunnableLambda def process_messages(input_dict): """裁剪历史消息后传入链""" trimmed = trimmer.invoke(input_dict["chat_history"]) return {**input_dict, "chat_history": trimmed} chain = RunnableLambda(process_messages) | prompt | llm
提示词应该像代码一样可测试,确保变量替换正确、消息顺序合理:
import pytest from langchain_core.messages import SystemMessage, HumanMessage from langchain_core.prompts import ChatPromptTemplate def test_code_review_prompt(): """测试代码审查提示词的格式化结果""" prompt = ChatPromptTemplate.from_messages([ ("system", "你是 {language} 专家。"), ("human", "审查代码:\n{code}"), ]) messages = prompt.invoke({ "language": "Python", "code": "def f(): pass" }).to_messages() # 断言消息数量 assert len(messages) == 2 # 断言消息类型 assert isinstance(messages[0], SystemMessage) assert isinstance(messages[1], HumanMessage) # 断言变量替换正确 assert "Python" in messages[0].content assert "def f(): pass" in messages[1].content # 断言没有未替换的变量占位符 assert "{language}" not in messages[0].content assert "{code}" not in messages[1].content
本篇从"为什么需要工程化"出发,系统介绍了 LangChain 提示词体系的全部核心内容:
System / Human / AI
optional=True
FewShotChatMessagePromptTemplate
你现在应该能做到: - 用 ChatPromptTemplate 替换项目中所有散落的 f-string 提示词 - 用 MessagesPlaceholder 正确实现多轮对话的上下文管理 - 用 Few-shot 示例显著提升特定任务的模型输出质量 - 用工厂函数模式管理一个大型项目的提示词族 - 对提示词进行单元测试,保证格式化逻辑的正确性
《LCEL——LangChain 的核心设计思想》
掌握了 Model 和 Prompt 两个核心组件之后,第四篇将重点讲解把它们连接起来的"管道"——LCEL(LangChain Expression Language)。你会学到:
RunnablePassthrough
RunnableLambda
代码仓库:本系列所有可运行代码示例统一维护在 GitHub,每篇对应独立目录,可直接克隆运行。 系列导航:[第一篇] → [第二篇] → 第三篇(当前) → 第四篇 → ...
代码仓库:本系列所有可运行代码示例统一维护在 GitHub,每篇对应独立目录,可直接克隆运行。
系列导航:[第一篇] → [第二篇] → 第三篇(当前) → 第四篇 → ...
还没有评论,来抢沙发吧!
博客管理员
40 篇文章
还没有评论,来抢沙发吧!