LangChain 原理与实战（二）：用 LCEL 写出第一个 Chain

上一篇把全景摊开了。这一篇动手，目标只有一个：用 LCEL 把"拼提示词、调模型、取文本"串成一条能反复调用的链，并且理解它为什么比手写 API 调用好用。

LCEL 的全称是 LangChain Expression Language，本质就是管道符 |。左边节点的输出，自动喂给右边节点的输入。所有参与方都实现同一个 Runnable 接口，所以模型、提示词、解析器、甚至你自己的函数都能混着接。

先拼最短的一条链

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名资深后端工程师，回答要贴代码。"),
    ("human", "用 {language} 写一个 {task} 的函数，并说明要点。"),
])
parser = StrOutputParser()

chain = prompt | model | parser
print(chain.invoke({"language": "Go", "task": "限流"}))

三段职责很清楚：prompt 把字典填成一组消息，model 调模型拿到 AIMessage，parser 把消息剥成纯字符串。chain.invoke(...) 一口气跑完。

注意 StrOutputParser 不是必需的。不加它，chain 返回的是 AIMessage 对象，你能拿到 .content 也能拿到 token 用量。加了它，你只拿文本。多数时候你要的就是文本，所以默认接上。

模板里的变量和角色

ChatPromptTemplate.from_messages 接收的是「角色, 文本」的元组列表。角色有 system、human、ai 三种，对应对话里的不同身份。变量用 {name} 占位，调用时通过 invoke 的字典填进去。

想让模型基于前文继续，可以塞一条 ai 消息当示例：

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是代码审查助手。"),
    ("human", "这段 Python 有什么问题：{code}"),
    ("ai", "主要问题是没处理空值，且变量命名不清。"),
    ("human", "那给我改一版。"),
])

这种 few-shot 写法在调接口时很常见，比把示例硬塞进 system 提示词更可控。

输出不止字符串

模型有时候该吐结构化结果，比如让你抽取字段。用 PydanticOutputParser 或 JsonOutputParser 直接拿对象：

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class Repo(BaseModel):
    name: str = Field(description="仓库名")
    stars: int = Field(description="star 数")
    language: str = Field(description="主语言")

parser = JsonOutputParser(pydantic_object=Repo)
prompt = ChatPromptTemplate.from_messages([
    ("system", "从用户描述里抽取仓库信息，只输出 JSON。\n{format}"),
    ("human", "{text}"),
]).partial(format=parser.get_format_instructions())

chain = prompt | model | parser
result = chain.invoke({"text": "vue 是一个前端框架，12 万 star，主语言是 TS"})
print(result.name, result.stars)   # vue 120000

parser.get_format_instructions() 会生成一段格式约定，塞进 system 提示词，模型照着吐 JSON，解析器再校验成对象。省掉了你手写正则去抠字段。

并行与分支

链不一定是一条直线。要同时干两件事，用 RunnableParallel：

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel({
    "中文": prompt_zh | model | parser,
    "英文": prompt_en | model | parser,
})
out = chain.invoke({"concept": "索引"})
print(out["中文"], out["英文"])

两个子链拿到同一个输入，各自跑，结果收进字典。这种写法在"一次请求、多视角输出"时很顺手。

想给输入加工但不改结构，用 RunnablePassthrough：

from langchain_core.runnables import RunnablePassthrough

chain = {"query": RunnablePassthrough()} | retriever_chain

它把输入原样透传，常配合字典构造给下游喂字段。

出错和重试

Runnable 自带配置，不用你自己写 try/except：

chain = (prompt | model | parser).with_retry(stop_after_attempt=3)
chain = chain.with_config({"run_name": "回答生成"})

with_retry 在限流或网络抖断时自动重来，with_config 给这段链起个名字，后面接 LangSmith 或本地回调时能看到每一步的耗时和输入输出。

这一篇记住什么

| 把实现了 Runnable 的节点串起来，左出右进。
提示词负责拼输入，模型负责推理，StrOutputParser 负责剥文本。
要结构化就换 JsonOutputParser / PydanticOutputParser。
并行用 RunnableParallel，透传用 RunnablePassthrough。

下一篇进入检索：怎么把一份文档切成块、向量化、再按问题捞回来喂给模型。那是 RAG 的地基。