1. 项目概述为什么我们需要“万物皆 Runnable”如果你最近在折腾大语言模型应用开发尤其是用过 LangChain 这个框架大概率会对一个词感到既熟悉又困惑Runnable。它无处不在从最简单的模型调用到复杂的多步推理链再到有状态、带循环的智能体工作流最后都被封装成了一个叫Runnable的东西。这听起来有点“过度设计”一个接口就想统一所有但当你真正上手构建一个从简单到复杂的应用时才会发现这个设计背后的精妙之处。简单来说Runnable是 LangChain 框架中的一个核心抽象接口。它的目标是把 AI 应用开发中所有可执行的“计算单元”都标准化。无论是调用一个 OpenAI 的 GPT 模型执行一个检索增强生成RAG的链条还是运行一个由 LangGraph 定义的有向图工作流你都可以把它们当作一个Runnable对象来对待。这意味着你可以用一套完全相同的 API比如.invoke(),.batch(),.stream()来操作它们也可以像搭积木一样把不同的Runnable组合成更复杂的结构。这解决了什么痛点回想一下没有统一接口的日子。调用模型是一个库的用法定义链是另一套语法到了图Graph可能又是全新的概念。当你试图把一个链的输出作为另一个模型的输入或者想把一个链嵌入到一个更大的循环工作流中时你需要写大量的胶水代码来处理不同对象之间的数据传递和格式转换。这不仅容易出错也让代码的可读性和可维护性急剧下降。Runnable的出现就是为了消灭这些“接缝”让开发者能够专注于业务逻辑本身而不是底层对象的适配问题。所以这个标题“万物皆 Runnable”并非夸张它精准地概括了 LangChain 试图构建的开发者体验通过一个高度一致的接口降低认知负担提升组合效率让构建复杂 AI 应用变得像拼装乐高一样直观。接下来我们就深入这个“接口”的内部看看它是如何做到这一点的。2. Runnable 接口设计统一模型的“万能插座”要理解Runnable如何统一万物首先得看看这个“插座”本身长什么样。在 LangChain 中Runnable是一个协议Protocol它定义了一组必须实现的方法。最核心的几个方法是.invoke(input: Any) - Any: 同步调用输入一些东西得到输出。.batch(inputs: List[Any]) - List[Any]: 批量同步调用提升处理效率。.stream(input: Any) - Iterator[Any]: 流式调用用于逐词或逐块生成内容。.astream(input: Any) - AsyncIterator[Any]: 异步流式调用。看到这里你可能会想这不就是普通函数吗没错Runnable在理念上就是把一切计算单元都视为一个“函数”。但这个函数是加强版的它内置了对批处理、流式输出、异步操作以及类型安全的原生支持。2.1 核心方法解析与类型安全以.invoke为例它的强大之处在于其背后的类型系统。一个Runnable[Input, Output]是泛型的它声明了自己接受的输入类型Input和产生的输出类型Output。例如Runnable[str, str]: 表示一个输入字符串、输出字符串的 Runnable比如一个简单的聊天模型。Runnable[Dict, Dict]: 表示输入输出都是字典的 Runnable这在链Chain中很常见因为链通常在不同步骤间传递结构化的数据。这种类型声明不是摆设。当你组合多个Runnable时LangChain 的类型检查如果配合像 Pydantic 这样的工具可以帮助你在开发阶段就发现前后步骤输入输出不匹配的问题而不是等到运行时才报错。这是构建可靠、复杂工作流的关键保障。注意虽然 Python 是动态类型语言但充分利用Runnable的泛型提示和像mypy这样的静态类型检查工具可以极大提升大型 LangChain 项目的代码健壮性。我个人的习惯是为每个自定义的复杂链或图都明确定义其输入输出 Pydantic 模型。2.2 Runnable 的“子类”生态模型、链与图Runnable是一个抽象基类具体的功能由其众多的“子类”实现。它们构成了 LangChain 的三大支柱RunnableLambda 与自定义 Runnable这是最基础的实现。你可以用RunnableLambda将任何一个普通 Python 函数包装成Runnable。这为集成现有代码或实现简单转换逻辑提供了入口。from langchain_core.runnables import RunnableLambda def add_prefix(text: str) - str: return f前缀_{text} runnable_func RunnableLambda(add_prefix) result runnable_func.invoke(测试) # 输出前缀_测试模型Models作为 Runnable这是最直观的应用。ChatOpenAI、ChatAnthropic等聊天模型OpenAIEmbeddings等嵌入模型在 LangChain 中都是Runnable的子类。你可以直接对它们调用.invoke()发送消息。from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4) # 模型本身就是一个 Runnable response model.invoke(你好世界)链Chains作为 Runnable链是多个 Runnable 的顺序组合。LangChain 提供了|操作符类似于 Unix 管道来优雅地连接它们。连接后产生的整体本身也是一个Runnable。from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(请翻译以下中文到英文{input}) model ChatOpenAI(modelgpt-3.5-turbo) output_parser StrOutputParser() # 使用管道操作符组合成一个链这个链本身是 Runnable translation_chain prompt | model | output_parser result translation_chain.invoke({input: 今天天气真好}) # 输出The weather is really nice today.图Graphs作为 Runnable这是 LangChain 更高级的抽象通常通过LangGraph库实现。你可以定义包含条件分支和循环的状态机。当这个图被编译compile后它同样返回一个Runnable对象你可以用相同的方式调用它。from langgraph.graph import StateGraph, END # ... 假设定义了一个复杂的、带状态的工作流 graph compiled_graph graph.compile() # 编译后的图就是一个 Runnable final_state compiled_graph.invoke({input: 用户查询, chat_history: []})通过这种方式无论底层是简单的函数、复杂的模型调用、多步骤的链还是带循环的图在开发者眼中它们都变成了一个具有统一接口的“黑盒”。你只需要关心它的输入和输出而无需关心里面具体是哪种实现。这种抽象极大地简化了系统架构。3. 链Chain的构建用管道思维组装 Runnable链是 LangChain 中最常用的模式它代表了线性的、多步骤的处理流程。Runnable接口让构建链变得异常直观和灵活核心思想就是“管道”Pipeline。3.1 管道操作符|的魔法如前所述|操作符是连接Runnable的粘合剂。它的工作方式非常符合直觉前一个Runnable的输出自动成为后一个Runnable的输入。这种设计让代码看起来就像在描述数据流动的管道图清晰易懂。但这里有一个关键细节输入输出的类型必须兼容。如果Runnable A输出一个字符串而Runnable B的.invoke方法期望输入一个字典那么直接A | B就会在运行时出错。因此链中的每个环节都需要设计好其“接口”。3.2 构建一个完整的 RAG 链从检索到生成让我们用一个实际的检索增强生成RAG例子来演示如何用Runnable管道构建一个复杂链。假设我们已经有一个向量数据库如 Chroma并存储了文档。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableParallel # 1. 初始化组件 vectorstore Chroma(persist_directory./chroma_db, embedding_functionOpenAIEmbeddings()) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索器也是一个 Runnable model ChatOpenAI(modelgpt-4) output_parser StrOutputParser() # 2. 定义提示词模板 template 请根据以下上下文来回答问题。如果你不知道答案就老实说不知道。 上下文 {context} 问题{question} 请用中文给出答案 prompt ChatPromptTemplate.from_template(template) # 3. 构建链 # 这是一个经典的 RAG 链包含检索和生成两个主要阶段。 rag_chain ( # 第一步并行处理。同时保留原始问题并检索相关上下文。 RunnableParallel({ context: retriever, # 输入是 question输出是检索到的文档列表 question: RunnablePassthrough() # 将输入原封不动传递下去 }) | prompt # 第二步将上一步输出的字典包含context和question填充到提示词模板 | model # 第三步将填充好的提示词发送给大模型 | output_parser # 第四步解析模型的输出如 AIMessage为字符串 ) # 4. 调用链 answer rag_chain.invoke(LangChain 是什么) print(answer)关键点解析RunnableParallel: 它允许你并行执行多个Runnable并将结果合并成一个字典。在这里我们同时执行检索和传递问题。RunnablePassthrough: 一个特殊的Runnable它只是将输入原样输出。在这里用于保留原始问题。数据流invoke(“LangChain 是什么”)-RunnableParallel输出{“context”: [doc1, doc2, doc3], “question”: “LangChain 是什么”}-prompt接收这个字典并渲染成最终提示文本 -model生成回答 -output_parser提取文本。实操心得在构建复杂链时我强烈建议使用RunnableParallel来显式定义每个步骤的输入结构。这比依赖隐式的参数传递要清晰和可靠得多。调试时你可以在任何两个Runnable之间插入一个RunnableLambda(lambda x: print(f”Step Output: {x}”))来查看中间数据这是定位问题最快的方法。3.3 链的调试与可视化当链变得复杂时理解数据流可能变得困难。LangChain 提供了langchain.debug True的全局调试模式但它可能输出过多信息。更精准的方法是使用RunnableLambda进行打点或者利用 LangSmith 这样的追踪平台。此外由于链本身就是Runnable你可以很容易地将其一部分子链单独提出来测试这符合单元测试的思想。例如你可以单独测试retriever.invoke(“某问题”)看检索结果是否相关或者测试prompt | model看模型在给定上下文下的生成质量。4. 图Graph与状态管理当链需要循环和分支链适合线性流程但很多 AI 应用场景需要更复杂的控制流比如智能体Agent根据模型输出决定是调用工具还是直接给出最终答案。多轮对话需要维护对话历史状态。复杂决策流程包含条件判断和循环。这就是LangGraph的用武之地。它建立在Runnable之上用于构建有状态、可循环的图工作流。4.1 LangGraph 核心概念状态State与节点Node在 LangGraph 中你首先定义一个State它通常是一个 TypedDict 或 Pydantic 模型描述了在整个图执行过程中需要维护的所有数据。例如一个聊天智能体的状态可能包括用户输入、聊天历史、中间步骤、最终答案。然后你定义多个节点Node。每个节点本质上就是一个Runnable它接收当前的State执行一些操作比如调用模型、调用工具并返回一个更新后的State或部分更新。最后你定义边Edges即节点之间的流转逻辑。边可以是固定的也可以是基于State内容的条件边这就引入了分支和循环。4.2 构建一个简单的 ReAct 智能体图下面我们构建一个极简版的 ReAct 智能体它可以使用一个“计算器”工具。from typing import TypedDict, Annotated, Sequence import operator from langgraph.graph import StateGraph, END from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI from langchain.tools import tool from langchain.agents import create_react_agent # 1. 定义状态 class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 关键这是一个消息列表操作符add表示追加 next: str # 指示下一步该去哪个节点 # 2. 定义工具 tool def calculator(expression: str) - str: 计算一个数学表达式例如 ‘(23)*4’. try: # 警告实际生产中切勿使用eval此处仅为演示 return str(eval(expression)) except: return 计算错误 # 3. 初始化模型并绑定工具 model ChatOpenAI(modelgpt-3.5-turbo).bind_tools([calculator]) # 4. 定义节点函数每个函数都返回更新后的状态 def call_model(state: AgentState): 调用模型节点 messages state[“messages”] response model.invoke(messages) # 模型是一个 Runnable # 将模型的响应追加到消息历史中 return {“messages”: [response], “next”: “tools” if response.tool_calls else “end”} def call_tool(state: AgentState): 调用工具节点 messages state[“messages”] last_message messages[-1] tool_calls last_message.tool_calls tool_messages [] for tc in tool_calls: tool {t.name: t for t in [calculator]}[tc[“name”]] # 根据名称找到工具 result tool.invoke(tc[“args”]) # 工具也是 Runnable (通过tool装饰) tool_messages.append(ToolMessage(contentresult, tool_call_idtc[“id”])) # 将工具执行结果追加到消息历史 return {“messages”: tool_messages, “next”: “model”} # 执行完工具后回到模型节点 # 5. 构建图 graph StateGraph(AgentState) graph.add_node(“model”, call_model) # 添加节点call_model 函数本身可视为 Runnable graph.add_node(“tools”, call_tool) graph.add_node(“end”, lambda state: state) # 结束节点 # 6. 设置边 graph.set_entry_point(“model”) # 入口是模型节点 # 从 model 节点出发根据 state[“next”] 的值决定去向 graph.add_conditional_edges( “model”, lambda state: state[“next”], # 路由函数返回下一个节点的名称 {“tools”: “tools”, “end”: “end”} ) graph.add_edge(“tools”, “model”) # 从 tools 节点无条件回到 model 节点 graph.add_edge(“end”, END) # 连接到结束标志 # 7. 编译图 app graph.compile() # 8. 运行图编译后的 app 就是一个 Runnable initial_state {“messages”: [HumanMessage(content“请问 (1234)*2 等于多少”)], “next”: “”} final_state app.invoke(initial_state) for msg in final_state[“messages”]: print(f”{msg.type}: {msg.content}“)流程解读用户输入问题进入model节点。模型思考后可能决定调用calculator工具并生成一个包含tool_calls的响应。状态中的next被设为”tools”。根据条件边图流转到tools节点。该节点执行工具调用将结果作为ToolMessage追加。next被设为”model”。从tools节点有固定边回到model节点。模型收到工具执行结果生成最终答案。这次没有工具调用next被设为”end”。根据条件边图流转到end节点然后结束。这个例子展示了Runnable理念的延伸图中的每个节点函数虽然看起来是普通函数但它们操作的是标准化的State并且可以被视为一个计算单元。最终编译成的app完美地封装了所有复杂性对外暴露的依然是简单的.invoke()接口。5. 高级特性与生产实践掌握了Runnable、Chain和Graph的基础后我们来看看一些能让你在生产环境中游刃有余的高级特性和实践。5.1 批量处理、流式输出与异步支持这是Runnable接口带来的直接好处。无论你的计算单元多复杂都可以轻松获得这些能力。批量处理.batch当你需要处理大量输入时使用.batch()可以显著提升效率因为底层可能会并行调用或进行优化。questions [“问题1”, “问题2”, “问题3”] answers rag_chain.batch([{“question”: q} for q in questions])流式输出.stream / .astream对于生成模型流式输出能极大提升用户体验。由于链和图都是Runnable你可以直接对整个复杂流程进行流式调用。for chunk in rag_chain.stream({“question”: “请解释AI”}): print(chunk, end“”, flushTrue) # 逐词或逐块输出异步Async在现代 Web 应用中异步操作至关重要。所有Runnable都提供了ainvoke,abatch,astream等异步方法。import asyncio async def process(): result await rag_chain.ainvoke({“question”: “异步问题”}) asyncio.run(process())5.2 自定义 Runnable 与中间件有时你需要实现一些框架未提供的特殊逻辑。你可以通过继承Runnable基类或使用RunnableLambda来创建自定义Runnable。更强大的模式是使用中间件。Runnable支持with_config方法你可以注入一些在调用前后执行的逻辑比如日志记录、性能监控、重试机制、输入输出格式化等。这类似于 Web 框架中的中间件允许你以非侵入式的方式增强功能。from langchain_core.runnables import RunnableLambda import time import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def logging_middleware(runnable): 一个简单的日志记录中间件 def wrapper(input_data): logger.info(f”开始调用 {runnable.__class__.__name__}输入: {input_data}“) start_time time.time() output runnable.invoke(input_data) end_time time.time() logger.info(f”调用完成耗时: {end_time - start_time:.2f}s输出: {output}“) return output # 将包装函数转化为 Runnable return RunnableLambda(wrapper) # 包装一个链 logged_chain logging_middleware(translation_chain) logged_chain.invoke({“input”: “测试”})5.3 配置管理与上下文Runnable可以通过.with_config()方法携带配置信息比如模型温度、API 密钥、回调函数等。配置可以在运行时被覆盖这为 A/B 测试、多租户环境提供了便利。from langchain_core.callbacks import StdOutCallbackHandler chain_with_config translation_chain.with_config( run_name“MyTranslationChain”, # 给这个运行起个名字便于在 LangSmith 追踪 callbacks[StdOutCallbackHandler()], # 添加回调 configurable{“temperature”: 0.7} # 可配置参数 ) # 调用时临时覆盖配置 result chain_with_config.invoke( {“input”: “测试”}, config{“configurable”: {“temperature”: 0.9}} # 临时使用更高的温度 )5.4 与 LangSmith 集成可观测性的关键对于生产系统可观测性Observability至关重要。你需要知道你的链或图在哪里出错了、性能瓶颈在哪、模型的输入输出具体是什么。LangChain 的亲兄弟LangSmith就是为此而生。任何Runnable的执行都可以被 LangSmith 自动追踪。你只需要设置好环境变量 (LANGSMITH_API_KEY,LANGSMITH_TRACINGtrue)所有的invoke、batch、stream调用都会被记录。你可以在 LangSmith 的 UI 中看到详细的执行轨迹、每一步的输入输出、耗时和 token 消耗。这是调试复杂工作流、分析成本、监控异常不可或缺的工具。生产环境建议在开发阶段就接入 LangSmith。它不仅能帮你快速定位问题其“数据集测试”和“在线评估”功能还能帮助你系统地评估和迭代你的 AI 应用效果确保每次改动都是可衡量、可回溯的。6. 常见问题、排查技巧与性能优化在实际使用中你一定会遇到各种问题。下面是我踩过的一些坑和总结的经验。6.1 输入输出类型不匹配这是最常见的问题。错误可能表现为ValidationError或晦涩的键错误。排查在链的每个步骤后插入RunnableLambda打印中间状态。确保前一个节点的输出字典的键与后一个节点输入所期望的键完全匹配。技巧善用 Pydantic 模型来定义链的输入和输出。这能在编码阶段就捕获许多类型错误。from pydantic import BaseModel class ChainInput(BaseModel): question: str chat_history: list [] # 你可以用 RunnableLambda 来验证输入 validate_input RunnableLambda(lambda x: ChainInput(**x))6.2 图Graph陷入无限循环在 LangGraph 中如果条件边逻辑有误可能导致状态在几个节点间无限循环。排查为你的State添加一个step_count计数器并在每次经过主要节点时递增。在条件边判断中如果step_count超过某个阈值比如 20就强制跳转到END。技巧在开发阶段使用app.invoke(initial_state, debugTrue)可以打印出每一步的状态转换帮助你理解图的执行路径。6.3 性能瓶颈复杂的链或图可能很慢。瓶颈通常出现在网络 I/O调用外部模型 API、检索向量数据库。顺序执行某些本可并行的步骤被顺序执行了。优化并行化用RunnableParallel将独立的操作并行。例如在 RAG 中检索文档和生成查询改写Query Rewriting可以同时进行。批处理对于大量数据务必使用.batch()方法它能减少网络往返开销。缓存对于昂贵的操作如嵌入生成、固定提示词的模型调用考虑使用Runnable的缓存中间件或外部缓存如 Redis。异步化在 Web 服务中使用异步调用避免阻塞。6.4 错误处理与重试网络请求可能失败模型 API 可能限流。策略使用tenacity库或 LangChain 内置的RunnableRetry中间件为脆弱的Runnable特别是模型调用添加重试逻辑。from langchain_core.runnables import RunnableRetry retry_model RunnableRetry( boundmodel, retry_if_exception_type(Exception,), # 重试的异常类型 stop_after_attempt3, # 最大重试次数 wait_exponential_jitterTrue, # 指数退避 )兜底方案在关键链的最后可以设置一个RunnableLambda作为错误处理器捕获异常并返回一个友好的默认答案。6.5 提示词Prompt管理混乱当链很多时硬编码在代码中的提示词会难以维护。最佳实践将提示词模板提取到外部文件如 YAML、JSON或数据库中。使用ChatPromptTemplate.from_messages()动态加载。可以为不同场景、不同模型版本管理不同的提示词。6.6 记忆Memory管理在多轮对话中如何管理chat_history是关键。注意直接将所有历史对话都塞进上下文会导致 token 消耗快速增长和模型性能下降。策略摘要记忆定期将长的对话历史总结成一段摘要。滑动窗口只保留最近 N 轮对话。向量存储记忆将历史对话存入向量数据库每次只检索与当前问题最相关的片段。这本质上是将 RAG 技术用于记忆管理。LangChain 提供了多种记忆后端ConversationBufferMemory,ConversationSummaryMemory,VectorStoreRetrieverMemory可以根据Runnable的模式进行集成通常是通过在State中维护messages列表来实现。7. 总结与展望Runnable 生态的思考回过头看“万物皆 Runnable” 不仅仅是一个技术实现更是一种设计哲学。它通过极致的抽象和一致性将 AI 应用开发的复杂度封装在标准接口之后。开发者从“如何调用不同对象”的琐碎中解放出来得以更专注于业务逻辑和创新本身。这种设计也带来了极佳的可组合性和可测试性。你可以像单元测试函数一样测试每个Runnable也可以轻松地将一个小链替换成更复杂的图而系统的其他部分几乎无需改动。随着 LangChain 生态的发展Runnable接口正在成为事实上的标准。越来越多的第三方工具和库开始提供Runnable兼容的接口这使得技术选型和集成变得更加容易。当你掌握了Runnable的核心思想你不仅是在学习一个框架更是在掌握一种构建现代、可维护、可扩展的 AI 应用的方法论。最后一个我个人的体会是初期学习 LangChain 时可能会被其众多的概念和抽象层所困扰。但一旦你理解了Runnable这个核心一切都会变得清晰起来。我的建议是不要一开始就试图构建复杂的智能体而是从将一个简单的函数包装成Runnable开始然后用管道|连接两个模型调用逐步增加复杂度。在实践中你会深刻体会到这种统一接口带来的优雅和力量。