LangChain LCEL表达式语言与流式管道深度实战：从Runnable组合到流式传播与并行执行的全解析

引言

LangChain Expression Language（LCEL）是LangChain的声明式管道组合语言，以Runnable协议统一了Prompt、LLM、Output Parser、Retriever等组件，支持流式传播、异步执行、批处理、重试与回退。本文从LCEL的核心抽象讲起，覆盖Runnable协议与管道组合、流式传播机制、并行执行与批处理、重试与回退策略、动态路由与条件分支、自定义Runnable、可观测性集成、性能优化，构建LCEL管道的工程体系。

一、Runnable协议

# lcel/runnable.py
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any, AsyncIterator, Generic, TypeVar
import asyncio

T = TypeVar('T')
U = TypeVar('U')

class Runnable(Generic[T, U], ABC):
    """Runnable协议：统一管道组件接口"""
    
    @abstractmethod
    def invoke(self, input: T, config: dict = None) -> U: ...
    
    async def ainvoke(self, input: T, config: dict = None) -> U:
        """异步调用（默认委托同步）"""
        return self.invoke(input, config)
    
    def stream(self, input: T, config: dict = None) -> AsyncIterator[U]:
        """流式输出（默认返回单次结果）"""
        yield self.invoke(input, config)
    
    def batch(self, inputs: list[T], config: dict = None) -> list[U]:
        """批处理"""
        return [self.invoke(i, config) for i in inputs]
    
    async def abatch(self, inputs: list[T], config: dict = None) -> list[U]:
        """异步批处理"""
        tasks = [self.ainvoke(i, config) for i in inputs]
        return await asyncio.gather(*tasks)
    
    # 管道组合操作符
    def __or__(self, other: 'Runnable') -> 'RunnableSequence':
        return RunnableSequence(self, other)
    
    def __ror__(self, other) -> 'RunnableSequence':
        if isinstance(other, Runnable):
            return RunnableSequence(other, self)
        return RunnableLambda(lambda x: x).pipe(self)

class RunnableSequence(Runnable):
    """管道序列：A | B | C"""
    
    def __init__(self, first: Runnable, last: Runnable):
        self.first = first
        self.last = last
    
    def invoke(self, input: Any, config: dict = None) -> Any:
        result = self.first.invoke(input, config)
        return self.last.invoke(result, config)
    
    async def ainvoke(self, input: Any, config: dict = None) -> Any:
        result = await self.first.ainvoke(input, config)
        return await self.last.ainvoke(result, config)
    
    def stream(self, input: Any, config: dict = None) -> AsyncIterator:
        """流式传播：上游的流式输出传递给下游"""
        for chunk in self.first.stream(input, config):
            yield from self.last.stream(chunk, config)

class RunnableLambda(Runnable):
    """函数包装为Runnable"""
    
    def __init__(self, func):
        self.func = func
    
    def invoke(self, input: Any, config: dict = None) -> Any:
        return self.func(input)
    
    async def ainvoke(self, input: Any, config: dict = None) -> Any:
        if asyncio.iscoroutinefunction(self.func):
            return await self.func(input)
        return self.func(input)

class RunnableParallel(Runnable):
    """并行执行多个Runnable"""
    
    def __init__(self, **runnables: Runnable):
        self.runnables = runnables
    
    def invoke(self, input: Any, config: dict = None) -> dict:
        return {k: r.invoke(input, config) for k, r in self.runnables.items()}
    
    async def ainvoke(self, input: Any, config: dict = None) -> dict:
        tasks = {k: r.ainvoke(input, config) for k, r in self.runnables.items()}
        results = await asyncio.gather(*tasks.values())
        return dict(zip(tasks.keys(), results))

class RunnablePassthrough(Runnable):
    """直通：输入即输出（常用于保留原始输入）"""
    
    def invoke(self, input: Any, config: dict = None) -> Any:
        return input

class RunnableWithFallback(Runnable):
    """回退：主Runnable失败时尝试备用"""
    
    def __init__(self, primary: Runnable, fallbacks: list[Runnable]):
        self.primary = primary
        self.fallbacks = fallbacks
    
    def invoke(self, input: Any, config: dict = None) -> Any:
        try:
            return self.primary.invoke(input, config)
        except Exception:
            for fb in self.fallbacks:
                try:
                    return fb.invoke(input, config)
                except Exception:
                    continue
            raise

二、LCEL管道实战

# lcel/pipeline.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

class RAGPipeline:
    """LCEL构建的RAG管道"""
    
    def __init__(self, retriever, llm, embed_model):
        self.retriever = retriever
        self.llm = llm
        
        # Prompt模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "基于以下检索到的信息回答问题。"),
            ("human", "检索结果：{context}\n\n问题：{question}"),
        ])
        
        # LCEL管道：检索 → 组装 → 生成 → 解析
        self.chain = (
            RunnableParallel({
                "context": retriever | (lambda docs: "\n".join(d.page_content for d in docs)),
                "question": RunnablePassthrough(),
            })
            | self.prompt
            | self.llm
            | StrOutputParser()
        )
    
    async def answer(self, question: str) -> str:
        return await self.chain.ainvoke(question)
    
    async def stream_answer(self, question: str):
        """流式回答"""
        async for chunk in self.chain.astream(question):
            yield chunk

class CodeReviewPipeline:
    """代码审查管道：分析 → 安全检查 → 修复建议"""
    
    def __init__(self, llm):
        analyze_prompt = ChatPromptTemplate.from_messages([
            ("system", "分析代码的功能、结构、潜在问题。"),
            ("human", "{code}"),
        ])
        security_prompt = ChatPromptTemplate.from_messages([
            ("system", "基于代码分析检查安全漏洞。"),
            ("human", "分析结果：{analysis}\n代码：{code}"),
        ])
        fix_prompt = ChatPromptTemplate.from_messages([
            ("system", "基于安全检查结果生成修复建议。"),
            ("human", "安全问题：{security}\n代码：{code}"),
        ])
        
        # LCEL管道链
        self.pipeline = (
            RunnableParallel({
                "code": RunnablePassthrough(),
                "analysis": analyze_prompt | llm | StrOutputParser(),
            })
            | RunnableParallel({
                "code": lambda x: x["code"],
                "security": security_prompt | llm | StrOutputParser(),
            })
            | (fix_prompt | llm | StrOutputParser())
        )
    
    async def review(self, code: str) -> str:
        return await self.pipeline.ainvoke(code)

# 带回退的管道
class ResilientChain:
    """带回退与重试的弹性管道"""
    
    def __init__(self):
        primary = ChatOpenAI(model="gpt-4o", temperature=0)
        fallback1 = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        fallback2 = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        
        prompt = ChatPromptTemplate.from_messages([
            ("human", "{input}"),
        ])
        
        # 主模型失败 → 回退到小模型
        self.chain = (
            prompt
            | primary.with_fallbacks([fallback1, fallback2])
            | StrOutputParser()
        )
    
    async def run(self, input: str) -> str:
        # 自动重试3次，失败后回退
        return await self.chain.ainvoke(
            {"input": input},
            config={"max_concurrency": 5},
        )

三、动态路由

# lcel/dynamic_router.py
from langchain_core.runnables import RunnableLambda

class DynamicRouter:
    """动态路由：根据输入选择不同处理链"""
    
    def __init__(self, llm):
        self.llm = llm
        self.routes: dict[str, Runnable] = {}
        
        # 路由分类器
        self.classifier_prompt = ChatPromptTemplate.from_messages([
            ("system", """根据用户问题分类到以下类别之一：
- code：代码相关（编写/调试/审查）
- search：信息检索
- chat：闲聊
- analysis：数据分析
输出类别名（仅名称）。"""),
            ("human", "{input}"),
        ])
        self.classifier = self.classifier_prompt | llm | StrOutputParser()
    
    def add_route(self, name: str, chain: Runnable):
        self.routes[name] = chain
    
    def build(self) -> Runnable:
        """构建路由管道"""
        def route(input_data: dict) -> str:
            # 同步获取路由
            category = self.classifier.invoke({"input": input_data.get("input", "")})
            return category.strip().lower()
        
        return RunnableLambda(route) | {
            name: chain for name, chain in self.routes.items()
        }
    
    # 使用示例
    # router.add_route("code", code_chain)
    # router.add_route("search", search_chain)
    # router.add_route("chat", chat_chain)
    # pipeline = router.build()
    # result = pipeline.invoke({"input": "写一个排序函数"})

四、可观测性

# lcel/observability.py
class LCELObservability:
    """LCEL管道可观测性"""
    
    @staticmethod
    def with_tracing(chain: Runnable, tracer) -> Runnable:
        """添加追踪"""
        original_invoke = chain.invoke
        
        def traced_invoke(input, config=None):
            span = tracer.start_span(chain.__class__.__name__)
            try:
                result = original_invoke(input, config)
                span.set_attribute("input", str(input)[:200])
                span.set_attribute("output", str(result)[:200])
                span.end()
                return result
            except Exception as e:
                span.record_exception(e)
                span.end()
                raise
        
        chain.invoke = traced_invoke
        return chain
    
    @staticmethod
    def with_timing(chain: Runnable) -> Runnable:
        """添加计时"""
        original = chain.ainvoke
        
        async def timed(input, config=None):
            import time
            start = time.monotonic()
            result = await original(input, config)
            duration = time.monotonic() - start
            print(f"[{chain.__class__.__name__}] {duration:.3f}s")
            return result
        
        chain.ainvoke = timed
        return chain

总结

LCEL以Runnable协议统一了LLM应用的所有组件：管道操作符|让Prompt→LLM→Parser的组合像Unix管道一样直观，RunnableParallel实现多路并行检索与生成，流式传播让token从LLM流经Parser直达用户无中间缓冲，with_fallbacks实现模型级回退（GPT-4o失败→mini→3.5），动态路由根据LLM分类结果选择不同处理链，批处理与max_concurrency控制并发度。可观测性以span级追踪每个Runnable的输入输出与耗时。当LLM应用从"命令式代码拼接"进化为"声明式管道组合"，LCEL让管道的构建、调试、优化与监控都获得了统一的抽象层。