Runnable 与 LCEL 的词汇表

# 1  │ Runnable 约定的四个核心方法（签名示意，非源码）
# 2  │ 关键：四个方法的入参出参形状彼此对齐，只在"同步/异步""单条/多条"上区分
# 3  │
# 4  │ invoke(input)        -> output          # 单条、同步、一次性返回
# 5  │ stream(input)        -> Iterator[chunk] # 单条、同步、逐块吐出
# 6  │ batch([in1, in2])    -> [out1, out2]    # 多条、同步、可并发
# 7  │ ainvoke(input)       -> output          # 单条、异步（await）
# 8  │
# 9  │ 任意组件只要实现了 invoke，框架就能为它兜底另外三个
class MyStep(Runnable):
    def invoke(self, input, config=None):
        return transform(input)          # 你只需定义这一个核心动作

step = MyStep()
step.invoke("x")                         # 直接可用
list(step.stream("x"))                   # stream 也可用 —— 来自接口兜底
step.batch(["x", "y"])                   # batch 也可用 —— 同理

# 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  │
prompt = ChatPromptTemplate.from_template("用一句话解释 {topic}")
model  = init_chat_model("gpt-4o-mini")     # 返回一个 Runnable
parser = StrOutputParser()                  # 也是一个 Runnable

# 8  │ 关键一行：三个竖线，构造一条管道描述
chain = prompt | model | parser
# 10 │ 此刻没有任何网络请求发生 —— chain 只是一个 RunnableSequence 对象
print(type(chain).__name__)                 # RunnableSequence

# 13 │ 直到 invoke，数据才真正从左流到右
answer = chain.invoke({"topic": "向量数据库"})
print(answer)                               # 一句话解释（str）

# 1  │ from langchain_core.runnables import (
# 2  │     RunnablePassthrough, RunnableLambda,
# 3  │     RunnableParallel, RunnableBranch,
# 4  │ )
# 5  │
# 6  │ —— RunnableParallel：对同一输入并行跑两条子链，汇成 dict
parallel = RunnableParallel(
    upper=RunnableLambda(lambda s: s.upper()),   # 子链 A
    length=RunnableLambda(lambda s: len(s)),     # 子链 B
)
parallel.invoke("hi")          # {"upper": "HI", "length": 2}

# 12 │ —— assign：透传原 dict，再追加一个新键 "len"
add_len = RunnablePassthrough.assign(
    length=RunnableLambda(lambda d: len(d["text"]))
)
add_len.invoke({"text": "hi"}) # {"text": "hi", "length": 2}
#                                ↑ 原键还在     ↑ 新键追加

# 19 │ —— Branch：按条件选子链（条件, 子链）...最后一个是 else 兜底
route = RunnableBranch(
    (lambda x: len(x) > 10, RunnableLambda(lambda s: "long")),
    RunnableLambda(lambda s: "short"),           # else
)
route.invoke("hello")          # "short"

# 1  │ chain 是 §1.2 串好的 prompt | model | parser，本身是一个 Runnable
# 2  │ 下面四样，没有一行是在 prompt / model / parser 内部单独实现的
# 3  │
# 4  │ —— 流式：逐块打印，无需链里任何组件"支持流式"的额外代码
for chunk in chain.stream({"topic": "B 树"}):
    print(chunk, end="", flush=True)

# 8  │ —— 批量：一次跑多条输入，框架并发调度
chain.batch([{"topic": "B 树"}, {"topic": "跳表"}])

# 11 │ —— 重试：返回一个包装后的新链，对失败自动退避重试
robust = chain.with_retry(stop_after_attempt=3)

# 14 │ —— fallback：主链失败时自动切到备用链
backup = prompt | init_chat_model("claude-haiku-4") | parser
safe   = chain.with_fallbacks([backup])
safe.invoke({"topic": "B 树"})   # 主链异常时，自动改走 backup

# 1  │ from langchain_core.prompts import ChatPromptTemplate
# 2  │ from langchain_core.output_parsers import StrOutputParser
# 3  │
# 4  │ —— ChatPromptTemplate：把"模板 + 变量"渲染成带角色的消息列表
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是简洁的技术讲师，只用一句话回答"),
    ("human", "解释 {topic}"),
])
msgs = prompt.invoke({"topic": "幂等"})
# 10 │ msgs 是消息列表：[SystemMessage(...), HumanMessage(content="解释 幂等")]

# 12 │ —— 模型输出是一个消息对象（AIMessage），不是裸 str
#       ai_msg = model.invoke(msgs)
#       ai_msg.content        -> "幂等是指..."（真正的文本在 .content 里）
#       ai_msg.usage_metadata -> token 用量等元数据

# 17 │ —— StrOutputParser：把 AIMessage 收成纯字符串，喂给下游
parser = StrOutputParser()
chain  = prompt | model | parser
chain.invoke({"topic": "幂等"})    # -> "幂等是指..."（str，元数据已剥离）