目录[[#一、LangChain — 或运算RunnableBranch与重写]][[#二、Runnable 接口]][[#三、输出解析器]]一、LangChain — 或运算RunnableBranch与重写或运算即条件分支逻辑根据输入内容动态路由到不同的 Chain类似编程中的if / else。核心概念RunnableBranch按条件选择不同分支执行每个分支是一个(条件函数, Chain)的元组最后一个参数是默认分支所有条件都不满足时执行基本语法fromlangchain_core.runnablesimportRunnableBranch branchRunnableBranch((条件函数1,chain_1),# 条件1成立 → 执行 chain_1(条件函数2,chain_2),# 条件2成立 → 执行 chain_2default_chain# 默认分支)完整示例 — 问题分类路由fromlangchain_core.runnablesimportRunnableBranch,RunnableLambdafromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAI llmChatOpenAI()# 各领域的 Chaintech_chainChatPromptTemplate.from_template(请用技术角度回答{question})|llm general_chainChatPromptTemplate.from_template(请通俗地回答{question})|llm fallback_chainChatPromptTemplate.from_template(请简要回答{question})|llm# 分支路由branchRunnableBranch((lambdax:代码inx[question]or编程inx[question],tech_chain),(lambdax:天气inx[question]or新闻inx[question],general_chain),fallback_chain# 默认)resultbranch.invoke({question:如何用Python写排序算法})# → 命中 tech_chain用 LCEL 管道符重写简化版# 先用一个 Chain 判断类别再路由fromlangchain_core.runnablesimportRunnableLambdadefroute(info):根据分类结果选择对应 Chaincategoryinfo[category]ifcategory技术:returntech_chainelifcategory生活:returngeneral_chainreturnfallback_chain# 组合分类 → 路由 → 执行full_chainclassify_chain|RunnableLambda(route)重写的核心思想将原本写死的if/else逻辑替换为可动态组合的RunnableBranch或RunnableLambda让流程更灵活、更易测试。二、Runnable 接口Runnable是 LangChain 中所有组件的统一接口凡是可以被调用的组件LLM、Prompt、Chain、Parser 等都实现了这个接口。核心方法方法说明适用场景.invoke(input)同步调用返回单个结果普通单次调用.ainvoke(input)异步调用异步程序async/await.stream(input)流式输出逐 token 返回实时打字机效果.astream(input)异步流式输出异步 流式.batch(inputs)批量输入并行处理批量任务.abatch(inputs)异步批量处理大规模异步任务代码示例fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate promptChatPromptTemplate.from_template(用一句话介绍{topic})llmChatOpenAI()chainprompt|llm# ① 同步调用resultchain.invoke({topic:Python})# ② 流式输出逐 token 打印forchunkinchain.stream({topic:Vue.js}):print(chunk.content,end,flushTrue)# ③ 批量处理resultschain.batch([{topic:Python},{topic:JavaScript},{topic:LangChain},])# ④ 异步调用importasyncioasyncdefmain():resultawaitchain.ainvoke({topic:AI})print(result)asyncio.run(main())用管道符|组合 Runnable# 每一个 | 连接的组件都是 Runnablechainprompt|llm|output_parser# 等价于# step1 prompt.invoke(input)# step2 llm.invoke(step1)# step3 output_parser.invoke(step2)RunnableLambda — 将普通函数变成 Runnablefromlangchain_core.runnablesimportRunnableLambdadefdouble(x):returnx*2runnable_doubleRunnableLambda(double)runnable_double.invoke(5)# → 10# 可直接接入管道chainsome_chain|RunnableLambda(lambdax:x.content.upper())RunnablePassthrough — 透传输入fromlangchain_core.runnablesimportRunnablePassthrough# 将原始输入原封不动地传给下一步常用于保留上下文chainRunnablePassthrough()|some_processing_chain三、输出解析器输出解析器Output Parser负责将 LLM 的原始文本输出转换为程序可用的数据格式。① StrOutputParser — 字符串输出解析器将模型返回的AIMessage对象提取为纯字符串。fromlangchain_core.output_parsersimportStrOutputParser chainprompt|llm|StrOutputParser()resultchain.invoke({topic:黑洞})print(result)# 普通字符串而非 AIMessage 对象print(type(result))# class str适用场景只需要文本结果不需要结构化数据时使用。② JsonOutputParser — JSON 输出解析器将模型输出的 JSON 字符串自动解析为 Python 字典/列表。fromlangchain_core.output_parsersimportJsonOutputParserfromlangchain_core.pydantic_v1importBaseModel,Field# 可选用 Pydantic 定义期望的数据结构classMovieInfo(BaseModel):title:strField(description电影名称)year:intField(description上映年份)genre:strField(description电影类型)parserJsonOutputParser(pydantic_objectMovieInfo)# 在 Prompt 中注入格式说明promptChatPromptTemplate.from_messages([(system,你是一个电影专家请按要求的格式输出。),(user,介绍电影《{movie}》\n\n{format_instructions}),])chainprompt|llm|parser resultchain.invoke({movie:星际穿越,format_instructions:parser.get_format_instructions()})print(result)# → {title: 星际穿越, year: 2014, genre: 科幻}print(result[title])# → 星际穿越两者对比|对比项|StrOutputParser|JsonOutputParser||输出类型|str字符串|dict/list||使用场景|纯文本回答|需要结构化数据||需要 Prompt 配合|否|是需注入格式说明||支持流式输出|✅|✅增量解析|