使用者痛点：在传统的 Agent 开发中，代码往往冗长且难以维护；批量处理和流式输出需要自行循环实现；错误重试、日志记录还有多轮对话的历史管理都缺少统一的抽象，导致项目后期调试成本高。

Runnable 是 LangChain 里统一的执行协议

Prompt、Model、Output Parser、普通函数、分支逻辑、并行逻辑，都可以被看成 Runnable。

再看先记住一句话。Runnable 不是某一个具体功能，而是一套统一的调用方式。

它统一的是这几件事：

• invoke处理一个输入。
• batch批量处理多个输入。
• stream流式返回结果。
• config给本次运行附加配置。
• |把多个 Runnable 串成一条链。

这套链式写法也叫 LCEL。LCEL 的主要不是新语法，而是把 Agent 流程表达成一条可组合、可追踪、可批量、可流式执行的链。

典型公式：

Prompt | Model | Parser

为什么需要 Runnable

痛点体现：没有 Runnable 时需要手动写每一步的调用并显式传递结果，代码重复且易出错。

传统写法：

formatted_prompt = prompt.format
response = model.invoke
result = parser.invoke

使用 Runnable 的写法：

chain = prompt | model | parser
result = chain.invoke

这里不是少执行了步骤，而是把步骤提前组装成一条链。数据流示意：

输入 → prompt.invoke → model.invoke → parser.invoke → 输出

User Pain Point："每次改动都要手动同步所有步骤，维护成本极高"

安装依赖

uv add langchain-core

Runnable 的主要 API

Avoid diving into every combination class at first;focus on three execution APIs:

• invoke: 单次处理一个输入。
• batch: 同时处理多个输入，提高吞吐。
• stream: 边生成边返回，实现实时交互。

The three APIs all accept an optional .

chain.invoke
chain.batch
chain.stream

invoke

If a plain function lacks .invoke。wrap it with RunnableLambda .

from langchain_core.runnables import RunnableLambda
def add_one -> int:
return number + 1
chain = RunnableLambda
print) # 输出 6

This makes ordinary functions part of LangChain’s unified execution system.

batch
from langchain_core.runnables import RunnableLambda
chain = RunnableLambda)
print)
#


User Pain Point： "我想一次性处理上百篇文档，却只能循环单调地调用 invoke"


批量处理文档 chunk。

批量生成摘要。

batch 而非手写循环。


stream
from collections.abc import Iterator
from langchain_core.runnables import RunnableLambda
def stream_words -> Iterator:
for word in text.split:
yield word
chain = RunnableLambda
for chunk in chain.stream:
print
# 输出每个单词


The real LLM stream returns small chunks . In Agents this is used for:


页面上逐字显示回答。

长任务实时显示进度。

调试链路中间过程。


运行配置 config 

The config  object is not business input;怎么说呢，it supplies auxiliary information for a single run.
config = {
"configurable": {
"locale": "zh-CN","session_id": "user-123",},"tags":。"metadata": {"page": "agent-runnable"},"run_name": "translate_demo",}



configurable  用于自定义参数，如语言、使用者 ID 等。

tags / metadata / run_name  用于观测与追踪，便于日志或 LangSmith 定位。

from langchain_core.runnables import RunnableConfig,RunnableLambda
translations = {
"zh-CN": {"greeting": "你好"},"en-US": {"greeting": "Hello"}。}
def translate -> str:
locale = config.get.get
return translations.get
chain = RunnableLambda
print)
print)


The business input and run configuration are clearly separated.

LCEL 与组合 API 

The core APIs answer “how to execute”. The combination APIs answer “how to compose multiple Runnables into a workflow”. Four typical composition patterns:


串行：RunnableSequence / | 

并行：RunnableParallel 

保留中间结果：RunnablePassthrough 

条件分支：RunnableBranch 


串行组合 – RunnableSequence 

Each step’s output automatically becomes next step’s input.
`input -> add_one -> multiply_two -> to_text -> output` 


The pipe operator is most readable way;you can also instantiate RunnableSequence explicitly.

实际案例 

最常见的 Agent 链路就是 Prompt → Model → Parser。下面用普通函数模拟模型，以免依赖真实 LLM 配置。
按理说，python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
prompt = PromptTemplate.from_template
def fake_model -> str:
# PromptTemplate.invoke 返回的是 PromptValue 对象。需要转成字符串发送给模型。prompt_text = prompt_value.to_string
return f"模型收到：{prompt_text}
从回答来看，Runnable 是统一执行协议。"
model = RunnableLambda
parser = StrOutputParser
# 通过管道符号串联三段逻辑
chain = prompt | model | parser
print)
Data flow:
{"concept":"Runnable"} → PromptTemplate → fake_model → StrOutputParser → str
All three components implement `Runnable` protocol。so y can be linked with `|`.

并行组合 – RunnableParallel 

User Pain Point：在同一次请求里需要同时生成标题、摘要和关键词，却只能顺序调用，导致响应时间不必要地拉长。


串行 : 前一步输出→后一步输入 

并行 : 同一输入→多分支→合并为 dict 

保留中间结果 – RunnablePassthrough.assign 

User Pain Point：“我需要在后续步骤里
使用原始使用者提问或检索到的文档，但默认只返回最终一步结果。”
Solution: use `Passthrough` to merge intermediate data.
条件分支 – RunnableBranch
`RunnableBranch` works like an `if/elif/else` chain.
python
from langchain_core.runnables import RunnableBranch,RunnableLambda
branch = RunnableBranch(
RunnableLambda),
,RunnableLambda),# 默认分支
RunnableLambda


)

print)


Execution order:


是否 VIP → 否 → 是否首单 → 是 → 执行新使用者分支 → 返回结果


Agent 场景常见分支：


判断是否需要调用工具 

判断是否走 RAG 检索 

判断是否需要结构化解析 

判断错误是否需要重试 


API 

The enhancement layer wraps existing Runnables without changing ir core logic.


.

with_fallbacks.

with_listeners.

callbacks.
<\/table\>


Name
Description 

with_config
Simplifies repeated passing of identical runtime settings.

with_retrySolves transient failures such as API time‑outs or rate limits.
Saves you from writing manual try/except‑fallback blocks when one strategy fails.
Adds lightweight lifecycle logging without pulling in full callback infrastructure.
Dives deeper—captures token‑level events。tool calls,retriever steps etc.,ideal for observability dashboards.





绑定默认配置 – with_config 

If a configuration is reused many times,bind it once:

python
from langchain_core.runnables import RunnableLambda

def translate:
locale=config.get.get
return f"{locale}: {key}"

base_chain = RunnableLambda

zhchain = basechain.withconfig(configurable={"locale":"zh-CN"},tags=,metadata={"src":"docs"})
enchain = basechain.withconfig(configurable={"locale":"en-US"},tags=,metadata={"src":"docs"})

print)
print)


with_config returns a new runnable;original remains unchanged.

失败重试 – with_retry 

Agents often hit temporary service glitches . 

class TemporaryServiceError: pass

attempts=0
def unstable_search:
global attempts
attempts+=1
if attempts <3:
        raise TemporaryServiceError
    return f"检索结果：{query}"

searchchain =
.withretry,stopafterattempt=5。waitexponentialjitter=False))

Use retry only for transient issues—not for logic or permission errors.

失败降级 – with_fallbacks 

When one strategy permanently fails,fall back to alternatives:

def vectorsearch: raise RuntimeError
def keywordsearch: return
def empty_result: return 

retrieve =
.with_fallbacks,CallableFallback)]))

Difference: with_retry → same method multiple tries;with_fallbacks → 切换到其他实现。

运行监听 – with_listeners 

Add simple start/end/error hooks:

python
from langchain_core.runnables import CallableLambd a。Run na bleListener
from typing import Any

def normalize->str:return text.strip.lower

def onstart: print
def onend: print
def on_err: print

pipeline =
.with_listeners)

If you only care about overall chain lifecycle,with_listeners is enough;orwise use full callbacks for token‑level tracing.

Callbacks 

Callbacks capture events from every component .

python
config={
"callbacks":,"tags":。"metadata":{"user_id":"user-42"}
}
result = chain.invoke


Use callbacks when you need precise latency metrics,token usage stats or UI progress updates.

RunnableWithMessageHistory 

User Pain Point：“多轮对话时我必须自己维护会话状态，代码散落在不同层。”
Solution: wrap any chain with history management.

bash
uv add langchain-openai


python
import os
from langchaincore.chathistory import InMemoryChatMessageHistory
from langchan gcore.prompts im po rt ChatPromptTemplate,Messa gesPlaceholder
fro mlangch aincor e.r unnab le s.history impor t RuneableW ithMessageHi story
fro mlangcha in_openai imp ortChatOpenAI 

model = ChatOpenAI,apikey=os.getenv,baseurl=os.getenv,temperature=.7)

prompt = ChatPromptTemplate.from_messages()

base_chain =)

histories={}
def gethistory->InMemoryChatMessageHistory:
if sessionid not in histories:
histories=InMemoryChatMessageHistory
return histories

chatchain =(Runna b leWithMessageHistory(
basechain。gethistory,inputmessageskey ="question",historymessages_key ="history"
))

cfg={"configurable":{"session_id":"user-001"}}
print,cfg )
print)


Key points:

1️⃣  tells prompt where historic messages go.
2️⃣ get_history fetches/creates per‑session storage.
3️⃣ configurable.session_id selects which history to read/write.

Different session_ids ⇒ isolated conversation histories.

Runnable 类型？​​​​​​​​​​​


需求场景
对应 API/组合方式​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​‬‬‬‬‬‬‬‬‬‍‍‍‍‍‍‍‍‌‏‏‏‏‏‏‏‏‌‎‎‎ ‎⁠⁠⁠⁠⁠⁠⁠ ⁣⁣⁣︎︎︎⟨⟩  ‌️️️️️️⁤⁤⁤⁤⬛⬛⬛⬛⬛⬛⬛⬛⬛⬛ ⬜ ⬜ ⬜ ⬜ ⬜ ⬜ ⬜ ‌‌‌‌‌‌‮‮‪  ‑‑‑ ​—–———–———–—————–——-

User Pain Points:
① 代码冗余且难以维护；不过，② 批量与流式需求需自行循环实现；③ 错误重试与降级逻辑散落各处；④ 缺少统一日志与监控，⑤ 多轮对话时历史管理繁琐。这些痛点正是引入 **Runnable** 的根本原因。

什么是 ?LangChain 的统一执行协议


Prompt、Model、Output Parser、普通函数、分支逻辑还有并行逻辑，都可以视作 Runnable 。它本身不是具体功能，而是一套统一的调用方式。它统一以下几件事：



invoke: 单次处理一个输入；

batch: 批量处理多个输入；其实，

streamconfig| : 把多个 Pipelined Runnables  连成一条链。






这套链式写法被称为 LCEL，其价值在于把 Agent 流程抽象为「可组合·可追踪·可批量·可流式」的一条执行链。TYPICAL FORMULA:


Prompt | Model | Parser



为什么要使用 ?

原始命令式写法













---















--

-------


-

-
-

-
-
-
-
-
-
-
-
-
-




The rest of your response seems truncated due to an internal error.
Please try again.

Sorry!It appears re was an unexpected problem while formatting response. Let me try again and provide you complete HTML content as requested.

Below is a fresh attempt.