Prompt Engineering最佳实践与链式推理

引言

去年双十一大促前夜，我们的智能客服系统突然集体“降智”：同一个退款问题，上午还能给出准确的三步操作指引，晚上就开始胡编乱造，甚至建议用户“联系火星客服”。排查了半天，模型没换、温度没调、知识库没更新，唯一的变量是——我们在Prompt里加了一句“请尽量简洁地回答用户问题”。

这就是Prompt Engineering的残酷现实：它不是“写提示词”，而是在概率空间里做接口设计。你面对的不是一个确定性函数，而是一个拥有千亿参数的随机过程。一句措辞的改动，可能让准确率从92%掉到67%，也可能让Token成本翻三倍。

更麻烦的是，当业务逻辑从“单轮问答”升级到“多步骤推理”——比如先识别意图、再检索知识、再校验事实、最后生成回答——单条Prompt已经撑不住了。你需要的是链式推理（Chain-of-Thought / Chain of Reasoning），把一个大问题拆成多个可验证、可观测、可回滚的子步骤。

这篇文章面向有3-5年经验的工程师，不讲“什么是Prompt”，而是讲如何像设计分布式系统一样设计Prompt流水线。我会带你从源码级别理解LangChain/LangGraph的链式编排机制，给出三个可直接跑的生产级示例，并对比不同方案的工程取舍。

---

核心概念：从“点菜”到“后厨流水线”

生活类比：单条Prompt是“跟厨师喊一嗓子”，链式推理是“后厨SOP”

想象你去餐厅点菜。

单条Prompt就像你站在厨房门口喊：“来个鱼香肉丝，不要辣，多放糖，肉要嫩，菜要脆，顺便帮我算一下这顿饭多少钱。” 厨师可能听懂了，也可能只听懂了前半句。你无法知道他在哪个环节出了错。

链式推理则像后厨的标准作业流程（SOP）：

1. 接单员（意图识别）确认你要的是鱼香肉丝，备注“不辣、多糖”；
2. 配菜员（信息抽取）从你的历史订单里查出你对花生过敏；
3. 主厨（推理生成）根据配菜结果调整配方；
4. 质检员（事实校验）检查成品是否含花生；
5. 传菜员（格式化输出）把菜和账单一起端给你。

每个环节都有明确的输入输出契约，任何一步出错都能被定位和回滚。

技术定义

Prompt Engineering：通过设计输入文本的结构、示例、约束和格式，引导LLM在特定任务上产生期望输出的工程实践。核心变量包括：指令清晰度、Few-shot示例质量、输出格式约束、上下文窗口管理、温度/Top-p采样参数。

链式推理（Chain-of-Thought, CoT）：将复杂任务分解为多个中间步骤，每一步的输出作为下一步的输入，形成有向无环图（DAG）或状态机。典型模式包括：

• Sequential Chain：线性串联，A→B→C

• Router Chain：条件分支，根据输入选择不同子链

• Map-Reduce Chain：并行拆分再聚合

• Reflection Chain：生成→批判→修正的循环

LangGraph：LangChain团队推出的有状态图编排框架，把链式推理建模为状态机，支持循环、条件边、持久化和人工介入。它解决的核心问题是：当Chain不再是直线，而是带环的图时，如何管理状态和容错。

---

源码/原理深度分析：LangChain Expression Language（LCEL）与LangGraph的运行时

LCEL的Runnable协议：一切皆可管道

LangChain v0.1之后，核心抽象是Runnable。它的源码定义在langchain_core/runnables/base.py，简化后如下：

class Runnable(Generic[Input, Output], ABC):
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        return self._call_with_config(
            lambda inner_input: self.invoke(inner_input, config),
            input, config
        )

    def __or__(self, other: "Runnable") -> "RunnableSequence":
        return RunnableSequence(first=self, last=other)

    def batch(self, inputs: List[Input], config=None) -> List[Output]:
        # 默认串行，子类可覆盖为并行
        return [self.invoke(i, config) for i in inputs]

关键点在于__or__方法：它重载了Python的位或运算符，让prompt | model | parser这种写法等价于RunnableSequence(first=prompt, last=model).__or__(parser)。RunnableSequence的invoke实现是：

class RunnableSequence(RunnableSerializable[Input, Output]):
    def invoke(self, input: Input, config=None) -> Output:
        for i, step in enumerate(self.steps):
            input = step.invoke(input, config)
        return input

这就是LCEL的本质：一个轻量级的管道模式，每一步的Output必须是下一步的Input类型。它没有魔法，但通过类型系统和RunnableConfig实现了回调、重试、流式输出的统一注入。

工程启示：LCEL适合线性、无状态、可并行的场景。一旦你需要“根据中间结果决定下一步走哪个分支”或“失败后回到上一步重试”，LCEL的RunnableSequence就不够用了——因为它是单向的，没有状态记忆。

LangGraph的StateGraph：把Chain升级为状态机

LangGraph的核心是StateGraph，源码在langgraph/graph/state.py。它的设计借鉴了Apache Beam和Temporal的工作流思想。简化后的关键结构：

class StateGraph(Generic[StateT]):
    def __init__(self, state_schema: Type[StateT]):
        self.nodes: Dict[str, Runnable] = {}
        self.edges: Set[Tuple[str, str]] = set()
        self.branches: Dict[str, Dict[str, str]] = {}
        self.state_schema = state_schema

    def add_node(self, name: str, action: Runnable):
        self.nodes[name] = action

    def add_edge(self, start: str, end: str):
        self.edges.add((start, end))

    def add_conditional_edges(self, source: str, condition: Callable, mapping: Dict[str, str]):
        self.branches[source] = mapping

    def compile(self, checkpointer=None) -> CompiledGraph:
        return CompiledGraph(self, checkpointer)

CompiledGraph的invoke实现是一个事件循环：

def invoke(self, input: StateT, config=None) -> StateT:
    state = input
    current_node = self.entry_point
    while current_node != END:
        node_fn = self.nodes[current_node]
        # 关键：节点函数接收完整state，返回部分更新
        update = node_fn.invoke(state, config)
        state = {**state, **update}  # 状态归并
        # 决定下一个节点
        if current_node in self.branches:
            condition = self.branches[current_node]["condition"]
            next_node = condition(state)
        else:
            next_node = self._get_next_node(current_node)
        current_node = next_node
    return state

三个关键设计：

1. 状态归并（State Merge）：每个节点返回的是dict，通过{state, update}合并。这要求你的状态schema是“可增量更新”的。LangGraph默认用TypedDict，但支持Annotated类型来定义归并策略，比如Annotated[List, operator.add]表示列表追加而非覆盖。

1. 条件边（Conditional Edge）：condition是一个纯函数，接收state，返回下一个节点名。这让图可以表达if-else和循环。

1. Checkpointer：如果传入checkpointer（如SqliteSaver），每次状态变更都会持久化。这是实现“人工介入”和“断点续跑”的基础——本质上和Temporal的Workflow History是同一思路。

为什么这很重要？

对比一下：如果你用裸OpenAI API写链式推理，代码会长这样：

resp1 = openai.chat.completions.create(...)
intent = parse(resp1)
if intent == "refund":
    resp2 = openai.chat.completions.create(...)
    # 手动管理上下文、错误、重试

当链路变成10步、有3个分支、需要重试和人工审核时，这段代码会变成意大利面条。LangGraph的价值在于把控制流从代码里抽出来，变成可观测、可持久化的图。

graph TD A[用户输入] --> B[意图识别节点] B -->|退款| C[订单查询节点] B -->|咨询| D[知识库检索节点] B -->|投诉| E[人工审核节点] C --> F[退款政策校验] F -->|通过| G[生成退款方案] F -->|拒绝| H[生成解释话术] D --> I[答案生成] G --> J[事实校验节点] H --> J I --> J J -->|校验失败| B J -->|校验通过| K[格式化输出] E --> K K --> L[返回用户]

这张图里有两个关键点：J→B的回环（校验失败重新识别意图）和E的人工介入（投诉直接转人工）。这两点用LCEL都很难优雅实现，但LangGraph只需要add_conditional_edges加一个interrupt_before配置。

---

实战代码：三个可运行的生产级示例

示例1：带事实校验的RAG链（LCEL版本）

这个例子展示如何用LCEL构建一个“检索→生成→校验→修正”的线性链。适合中等复杂度、无需循环的场景。

# requirements: langchain-openai, langchain-core, chromadb, pydantic
import os
from typing import List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_community.vectorstores import Chroma

# ---------- 1. 定义结构化输出 ----------
class AnswerWithCitation(BaseModel):
    answer: str = Field(description="最终答案")
    citations: List[str] = Field(description="引用的文档ID列表")
    confidence: float = Field(description="置信度0-1")

class FactCheckResult(BaseModel):
    is_supported: bool = Field(description="答案是否被引用文档支持")
    reason: str = Field(description="判断理由")
    corrected_answer: str = Field(description="如果不受支持，给出修正答案；否则原样返回")

# ---------- 2. 初始化组件 ----------
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 模拟知识库
docs = [
    "退款政策：订单支付后7天内可无理由退款，超过7天需提供质量问题证明。",
    "退款到账时间：原路返回，信用卡3-5个工作日，支付宝实时到账。",
    "运费规则：无理由退款用户承担运费，质量问题商家承担。",
]
vectorstore = Chroma.from_texts(docs, embeddings, collection_name="refund_policy")
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# ---------- 3. 构建生成链 ----------
answer_parser = PydanticOutputParser(pydantic_object=AnswerWithCitation)
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是退款政策客服。基于以下文档回答问题，必须引用文档。\n{format_instructions}"),
    ("human", "文档：\n{context}\n\n问题：{question}")
]).partial(format_instructions=answer_parser.get_format_instructions())

def format_docs(docs):
    return "\n".join(f"[{i}] {d.page_content}" for i, d in enumerate(docs))

generate_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | answer_prompt
    | llm
    | answer_parser
)

# ---------- 4. 构建校验链 ----------
check_parser = PydanticOutputParser(pydantic_object=FactCheckResult)
check_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是事实校验员。判断答案是否被文档支持。\n{format_instructions}"),
    ("human", "文档：\n{context}\n\n答案：{answer}")
]).partial(format_instructions=check_parser.get_format_instructions())

# 关键：用RunnableLambda把上一步的输出和原始context拼起来
def prepare_check(inputs):
    answer = inputs["answer"]
    context = format_docs(retriever.invoke(inputs["question"]))
    return {"answer": answer.answer, "context": context}

check_chain = RunnableLambda(prepare_check) | check_prompt | llm | check_parser

# ---------- 5. 组装完整链 ----------
full_chain = (
    RunnablePassthrough.assign(answer=generate_chain)
    | RunnableLambda(lambda x: {"question": x["question"], "answer": x["answer"]})
    | RunnablePassthrough.assign(check=check_chain)
    | RunnableLambda(lambda x: {
        "final": x["check"].corrected_answer if not x["check"].is_supported else x["answer"].answer,
        "confidence": x["answer"].confidence,
        "was_corrected": not x["check"].is_supported,
    })
)

# ---------- 6. 运行 ----------
if __name__ == "__main__":
    result = full_chain.invoke({"question": "我昨天买的衣服想退款，运费谁出？"})
    print(result)
    # 预期输出：final提到"无理由退款用户承担运费"，was_corrected=False

设计要点：

• 用PydanticOutputParser强制结构化输出，避免解析正则的脆弱性。

• RunnablePassthrough.assign是LCEL里最实用的工具，它能在不破坏原输入的前提下“挂载”中间结果。

• 校验链是独立的Runnable，可以单独测试和复用。

局限：如果校验失败，这里只是返回修正答案，没有“重新生成”的循环。要支持循环，需要LangGraph。

---

示例2：LangGraph实现带循环和人工介入的客服链

这个例子展示LangGraph的完整能力：条件分支、循环重试、人工审核中断。

# requirements: langgraph, langchain-openai, langchain-core
from typing import TypedDict, Literal, Annotated
import operator
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# ---------- 1. 定义状态 ----------
class AgentState(TypedDict):
    user_input: str
    intent: str
    answer: str
    check_passed: bool
    retry_count: int
    history: Annotated[list, operator.add]  # 追加而非覆盖

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

# ---------- 2. 定义节点函数 ----------
def classify_intent(state: AgentState) -> dict:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "分类用户意图，只返回：refund / inquiry / complaint"),
        ("human", "{input}")
    ])
    result = (prompt | llm).invoke({"input": state["user_input"]})
    intent = result.content.strip().lower()
    return {"intent": intent, "history": [f"intent={intent}"]}

def handle_refund(state: AgentState) -> dict:
    # 实际项目中这里会查订单系统
    answer = "退款将在3-5个工作日原路返回。"
    return {"answer": answer, "history": ["handled_refund"]}

def handle_inquiry(state: AgentState) -> dict:
    answer = "请提供订单号，我帮您查询。"
    return {"answer": answer, "history": ["handled_inquiry"]}

def handle_complaint(state: AgentState) -> dict:
    # 投诉直接转人工，不生成答案
    return {"answer": "已为您转接人工客服。", "history": ["escalated"]}

def fact_check(state: AgentState) -> dict:
    # 模拟校验：如果答案包含"保证"字样则判定不通过（避免过度承诺）
    passed = "保证" not in state["answer"]
    return {
        "check_passed": passed,
        "retry_count": state.get("retry_count", 0) + 1,
        "history": [f"check={'pass' if passed else 'fail'}"]
    }

def regenerate(state: AgentState) -> dict:
    # 修正答案：去掉过度承诺
    corrected = state["answer"].replace("保证", "通常")
    return {"answer": corrected, "history": ["regenerated"]}

# ---------- 3. 路由函数 ----------
def route_by_intent(state: AgentState) -> Literal["handle_refund", "handle_inquiry", "handle_complaint"]:
    return {
        "refund": "handle_refund",
        "inquiry": "handle_inquiry",
        "complaint": "handle_complaint",
    }.get(state["intent"], "handle_inquiry")

def route_after_check(state: AgentState) -> Literal["regenerate", "end"]:
    if state["check_passed"]:
        return "end"
    if state["retry_count"] >= 2:  # 最多重试2次
        return "end"
    return "regenerate"

# ---------- 4. 构建图 ----------
graph = StateGraph(AgentState)
graph.add_node("classify", classify_intent)
graph.add_node("handle_refund", handle_refund)
graph.add_node("handle_inquiry", handle_inquiry)
graph.add_node("handle_complaint", handle_complaint)
graph.add_node("fact_check", fact_check)
graph.add_node("regenerate", regenerate)

graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_by_intent)
graph.add_edge("handle_refund", "fact_check")
graph.add_edge("handle_inquiry", "fact_check")
graph.add_edge("handle_complaint", END)  # 投诉不校验
graph.add_conditional_edges("fact_check", route_after_check, {"regenerate": "regenerate", "end": END})
graph.add_edge("regenerate", "fact_check")  # 循环回校验

# ---------- 5. 编译并运行 ----------
memory = MemorySaver()
app = graph.compile(checkpointer=memory, interrupt_before=["handle_complaint"])

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "user-123"}}
    result = app.invoke({"user_input": "我要投诉！你们保证明天到货的！", "retry_count": 0, "history": []}, config)
    print("最终答案:", result.get("answer"))
    print("执行历史:", result["history"])
    
    # 人工介入示例：投诉节点被中断，恢复后继续
    # app.invoke(None, config)  # 人工审核后继续

关键设计：

• Annotated[list, operator.add]让history字段自动追加，这是LangGraph状态归并的核心机制。

• interrupt_before=["handle_complaint"]让投诉场景暂停，等待人工审核。这是生产环境合规要求的常见模式。

• route_after_check里的retry_count防止无限循环——这是链式推理最容易踩的坑。

• MemorySaver把状态存在内存里，生产环境换SqliteSaver或PostgresSaver。

---

示例3：带自一致性投票的推理链（Self-Consistency）

这个例子展示如何用并行采样提升推理可靠性，适合数学、逻辑等需要高准确率的场景。

# requirements: langchain-openai, langchain-core
import asyncio
from collections import Counter
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)  # 高温度用于采样多样性

# ---------- 1. CoT Prompt ----------
cot_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是数学推理专家。请一步步思考，最后一行用"答案：X"格式输出。
示例：
问题：小明有5个苹果，吃了2个，又买了3个，现在有几个？
思考：5-2=3，3+3=6。
答案：6"""),
    ("human", "{question}")
])

# ---------- 2. 并行采样N次 ----------
async def sample_once(question: str) -> str:
    chain = cot_prompt | llm
    result = await chain.ainvoke({"question": question})
    # 提取"答案："后的内容
    for line in reversed(result.content.strip().split("\n")):
        if "答案：" in line:
            return line.split("答案：")[-1].strip()
    return result.content.strip()

async def self_consistency(question: str, n: int = 5) -> dict:
    tasks = [sample_once(question) for _ in range(n)]
    answers = await asyncio.gather(*tasks)
    
    # 投票
    counter = Counter(answers)
    best_answer, votes = counter.most_common(1)[0]
    return {
        "answer": best_answer,
        "confidence": votes / n,
        "all_samples": answers,
    }

# ---------- 3. 运行 ----------
if __name__ == "__main__":
    question = "一个水池有甲乙两个进水管。甲管单独注满需6小时，乙管单独注满需4小时。两管同时开，多久注满？"
    result = asyncio.run(self_consistency(question, n=5))
    print(f"答案: {result['answer']}")
    print(f"置信度: {result['confidence']:.0%}")
    print(f"所有采样: {result['all_samples']}")
    # 预期：多数采样得到"2.4小时"或"12/5小时"

为什么有效：LLM在高温度下采样会产生不同的推理路径。如果多条路径收敛到同一答案，说明该答案更可能是“真实”的。这类似于集成学习中的Bagging思想。

工程取舍：

• 成本：N次采样=N倍Token成本。生产环境通常N=3~5。

• 延迟：并行采样不增加延迟（受限于API并发限制），但串行会。

• 适用场景：数学、逻辑、代码生成等有明确正确答案的任务。不适用于创意写作、开放问答。

---

方案对比：LCEL vs LangGraph vs 裸API vs DSPy

维度	裸OpenAI API	LCEL	LangGraph	DSPy
学习曲线	低	中	高	高
控制流	手写if-else	线性管道	状态机/图	编译优化
状态管理	手动	无状态	内置+持久化	无
循环/分支	手写	不支持	原生支持	通过编译
可观测性	需自建	LangSmith集成	LangSmith集成	有限
人工介入	手写	不支持	interrupt原生	不支持
适合场景	简单单轮	线性RAG	复杂Agent	研究/自动优化
生产成熟度	高	中	中高	低

DSPy值得单独提一句：它把Prompt Engineering变成了“编译问题”——你定义输入输出和评估指标，DSPy自动搜索最优Prompt。但它的黑盒性和调试难度让很多团队望而却步。目前更适合研究场景，生产环境仍以显式编排为主。

选型建议：

• 单轮问答、简单RAG → 裸API或LCEL，别过度设计。

• 多步骤、有分支、需要重试 → LangGraph。

• 需要人工审核、合规审计 → LangGraph + Checkpointer。

• Prompt优化是瓶颈且有人力标注 → 试点DSPy。

---

最佳实践与避坑指南

1. 结构化输出优先于正则解析

坑：用正则从LLM输出里抠JSON，遇到模型加个前缀就崩。

解：用PydanticOutputParser或OpenAI的response_format={"type": "json_schema"}。后者在API层面保证输出合法JSON，比Prompt约束可靠得多。

2. 温度不是“创意旋钮”，是“方差旋钮”

• temperature=0：适合分类、抽取、校验。但注意，0不保证确定性，因为GPU浮点运算和batch size仍会引入微小差异。

• temperature=0.7+：适合头脑风暴、自一致性采样。

• 生产环境的校验节点一律用0。

3. 上下文窗口是稀缺资源，不是垃圾桶

坑：把整个知识库塞进Prompt，Token成本爆炸且模型注意力涣散。

解：

• RAG检索Top-K控制在3-5，用rerank模型精排。

• 长对话用滑动窗口+摘要，而非全量拼接。

• 用tiktoken预估Token数，设置硬上限。

4. 链式推理必须有“熔断机制”

坑：校验失败→重新生成→再失败→无限循环，Token账单失控。

解：

• 每个循环设置max_retries。

• 用LangGraph的recursion_limit（默认25）兜底。

• 监控每步的Token消耗，设置预算告警。

5. 可观测性不是可选项

坑：线上准确率下降，但不知道是哪一步Prompt退化。

解：

• 用LangSmith或自建Trace，记录每步的输入、输出、Token数、延迟。

• 对关键节点做A/B测试，Prompt变更走灰度发布。

• 把Prompt当代码管理：版本控制、Code Review、回滚机制。

6. Few-shot示例的质量 > 数量

坑：塞10个示例，其中3个格式不一致，模型学到错误模式。

解：

• 3-5个高质量示例足够。

• 示例要覆盖边界情况（如空输入、多意图）。

• 示例格式必须与期望输出严格一致。

7. 别让LLM做它不擅长的事

• 不要让LLM做精确计算 → 用代码解释器工具。

• 不要让LLM做确定性路由 → 用规则引擎或分类小模型。

• 不要让LLM做实时数据查询 → 用工具调用（Function Calling）。

• LLM擅长的是：语义理解、模糊匹配、文本生成、推理链。

---

总结

回到开头那个“智能客服降智”的故事。根因是我们把“简洁”这个模糊指令放进了Prompt，而模型在长上下文里把它理解成了“省略关键步骤”。修复方案不是改那一句话，而是把退款流程拆成意图识别→政策检索→方案生成→事实校验四步，每步独立Prompt、独立测试、独立监控。准确率从67%回到94%，且再也没出现过“火星客服”。

核心要点回顾：

1. Prompt Engineering的本质是接口设计：你定义输入输出契约，模型是概率实现。契约越清晰，实现越稳定。
2. 链式推理解决的是“可验证性”问题：单条Prompt是黑盒，链式推理把黑盒拆成可观测的白盒步骤。
3. LCEL适合线性，LangGraph适合图：选型看控制流复杂度，别用大炮打蚊子。
4. 循环必须有熔断，输出必须结构化，Prompt必须版本化：这三条是生产环境的底线。
5. 可观测性决定迭代速度：没有Trace的Prompt工程就是盲人摸象。

延伸思考：随着模型能力提升，一部分链式推理会被“更长的上下文+更强的推理模型”内化。但这不意味着编排框架会消失——因为可观测性、合规审计、成本控制、人工介入这些工程需求，不会因为模型变强而消失。未来的方向可能是：模型负责推理，框架负责治理。

如果你正在设计Agent系统，建议从LangGraph的StateGraph入手，把业务逻辑画成图再写代码。图想清楚了，代码就水到渠成。