AI-NEWS

Appearance

LCEL 与 Runnable ​

🎯 学习目标

  • 用 pipe（|）组合 prompt、model、parser 形成 RunnableSequence
  • 使用 RunnableLambda、RunnablePassthrough、RunnableParallel 处理分支与并行
  • 掌握 invoke、batch、stream、astream_events 的适用场景

引言 ​

LCEL（LangChain Expression Language） 是 LangChain 0.1+ 的核心组合方式。所有 Runnable 共享同一调用协议，便于测试、流式与观测挂钩。

章节正文 ​

第 1 步：基础管道 ​

python
from langchain_core.runnables import RunnablePassthrough

rag_chain = (
    RunnablePassthrough.assign(context=lambda x: retriever.invoke(x["question"]))
    | RunnablePassthrough.assign(
        answer=lambda x: (
            prompt.partial(context=format_docs(x["context"]))
            | llm
            | StrOutputParser()
        ).invoke({"question": x["question"]})
    )
)
result = rag_chain.invoke({"question": "向量库如何选型？"})

更常见的扁平写法：

python
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

第 2 步：RunnableParallel 与 RunnableLambda ​

python
from langchain_core.runnables import RunnableParallel, RunnableLambda

def word_count(text: str) -> int:
    return len(text.split())

analysis_chain = RunnableParallel(
    summary=prompt | llm | StrOutputParser(),
    meta=RunnableLambda(lambda x: {"chars": len(x["text"]), "words": word_count(x["text"])}),
)
analysis_chain.invoke({"text": "LangChain 让 LLM 应用更易组合。"})
# → {"summary": "...", "meta": {"chars": 24, "words": 5}}

并行支路适合同时做摘要与元数据统计，互不阻塞。

第 3 步：条件分支 RunnableBranch ​

python
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "代码" in x["topic"], code_prompt | llm | StrOutputParser()),
    (lambda x: "法律" in x["topic"], legal_prompt | llm | StrOutputParser()),
    general_prompt | llm | StrOutputParser(),  # default
)
branch.invoke({"topic": "代码", "question": "如何用 Python 读 JSON？"})

与 Workflow 边界 一致：分支可枚举时用 Branch，路径不可预测再用 Agent。

第 4 步：stream 与 astream_events ​

python
# 流式最终 token
for chunk in chain.stream({"question": "解释 LCEL"}):
    print(chunk, end="", flush=True)

# 调试：观察每一步事件
async for event in chain.astream_events({"question": "..."}, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

astream_events 是 Agent 调试利器，可看到 retriever、tool、llm 各阶段耗时，对接 7.3 Trace。

动手练习 ​

  1. 用 RunnableParallel 实现：一路生成标题，一路生成正文摘要
  2. 为「技术 / 产品 / 其他」三类问题写 RunnableBranch 路由链
  3. 对 RAG 链开启 stream，在前端或终端逐字打印

常见问题 ​

Q：LCEL 和旧版 Chain 类（LLMChain）区别？

LLMChain 等已 legacy；新代码只用 LCEL Runnable，API 更统一且原生支持 async/stream。

本节小结 ​

LCEL 用 | 组合 Runnable，统一 invoke/batch/stream。RunnablePassthrough、Parallel、Branch 处理透传、并行与条件路由。astream_events 用于链路调试。