Runnable 与 LCEL：把模型应用组织成可组合的数据流

python -m pip install "langchain-core>=1,<2"

from langchain_core.runnables import (
    RunnableBranch,
    RunnableLambda,
    RunnableParallel,
)


clean = RunnableLambda(lambda x: {"text": x["text"].strip()})

features = RunnableParallel(
    text=RunnableLambda(lambda x: x["text"]),
    length=RunnableLambda(lambda x: len(x["text"])),
    urgent=RunnableLambda(lambda x: "立即" in x["text"] or "告警" in x["text"]),
)

route = RunnableBranch(
    (lambda x: x["urgent"], RunnableLambda(lambda x: {**x, "queue": "urgent"})),
    RunnableLambda(lambda x: {**x, "queue": "normal"}),
)

primary = RunnableLambda(
    lambda x: (_ for _ in ()).throw(RuntimeError("primary unavailable"))
)
fallback = RunnableLambda(lambda x: {**x, "summary": x["text"][:20]})

chain = clean | features | route | primary.with_fallbacks([fallback])

print(chain.invoke({"text": "  节点 AX-3 告警，请立即复核  "}))
print(chain.batch(
    [{"text": "设备恢复"}, {"text": "温度告警"}],
    config={"max_concurrency": 2},
))

for chunk in (prompt | model).stream({"event": "节点 AX-3 已恢复"}):
    print(chunk.text, end="", flush=True)