LangChain LCEL：面向生产级AI应用的高级模式

为什么选择LCEL而非传统LangChain

传统LangChain：LLMChain、SequentialChain、RouterChain——各自独立的类，接口不一致。难以组合、难以流式处理、难以调试。

LCEL（LangChain表达式语言）：管道操作符（|）可组合任意可运行组件。统一接口：每个组件都支持 invoke()、stream()、batch()、ainvoke()、astream()、abatch()。像Unix管道一样构建链。

python
传统方式
from langchain.chains import LLMChain
chain = LLMChain(llm=llm, prompt=prompt)
LCEL方式
chain = prompt | llm | output_parser

每个LCEL链自动支持：流式处理、异步执行、批处理、重试逻辑。

核心LCEL构建块

Runnables

任何实现了Runnable接口的可调用对象：PromptTemplate、ChatOpenAI、OutputParser、Lambda函数（RunnableLambda）、字典（RunnableParallel）。

python
from langchain_core.runnables import RunnableLambda, RunnableParallel
Lambda作为可运行对象
double = RunnableLambda(lambda x: x * 2)
并行执行
parallel_chain = RunnableParallel({
    "summary": summary_chain,
    "keywords": keyword_chain,
    "sentiment": sentiment_chain
})
三个链并行执行，结果合并为字典

流式处理

LCEL链默认支持流式处理。使用astream()进行异步流式处理：

python
async def stream_response(question: str):
    chain = prompt | llm | StrOutputParser()
    async for chunk in chain.astream({"question": question}):
        yield chunk  # 随着token生成而产出
使用流式回调
from langchain.callbacks import StreamingStdOutCallbackHandler
llm = ChatOpenAI(streaming=True, callbacks=[StreamingStdOutCallbackHandler()])

高级模式

模式1：动态路由

python
from langchain_core.runnables import RunnableBranch
route = RunnableBranch(
    (lambda x: x["topic"] == "technical", technical_chain),
    (lambda x: x["topic"] == "billing", billing_chain),
    default_chain  # 回退
)
或者使用LLM决定路由
router_chain = router_prompt | llm | JsonOutputParser()
def route_based_on_llm(input):
    route_decision = router_chain.invoke(input)
    if route_decision["route"] == "technical":
        return technical_chain.invoke(input)
    elif route_decision["route"] == "billing":
        return billing_chain.invoke(input)
    return general_chain.invoke(input)
intelligent_router = RunnableLambda(route_based_on_llm)

模式2：重试与回退

python
from langchain_core.runnables import RunnableWithFallbacks
失败时重试
chain_with_retry = chain.with_retry(
    retry_if_exception_type=(RateLimitError,),
    stop_after_attempt=3,
    wait_exponential_jitter=True
)
主模型失败时回退到不同模型
primary_chain = prompt | ChatOpenAI(model="gpt-4o")
fallback_chain = prompt | ChatOpenAI(model="gpt-4o-mini")
chain_with_fallback = primary_chain.with_fallbacks([fallback_chain])

模式3：基于LCEL的RAG流水线

python
from langchain_core.runnables import RunnablePassthrough
from langchain.retrievers import EnsembleRetriever
混合检索器（语义+关键词）
retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)
def format_docs(docs):
    return "\n\n".join([d.page_content for d in docs])
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)
流式输出答案
async for token in rag_chain.astream("什么是量子计算？"):
    print(token, end="", flush=True)

模式4：多步推理链

python
步骤1：提取关键信息
extract_chain = extract_prompt | llm | JsonOutputParser()
步骤2：研究每个提取的主题
def research_topics(extracted: dict) -> dict:
    results = {}
    for topic in extracted["topics"]:
        results[topic] = research_chain.invoke({"topic": topic})
    return {**extracted, "research": results}
步骤3：综合最终答案
synthesis_chain = synthesis_prompt | llm | StrOutputParser()
完整流水线
pipeline = (
    extract_chain
    | RunnableLambda(research_topics)
    | synthesis_chain
)

模式5：带流式处理的工具调用

python
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
@tool
def search_database(query: str) -> str:
    """搜索公司数据库以获取相关信息。"""
    return db.search(query)
@tool  
def send_email(to: str, subject: str, body: str) -> str:
    """向客户发送电子邮件。"""
    return email_client.send(to, subject, body)
agent = create_tool_calling_agent(llm, [search_database, send_email], agent_prompt)
executor = AgentExecutor(agent=agent, tools=[search_database, send_email], verbose=True)
流式处理智能体执行过程
async for event in executor.astream_events({"input": "研究订单#123并通过电子邮件将状态告知John"}):
    if event["event"] == "on_tool_end":
        print(f"使用的工具：{event['name']}，结果：{event['data']}")
    elif event["event"] == "on_chain_stream":
        print(event["data"]["chunk"], end="", flush=True)

测试LCEL链

使用模拟LLM进行单元测试

python
from langchain_core.runnables import RunnableLambda
def test_classification_chain():
    # 模拟LLM，返回确定性输出
    mock_llm = RunnableLambda(lambda _: "positive")
    
    test_chain = classification_prompt | mock_llm | StrOutputParser()
    
    result = test_chain.invoke({"text": "I love this product!"})
    assert result == "positive"

使用录制的集成测试

使用LangSmith录制真实的LLM调用，然后回放进行回归测试：

录制：对真实LLM运行链，将输入/输出保存到LangSmith数据集

回放：对录制的输入运行链，将输出与保存的响应进行比较

告警：当输出发生显著变化时标记

性能优化

批处理：LCEL的batch()方法可并发处理多个输入，并支持可配置的并发数：

python
results = chain.batch(inputs, config={"max_concurrency": 20})

缓存：添加缓存层以避免重复的LLM调用：

python
from langchain.globals import set_llm_cache
from langchain.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))
相同的输入将返回缓存结果

流式处理提升用户体验：即使后端处理需要5秒，流式处理也能让用户立即获得反馈。面向用户的聊天应用始终使用流式处理。