"""演示自建 ``lcel`` 模块中的 Runnable 与管道操作符（竖线 ``|``，LCEL 风格，无第三方 LangChain）。"""
from __future__ import annotations
from lcel import RunnableLambda, RunnableParallel, RunnablePassthrough
[docs]
def demo_linear_pipe() -> None:
"""线性链：上游输出直接作为下游输入。"""
normalize = RunnableLambda(lambda s: s.strip().upper())
decorate = RunnableLambda(lambda s: f"<<{s}>>")
chain = normalize | decorate
print("线性管道:", chain.invoke(" hello lcel "))
[docs]
def demo_dict_between_steps() -> None:
"""各步之间传递 dict，便于携带多个字段。"""
parse = RunnableLambda(lambda raw: {"n": int(raw), "tag": "input"})
scale = RunnableLambda(lambda d: {**d, "n": d["n"] * 10})
render = RunnableLambda(lambda d: f"[{d['tag']}] n={d['n']}")
chain = parse | scale | render
print("字典状态:", chain.invoke("7"))
[docs]
def demo_passthrough_assign() -> None:
"""``RunnablePassthrough.assign``：保留原字段并挂上派生字段。"""
enrich = RunnablePassthrough.assign(length=lambda x: len(x["text"]))
summarize = RunnableLambda(
lambda x: f"{x['text']!r} → length={x['length']}"
)
chain = enrich | summarize
print("assign 管道:", chain.invoke({"text": "LCEL"}))
[docs]
def demo_parallel_branches() -> None:
"""``RunnableParallel``：同一份输入并行走多路，输出 dict 再交给下游。"""
double = RunnableLambda(lambda x: x * 2)
triple = RunnableLambda(lambda x: x * 3)
merge = RunnableLambda(lambda d: d["d"] + d["t"])
chain = RunnableParallel(d=double, t=triple) | merge
print("并行分支:", chain.invoke(5))
[docs]
def run_lcel_demos() -> None:
"""运行上述管道示例。"""
print("=== 自建 LCEL：Runnable + | ===")
demo_linear_pipe()
demo_dict_between_steps()
demo_passthrough_assign()
demo_parallel_branches()
[docs]
def run_all() -> None:
"""入口：运行演示。"""
run_lcel_demos()
if __name__ == "__main__":
run_all()