LangChain Core 源码学习路线：第 1 篇 Runnable 源码解剖#

1
from langchain_core.prompts import ChatPromptTemplate
2
from langchain_core.output_parsers import StrOutputParser
3
from langchain.chat_models import init_chat_model
4

5
model = init_chat_model("openai:gpt-4.1-mini")
6

7
prompt = ChatPromptTemplate.from_messages([
8
    ("system", "你是一个旅行规划助手。"),
9
    ("user", "请为 {destination} 规划 {days} 天行程。")
10
])
11

12
chain = prompt | model | StrOutputParser()
13

14
result = chain.invoke({
15
    "destination": "东京",
16
    "days": 5
17
})
18

19
print(result)

1
先格式化 prompt，再调用模型，再转成字符串。

1
1. ChatPromptTemplate 是一个 Runnable。
2
2. ChatModel 是一个 Runnable。
3
3. StrOutputParser 是一个 Runnable。
4
4. `|` 调用 Runnable.__or__。
5
5. __or__ 会调用 coerce_to_runnable，把右侧对象转成 Runnable。
6
6. __or__ 返回 RunnableSequence。
7
7. 第二个 `| parser` 继续构造更长的 RunnableSequence。
8
8. chain.invoke(input) 实际调用 RunnableSequence.invoke(input)。
9
9. RunnableSequence.invoke 按 steps 顺序调用每个子 Runnable。
10
10. 每一步输出成为下一步输入。

1
print(type(prompt))
2
print(type(model))
3
print(type(StrOutputParser()))
4
print(type(chain))
5

6
print(chain.get_graph())
7
print(chain.input_schema.model_json_schema())
8
print(chain.output_schema.model_json_schema())

1
Runnable 是一个可以被调用、批处理、流式输出、转换、组合的工作单元。

1
class Runnable:
2
    def invoke(self, input, config=None, **kwargs):
3
        # 子类最应该实现的核心能力
4
        raise NotImplementedError
5

6
    async def ainvoke(self, input, config=None, **kwargs):
7
        # 默认异步：把同步 invoke 放进 executor
8
        return await run_in_executor(
9
            config,
10
            self.invoke,
11
            input,
12
            config,
13
            **kwargs,
14
        )
15

16
    def batch(self, inputs, config=None, **kwargs):
17
        # 默认批处理：每个 input 仍然走 invoke，只是由 executor 并发调度
18
        configs = get_config_list(config, len(inputs))
19
        with get_executor_for_config(configs[0]) as executor:
20
            return list(
21
                executor.map(
22
                    lambda pair: self.invoke(pair[0], pair[1], **kwargs),
23
                    zip(inputs, configs),
24
                )
25
            )
26

27
    def stream(self, input, config=None, **kwargs):
28
        # 默认流式：没有真正的 chunk 转换，只是最后 yield 一次完整结果
29
        yield self.invoke(input, config, **kwargs)

1
“支持 stream 方法”不等于“真正 token 级流式输出”。

1
如果你希望前端 SSE 逐 token 展示最终行程，链路中的中间步骤不能随便放阻塞型 RunnableLambda。
2
否则流式输出会被阻塞在该步骤之后才开始。

1
chain = prompt | model | parser

1
chain_1 = prompt.__or__(model)
2
chain_2 = chain_1.__or__(parser)

1
RunnableSequence(prompt, model, parser)

1
class Runnable:
2
    def __or__(self, other):
3
        return RunnableSequence(
4
            self,
5
            coerce_to_runnable(other),
6
        )
7

8
    def __ror__(self, other):
9
        return RunnableSequence(
10
            coerce_to_runnable(other),
11
            self,
12
        )
13

14
    def pipe(self, *others):
15
        chain = self
16
        for other in others:
17
            chain = chain | other
18
        return chain

1
{"context": retriever, "question": RunnablePassthrough()} | prompt

1
def coerce_to_runnable(thing):
2
    if isinstance(thing, Runnable):
3
        # 已经是 Runnable，直接进入链
4
        return thing
5

6
    if is_async_generator(thing) or is_generator_function(thing):
7
        # 生成器函数更适合表达 chunk-by-chunk 的转换
8
        return RunnableGenerator(thing)
9

10
    if callable(thing):
11
        # 普通函数包装成 RunnableLambda
12
        return RunnableLambda(thing)
13

14
    if isinstance(thing, dict):
15
        # dict 的每个 value 会继续递归转换成 Runnable
16
        return RunnableParallel({
17
            key: coerce_to_runnable(value)
18
            for key, value in thing.items()
19
        })
20

21
    raise TypeError

1
1. 把每个对象统一包装成 Runnable；
2
2. 把多个 Runnable 注册进 RunnableSequence。

1
声明一条可执行、可追踪、可批处理、可流式、可配置的执行管线。

1
from langchain_core.runnables import RunnableLambda
2

3

4
def normalize_destination(input: dict) -> dict:
5
    return {
6
        **input,
7
        "destination": input["destination"].strip()
8
    }
9

10
normalize_node = RunnableLambda(normalize_destination)
11

12
chain = normalize_node | prompt | model | StrOutputParser()

1
chain = normalize_destination | prompt | model | StrOutputParser()

1
如何把普通 Python 业务逻辑放进 LangChain 的统一执行协议？

1
字段清洗
2
输入归一化
3
输出格式修正
4
日志补充
5
规则判断
6
轻量数据转换

1
1. 接收一个 Python callable；
2
2. 检查它是同步函数、异步函数、生成器函数还是异步生成器函数；
3
3. invoke 时调用同步函数；
4
4. ainvoke 时优先调用异步函数，否则委托线程池；
5
5. batch 时继承 Runnable 默认 batch；
6
6. 如果函数返回另一个 Runnable，则继续调用返回的 Runnable。

1
class RunnableLambda(Runnable):
2
    def __init__(self, func, afunc=None, name=None):
3
        self.func = func
4
        self.afunc = afunc
5
        self.name = name or func.__name__
6

7
    def invoke(self, input, config=None, **kwargs):
8
        config = ensure_config(config)
9
        callback_manager = get_callback_manager_for_config(config)
10
        run_manager = callback_manager.on_chain_start(
11
            serialized=self.get_serialized(),
12
            inputs=input,
13
            name=config.get("run_name") or self.name,
14
        )
15

16
        try:
17
            output = call_func_with_variable_args(
18
                self.func,
19
                input,
20
                config,
21
                run_manager=run_manager,
22
                **kwargs,
23
            )
24

25
            if isinstance(output, Runnable):
26
                # 函数可以动态返回下一个 Runnable，形成运行时分支
27
                child_config = patch_config(
28
                    config,
29
                    callbacks=run_manager.get_child(),
30
                )
31
                output = output.invoke(input, child_config)
32

33
            run_manager.on_chain_end(output)
34
            return output
35

36
        except BaseException as e:
37
            run_manager.on_chain_error(e)
38
            raise

1
def call_func_with_variable_args(func, input, config, run_manager, **kwargs):
2
    if accepts_config(func):
3
        kwargs["config"] = config
4

5
    if accepts_run_manager(func):
6
        kwargs["run_manager"] = run_manager
7

8
    return func(input, **kwargs)

1
def f(x): ...
2

3
def f(x, config): ...
4

5
def f(x, run_manager): ...
6

7
def f(x, run_manager, config): ...

1
from langchain_core.runnables import RunnableConfig
2

3

4
def normalize_request(input: dict, config: RunnableConfig) -> dict:
5
    metadata = config.get("metadata", {})
6
    trace_id = metadata.get("trace_id")
7

8
    return {
9
        **input,
10
        "trace_id": trace_id,
11
        "destination": input["destination"].strip()
12
    }

1
chain = prompt | model | RunnableLambda(clean_text) | parser

1
非流式预处理：放在 prompt 前
2
非流式后处理：放在最终输出后
3
需要流式处理：实现 transform / RunnableGenerator

1
chain.invoke(input)
2
  ↓
3
ChatPromptTemplate.invoke(input)
4
  ↓
5
生成 PromptValue / messages
6
  ↓
7
ChatModel.invoke(messages)
8
  ↓
9
返回 AIMessage
10
  ↓
11
StrOutputParser.invoke(AIMessage)
12
  ↓
13
返回字符串

1
chain.invoke(input, config=None)
2
  ↓
3
RunnableSequence.invoke(input, config)
4
  ↓
5
ensure_config(config)
6
  ↓
7
创建 root run / callback manager
8
  ↓
9
遍历 steps：
10
    step_1 = ChatPromptTemplate
11
    step_2 = ChatModel
12
    step_3 = StrOutputParser
13
  ↓
14
每一步执行前 patch_config：
15
    为子步骤创建 child callback manager
16
    写入 seq:step:1 / seq:step:2 / seq:step:3
17
  ↓
18
output_1 = step_1.invoke(input, child_config)
19
  ↓
20
output_2 = step_2.invoke(output_1, child_config)
21
  ↓
22
output_3 = step_3.invoke(output_2, child_config)
23
  ↓
24
callback_manager.on_chain_end(output_3)
25
  ↓
26
return output_3

1
输入：dict
2
{
3
  "destination": "东京",
4
  "days": 5
5
}
6

7
ChatPromptTemplate 输出：PromptValue
8
PromptValue 可转 messages：
9
[
10
  SystemMessage(content="你是一个旅行规划助手。"),
11
  HumanMessage(content="请为 东京 规划 5 天行程。")
12
]
13

14
ChatModel 输出：AIMessage
15
AIMessage(content="以下是东京 5 天行程...")
16

17
StrOutputParser 输出：str
18
"以下是东京 5 天行程..."

1
模型输出不是普通字符串，而是 AIMessage。
2
AIMessage 既可以有 content，也可以有 tool_calls、usage_metadata、response_metadata。
3
Parser 只是把 AIMessage 进一步转成你需要的业务结构。

1
按顺序调用多个 Runnable，前一个 Runnable 的输出作为下一个 Runnable 的输入。

1
x0 = input
2
x1 = step1.invoke(x0)
3
x2 = step2.invoke(x1)
4
x3 = step3.invoke(x2)
5
return x3

1
chain = prompt | model | parser

1
1. prompt.__or__(model)
2
2. coerce_to_runnable(model)
3
3. RunnableSequence(prompt, model)
4
4. sequence.__or__(parser)
5
5. coerce_to_runnable(parser)
6
6. RunnableSequence(prompt, model, parser)

1
first
2
middle
3
last

1
steps = [first, *middle, last]

1
def invoke(self, input, config=None, **kwargs):
2
    # 1. 标准化 config，并把父级上下文配置合进来
3
    config = config_with_context(
4
        ensure_config(config),
5
        self.steps,
6
    )
7

8
    # 2. 当前 sequence 自己先创建一个 root run
9
    callback_manager = get_callback_manager_for_config(config)
10
    run_manager = callback_manager.on_chain_start(
11
        serialized=self.get_serialized(),
12
        inputs=input,
13
        name=config.get("run_name") or self.get_name(),
14
        run_id=config.get("run_id"),
15
    )
16

17
    try:
18
        for i, step in enumerate(self.steps):
19
            # 3. 每个 step 派生一个 child config
20
            child_config = patch_config(
21
                config,
22
                callbacks=run_manager.get_child(f"seq:step:{i + 1}")
23
            )
24

25
            # 4. 把 child_config 写入当前上下文，内部嵌套调用也能读到
26
            context = copy_context()
27
            context.run(_set_config_context, child_config)
28

29
            # 5. 第一个 step 接收外部 kwargs，后续 step 只接收上一步输出
30
            if i == 0:
31
                input = context.run(step.invoke, input, child_config, **kwargs)
32
            else:
33
                input = context.run(step.invoke, input, child_config)
34

35
        run_manager.on_chain_end(input)
36
        return input
37

38
    except BaseException as e:
39
        run_manager.on_chain_error(e)
40
        raise

1
[chain.invoke(x) for x in inputs]

1
先把所有 inputs 批量送进 step1；
2
再把 step1 的所有 outputs 批量送进 step2；
3
再把 step2 的所有 outputs 批量送进 step3。

1
inputs_0 = [case1, case2, case3]
2
inputs_1 = step1.batch(inputs_0)
3
inputs_2 = step2.batch(inputs_1)
4
inputs_3 = step3.batch(inputs_2)
5
return inputs_3

1
如果某个底层模型 API 支持真正 batch，它可以在自己的 batch 中优化；
2
如果不支持，Runnable 默认 batch 会用线程池并发执行 invoke。

1
results = chain.batch([
2
    {"destination": "东京", "days": 5},
3
    {"destination": "大阪", "days": 3},
4
    {"destination": "京都", "days": 4},
5
])

1
transform: Iterator[InputChunk] -> Iterator[OutputChunk]

1
input stream
2
  ↓
3
step1.transform
4
  ↓
5
step2.transform
6
  ↓
7
step3.transform
8
  ↓
9
output stream

1
step1 支持流式
2
step2 不支持流式，需要等完整输入
3
step3 支持流式
4

5
结果：stream 会从 step2 完成后才开始继续往下游输出。

1
如果你的旅行规划助手要做 SSE 流式响应，
2
不要在最终模型输出前随便插入阻塞型 RunnableLambda。

1
RunnableGenerator
2
或自定义 Runnable 并实现 transform / atransform

1
tags: list[str]
2
metadata: dict[str, Any]
3
callbacks: callback handlers 或 callback manager
4
run_name: 当前运行名称
5
run_id: 当前运行 ID
6
max_concurrency: 最大并发数
7
recursion_limit: 最大递归深度
8
configurable: 运行时可配置字段

1
result = chain.invoke(
2
    {"destination": "东京", "days": 5},
3
    config={
4
        "run_name": "travel_plan_chain",
5
        "tags": ["travel", "runnable-sequence", "demo"],
6
        "metadata": {
7
            "user_id": "u_001",
8
            "trace_id": "trace_20260608_001"
9
        },
10
        "max_concurrency": 5,
11
    }
12
)

1
def ensure_config(config=None):
2
    empty = {
3
        "tags": [],
4
        "metadata": {},
5
        "callbacks": None,
6
        "recursion_limit": DEFAULT_RECURSION_LIMIT,
7
        "configurable": {},
8
    }
9

10
    # 1. 先继承 ContextVar 里的父级配置
11
    if var_child_runnable_config.get():
12
        empty = merge_configs(empty, var_child_runnable_config.get())
13

14
    # 2. 再合并当前调用显式传入的 config
15
    if config is not None:
16
        for key, value in config.items():
17
            if key in CONFIG_KEYS:
18
                empty[key] = merge_config_value(empty.get(key), value)
19
            else:
20
                # 未知字段不会丢，通常进入 configurable
21
                empty["configurable"][key] = value
22

23
    # 3. 简单 configurable 字段同步到 metadata，方便 trace 检索
24
    for key, value in empty["configurable"].items():
25
        if is_simple_value(value) and not key.startswith("__"):
26
            empty["metadata"].setdefault(key, value)
27

28
    return empty

1
child_config = patch_config(
2
    config,
3
    callbacks=run_manager.get_child("seq:step:1")
4
)

1
def patch_config(config, **overrides):
2
    copied = config.copy()
3

4
    for key, value in overrides.items():
5
        if value is not None:
6
            copied[key] = value
7

8
    return ensure_config(copied)

1
父运行：travel_plan_chain
2
  子运行：seq:step:1 prompt
3
  子运行：seq:step:2 model
4
  子运行：seq:step:3 parser

1
base_chain = prompt | model | parser
2

3
chain = base_chain.with_config(
4
    tags=["travel"],
5
    metadata={"component": "planner"}
6
)
7

8
chain.invoke(
9
    input,
10
    config={
11
        "tags": ["eval"],
12
        "metadata": {"case_id": "C001"}
13
    }
14
)

1
tags: travel + eval
2
metadata: component + case_id
3
callbacks: 合并或继承
4
configurable: 合并

1
链路级 tags：标注模块，如 travel / planner / parser
2
请求级 metadata：标注业务请求，如 user_id / session_id / case_id
3
评估级 metadata：标注测试样本，如 dataset / case_id / expected_intent

1
chain 开始
2
chain 结束
3
chain 报错
4
LLM 开始
5
LLM 结束
6
tool 开始
7
tool 结束
8
retriever 开始
9
retriever 结束

1
from langchain_core.tracers import ConsoleCallbackHandler
2

3
chain.invoke(
4
    {"destination": "东京", "days": 5},
5
    config={
6
        "callbacks": [ConsoleCallbackHandler()],
7
        "tags": ["debug", "travel"],
8
        "metadata": {"case_id": "debug_001"},
9
    }
10
)

1
on_chain_start: RunnableSequence
2
  on_chain_start: ChatPromptTemplate
3
  on_chain_end: ChatPromptTemplate
4

5
  on_chat_model_start: ChatModel
6
  on_llm_end: ChatModel
7

8
  on_parser_start: StrOutputParser
9
  on_parser_end: StrOutputParser
10
on_chain_end: RunnableSequence

1
root_run = callback_manager.on_chain_start("RunnableSequence", input)
2

3
for i, step in enumerate(steps):
4
    child_config = patch_config(
5
        config,
6
        callbacks=root_run.get_child(f"seq:step:{i + 1}"),
7
    )
8
    output = step.invoke(input, child_config)
9

10
root_run.on_chain_end(output)

1
intent_chain = (intent_prompt | model | intent_parser).with_config(
2
    run_name="intent_chain",
3
    tags=["travel", "intent", "runnable"]
4
)
5

6
planner_chain = (planner_prompt | model | planner_parser).with_config(
7
    run_name="planner_chain",
8
    tags=["travel", "planner", "runnable"]
9
)

1
from langchain_core.runnables import RunnableParallel
2

3
parallel = RunnableParallel({
4
    "food": food_chain,
5
    "attractions": attraction_chain,
6
    "weather": weather_chain,
7
})
8

9
result = parallel.invoke({
10
    "destination": "东京",
11
    "days": 5
12
})

1
{
2
    "food": "东京美食推荐...",
3
    "attractions": "东京景点推荐...",
4
    "weather": "东京天气建议..."
5
}

1
parallel = {
2
    "food": food_chain,
3
    "attractions": attraction_chain,
4
    "weather": weather_chain,
5
}

1
chain = {
2
    "food": food_chain,
3
    "attractions": attraction_chain,
4
    "weather": weather_chain,
5
} | merge_prompt | model | parser

1
给每个子 Runnable 同一个输入，
2
并行执行，
3
最后把每个子 Runnable 的输出按 key 合并成 dict。

1
def invoke(self, input, config=None):
2
    config = ensure_config(config)
3
    callback_manager = get_callback_manager_for_config(config)
4
    run_manager = callback_manager.on_chain_start(
5
        serialized=self.get_serialized(),
6
        inputs=input,
7
        name=config.get("run_name") or self.get_name(),
8
    )
9

10
    steps = self.steps__
11
    futures = {}
12

13
    try:
14
        with get_executor_for_config(config) as executor:
15
            for key, step in steps.items():
16
                child_config = patch_config(
17
                    config,
18
                    callbacks=run_manager.get_child(f"map:key:{key}"),
19
                )
20

21
                futures[key] = executor.submit(
22
                    step.invoke,
23
                    input,
24
                    child_config,
25
                )
26

27
            output = {
28
                key: future.result()
29
                for key, future in futures.items()
30
            }
31

32
        run_manager.on_chain_end(output)
33
        return output
34

35
    except BaseException as e:
36
        run_manager.on_chain_error(e)
37
        raise

1
parallel.invoke(
2
    input,
3
    config={"max_concurrency": 3}
4
)

1
景点检索、餐厅检索、天气查询、交通建议
2
可以并行。
3

4
但创建订单、扣款、退款、写数据库
5
不能盲目并行，必须看副作用和事务顺序。

1
并行检索多个来源
2
并行调用多个轻量模型
3
并行生成多个候选方案
4
并行做多个维度评分

1
research_chain = {
2
    "attractions": attraction_retriever | format_docs,
3
    "food": food_retriever | format_docs,
4
    "weather": weather_tool,
5
    "transport": transport_tool,
6
} | synthesis_prompt | model | parser

1
信息收集阶段：RunnableParallel
2
综合生成阶段：RunnableSequence
3
整体任务范式：Plan-and-Execute / Workflow 中的某个节点

1
如果满足条件 A，执行 chain_a；
2
如果满足条件 B，执行 chain_b；
3
否则执行 default_chain。

1
from langchain_core.runnables import RunnableBranch
2

3
branch = RunnableBranch(
4
    (lambda x: "预算" in x["user_request"], budget_chain),
5
    (lambda x: "景点" in x["user_request"], attraction_chain),
6
    general_chain,
7
)

1
def invoke(self, input, config=None):
2
    config = ensure_config(config)
3
    callback_manager = get_callback_manager_for_config(config)
4
    run_manager = callback_manager.on_chain_start(
5
        serialized=self.get_serialized(),
6
        inputs=input,
7
    )
8

9
    try:
10
        for index, (condition, branch) in enumerate(self.branches):
11
            condition_config = patch_config(
12
                config,
13
                callbacks=run_manager.get_child(f"condition:{index + 1}"),
14
            )
15

16
            matched = condition.invoke(input, condition_config)
17

18
            if matched:
19
                branch_config = patch_config(
20
                    config,
21
                    callbacks=run_manager.get_child(f"branch:{index + 1}"),
22
                )
23
                output = branch.invoke(input, branch_config)
24
                run_manager.on_chain_end(output)
25
                return output
26

27
        default_config = patch_config(
28
            config,
29
            callbacks=run_manager.get_child("branch:default"),
30
        )
31
        output = self.default.invoke(input, default_config)
32
        run_manager.on_chain_end(output)
33
        return output
34

35
    except BaseException as e:
36
        run_manager.on_chain_error(e)
37
        raise

1
简单：根据用户请求走预算 chain / 景点 chain → RunnableBranch
2
复杂：多轮状态、工具失败重试、人工确认 → LangGraph conditional edge

1
如果只是链内部的轻量条件选择，用 RunnableBranch。
2
如果需要状态持久化、循环、回滚、HITL，用 LangGraph。

1
经过某个步骤后，原始输入丢了。

1
chain = retriever | prompt | model | parser

1
from langchain_core.runnables import RunnablePassthrough
2

3
rag_chain = (
4
    {
5
        "context": retriever | format_docs,
6
        "question": RunnablePassthrough(),
7
    }
8
    | prompt
9
    | model
10
    | StrOutputParser()
11
)

1
输入 question
2
  ↓
3
RunnableParallel:
4
  context = retriever(question) | format_docs
5
  question = 原样透传 question
6
  ↓
7
prompt 接收 {context, question}
8
  ↓
9
model
10
  ↓
11
parser

1
def rag_chain(question):
2
    parallel_output = {
3
        "context": format_docs(retriever.invoke(question)),
4
        "question": question,
5
    }
6

7
    prompt_value = prompt.invoke(parallel_output)
8
    message = model.invoke(prompt_value)
9
    return parser.invoke(message)