LangChain：LCEL（表达式语言）使用注意事项

LCEL 是 LangChain 的表达式语言，是一种编程范式，是一种语法糖，使代码看起来非常简洁，但同时隐藏了更多需要注意事项，避免踩坑犯错。

chain = prompt | model | parser 

使用注意事项

LCEL表达式语言的管道符|是有严格的先后顺序要求，顺序决定了数据流的类型和结构，错误的顺序会导致运行时错误。

LCEL 中的 | 管道操作符并不是简单的“拼接”，而是将前一个组件的输出作为后一个组件的输入。因此，顺序必须遵循各组件之间 输入/输出类型的匹配规则。

两侧都是Runnable对象

|操作符两侧都必须是Runnable对象。最常见的错误是直接将一个普通 Python 对象（如字符串）放在 | 后面。LCEL 期望每一个操作单元都是一个 Runnable，这是它的设计原则。

输入输出类型匹配

核心规则：类型必须匹配

每个 Runnable 组件（如 PromptTemplate、ChatModel、OutputParser）都有其预期的输入类型和输出类型，所以在写 LCEL 链时，首先明确每个组件的输入类型和输出类型。管道传递的本质就是：A | B 要能工作，必须满足 A.output_type ⊆ B.input_type（即 A 的输出是 B 能接受的输入）。

| 连接的两个组件，左边组件的输出类型必须是**右边组件的输入类型。例如，ChatPromptTemplate 输出 PromptValue，ChatOpenAI 接收 PromptValue 并输出 AIMessage，而 StrOutputParser 接收 AIMessage 并输出 str。

最常见的：PromptTemplate → ChatModel → OutputParser

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

prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
model = ChatOpenAI()
parser = StrOutputParser()

chain = prompt | model | parser   # 正确顺序
result = chain.invoke({"topic": "程序员"})

1. 输入 {"topic": "程序员"} → PromptTemplate 期望一个 dict，输出一个 PromptValue（或字符串化的消息列表）。
2. PromptValue → ChatModel 期望 PromptValue 或消息序列，输出 BaseMessage。
3. BaseMessage → StrOutputParser 期望 BaseMessage，输出 str。

错误1：ChatModel 在前，PromptTemplate 在后

# 错误：类型不匹配
chain = prompt | model | prompt  # ❌ 会报错，model输出是AIMessage，不能直接给prompt

# 正确：使用输出解析器转换
chain = prompt | model | StrOutputParser() | another_prompt

• model 输出 BaseMessage，但 prompt 期望输入 dict（包含模板变量）。BaseMessage 无法直接转换为 dict，因此抛出 TypeError。

错误2：OutputParser 在前，ChatModel 在后

# 错误示例
chain = parser | model   # ❌ 会报错

• parser 输出 str（或其他解析后的类型），而 model 通常期望 PromptValue、字符串或消息序列。直接传递 str 在某些模型下可能勉强工作（如旧版 LLM 接口），但不推荐，且会丢失 prompt 模板的能力。更重要的是，如果你的 parser 输出的是复杂结构（如 dict），model 很可能无法处理。

错误3：两个 PromptTemplate 直接连接

prompt1 = ChatPromptTemplate.from_template("先总结：{text}")
prompt2 = ChatPromptTemplate.from_template("再翻译：{summary}")

chain = prompt1 | prompt2   # ❌ 通常无意义且会报错

• prompt1 输出 PromptValue，但 prompt2 期望 dict。除非你手动包装一个适配器，否则类型不匹配。

字典格式的正确使用

LCEL 链条中的数据通常以字典（dict）的形式在组件间传递。确保每个组件的输入变量名与上一步输出的键（key）匹配。例如，prompt 需要 {topic}，那么前一个组件的输出或 invoke 的输入就必须包含 topic 这个键。

# 正确：RunnableParallel 或 字典字面量
chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough()
} | prompt | model

# 注意：字典的每个值必须是 Runnable 或可调用对象

提示词模板变量匹配

这是最常见的错误之一。如果 PromptTemplate 中定义了 {variable_name}，那么在调用链时，输入的字典中必须包含 variable_name 这个键，否则会抛出 ValueError。

输出解析器失败

输出解析器（如 JsonOutputParser）期望模型输出特定格式。如果模型输出不符合预期，解析会失败。调试时，可以先移除解析器，打印模型的原始输出，检查格式问题，然后通过优化提示词（明确告知模型输出格式）来解决。

上下文窗口超限

如果输入文本过长，加上提示词模板后可能会超出模型的最大 token 限制。可以使用 LangChain 的 TextSplitter 将长文档切分成小块来处理。

Agent的思考过程（agent_scratchpad）会随着工具调用次数增加而变长。对于复杂任务，可能需要设置 max_iterations防止无限循环，或使用能处理长上下文的模型。

内存管理(长链注意)

# 长文档处理时，注意 token 限制
long_chain = (
    load_long_docs 
    | (lambda docs: docs[:3])  # 限制文档数量，避免超限
    | format_docs
    | prompt
    | llm
)

版本兼容性与调试

遇到类型错误，可以分段执行（如先 chain1 = prompt | model，再单独 chain1.invoke(...) 观察输出类型）来调试。

LCEL 链本身就像一个“黑盒”，内部的中间结果不容易直接查看。可以使用 langchain_core.callbacks 模块中的 ConsoleCallbackHandler，它能在执行时打印每个组件的输入和输出。

LangChain 迭代非常快，不同版本间可能存在不兼容。建议使用 langchain-core 和 langchain-community 等拆分后的包，并锁定版本。

对于复杂的链条，调试可能比较困难，可以考虑使用 LangSmith 这样的工具进行全链路跟踪和可视化。

优先使用管道符|

优先使用 | 而非 .pipe()：管道符 | 是组合 LCEL 链的首选和推荐方式，因为它更简洁、直观，符合 Python 语言习惯。.pipe() 方法作为备选，主要在需要动态构建链时使用，例如在循环中根据条件添加组件。

拥抱现代化API

拥抱现代化API，舍弃旧用法：在 langchain.chains 模块中的 LLMChain、SequentialChain 等旧式链（Legacy Chains）已被标记为过时（deprecated），并将在未来的版本中被移除。请确保学习和实践都基于最新的 LCEL 语法。

Agent与Tool调试

1. verbose=True是你的好朋友：

   Agent开发阶段，将 verbose=True传递给 AgentExecutor或复杂链，可以打印出每一步的详细执行过程，极大方便调试。

   sqrt_tool = Tool(
       name="sqrt_calculator",
       func=calculate_sqrt,
       description="计算一个数字的平方根。输入应为单个数字。"
   )
   tools = [sqrt_tool]
   
   # 定义提示模板
   prompt = ChatPromptTemplate.from_messages([
       ("system", "你是一个专业的数学助手。请使用合适的工具来回答问题。"),
       ("human", "{input}"),
       ("placeholder", "{agent_scratchpad}"),  # 这是关键！用于存放Agent的思考过程和工具调用记录
   ])
   
   agent = create_tool_calling_agent(llm, tools, prompt)
   agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

2. **Agent的提示模板中必须包含 {agent_scratchpad}**：

这是一个占位符，LangChain会自动将Agent的思考过程和工具调用记录填充进去。忘记它，Agent将无法进行多步推理。

3. 工具描述要清晰准确：
   Tool的 description字段是LLM选择工具的唯一依据。描述必须清晰说明工具的功能、输入格式和输出格式。

异步与流式

# LCEL 链自动支持异步
result = await chain.ainvoke({"topic": "AI"})
# 或批量处理
results = await chain.abatch([{"topic": "AI"}, {"topic": "ML"}])

# 流式输出
for chunk in chain.stream({"input": "讲个故事"}):
    print(chunk, end="", flush=True)

# 批量处理
results = chain.batch([{"input": "A"}, {"input": "B"}, {"input": "C"}])

调试与追踪

# 查看链的结构
print(chain.get_graph().print_ascii())

# 开启详细日志
import langchain
langchain.debug = True

单步调试

# 使用 .get_graph() 可视化链结构
print(chain.get_graph().draw_ascii())

# 逐步测试
step1 = prompt | llm
print(step1.invoke({"x": "test"}))  # 先验证单步

step2 = step1 | parser
print(step2.invoke({"x": "test"}))  # 再验证下一步

状态传递

# 使用 RunnablePassthrough.assign 保留原始输入
from langchain_core.runnables import RunnablePassthrough

chain = RunnablePassthrough.assign(
    summary=(lambda x: x["text"][:100])  # 添加新字段，保留原有字段
) | prompt | model

调整顺序或适配类型

当需要实现非标准顺序时，可以使用 LCEL 提供的适配器组件：

场景	解决方案	示例
前一步输出类型不满足下一步输入	RunnableLambda 或 RunnablePassthrough 进行转换	`chain = prompt1
需要同时传递多个输入（分支合并）	RunnableParallel	`{“summary”: chain1, “topic”: chain2}
在管道中插入一个无输入的函数	RunnablePassthrough.assign()	保留原有输入并添加新字段
从 BaseMessage 中提取字符串内容	使用 StrOutputParser()	`model

LCEL 的 | 管道要求前后类型匹配。当不匹配时，使用 RunnableLambda、RunnablePassthrough、StrOutputParser 等适配器进行类型转换，或重新设计管道逻辑。

示例：修复“两个 prompt 不能直接连”的问题

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

prompt1 = ChatPromptTemplate.from_template("总结：{text}")
prompt2 = ChatPromptTemplate.from_template("将总结翻译成英文：{summary}")

model = ChatOpenAI()

# 错误写法：prompt1 | model | prompt2   ❌ (model输出BaseMessage, prompt2要dict)
# 正确写法：提取 model 输出的文本内容，并重命名为 summary
chain = (
    prompt1 
    | model 
    | StrOutputParser()                     # BaseMessage -> str
    | (lambda summary: {"summary": summary}) # str -> dict {"summary": ...}
    | prompt2 
    | model 
    | StrOutputParser()
)

result = chain.invoke({"text": "LangChain 是一个强大的框架"})
print(result)