LangChain 之 【Runnable与LCEL】

目录1.简单代码展示2. Runnable 接口2.1 什么是 Runnable2.2 为什么需要 Runnable2.3 代码印证3. LCELLangChain 表达式语言3.1 什么是 LCEL3.2 什么是 RunnableSequence4. 深入管道符三种等价的“组合”方式5. 数据流转的底层逻辑1.简单代码展示1.1申请 API key与配置环境变量这里是远程调用大模型 API所以需要在对应大模型官网上申请 api_key同时为了后续便捷调用可以将 api_key 与 base_url 设置进环境变量中1.2.安装 OpenAI 包与定义⼤模型命令行输入pip install -U langchain-openai安装 LangChain 官方提供的 OpenAI 集成包# 定义⼤模型 from langchain_openai import ChatOpenAI model ChatOpenAI(modeldeepseek-v4-flash)通过修改配置(上述两个环境变量等)DeepSeek API 可以使用与 OpenAI 兼容的 API 格式1.3.定义消息列表# 定义消息列表 from langchain_core.messages import HumanMessage, SystemMessage messages [ SystemMessage(content请翻译成中文), HumanMessage(contentHello world), ]消息列表参数说明SystemMessage表示系统角色消息。系统消息通常作为输入消息序列中的第一条传入是用来启动 AI 行为的消息HumanMessage表示用户角色消息。是来自用户的、从用户传递到模型的消息1.4.调⽤⼤模型model 是 LangChain Runnable 可运⾏接⼝的实例这意味着 model 提供了⼀个标准接⼝供我们与之交互。要简单地调⽤模型我们可以将 消息列表 传递给 .invoke ⽅法。使⽤ .invoke ⽅法进⾏⼤模型调⽤核⼼代码response model.invoke(messages) print(response)输出结果调试可以看⻅ response 类型为 AIMessageAIMessage(content你好世界, additional_kwargs{refusal: None}, response_metadata{token_usage: {completion_tokens: 54, prompt_tokens: 10, total_tokens: 64, completion_tokens_details: {accepted_prediction_tokens: None, audio_tokens: None, reasoning_tokens: 50, rejected_prediction_tokens: None}, prompt_tokens_details: {audio_tokens: None, cached_tokens: 0}, prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 10}, model_name: deepseek-v4-flash, system_fingerprint: fp_8b330d02d0_prod0820_fp8_kvcache_20260402, id: 91d952ef-75ff-4975-acc0-2def7d7a5d11, service_tier: None, finish_reason: stop, logprobs: None}, idrun--019f2f8b-04ef-7a01-9794-0406ca490dd6-0, usage_metadata{input_tokens: 10, output_tokens: 54, total_tokens: 64, input_token_details: {cache_read: 0}, output_token_details: {reasoning: 50}})输出说明AIMessage来自 AI 的消息。从聊天模型返回作为对提示输入的响应。content消息的内容。additional_kwargs与消息关联的其他有效负载数据。对于来自 AI 的消息可能包括模型提供程序编码的工具调用。response_metadata响应元数据。例如响应标头、logprobs、令牌计数、模型名称。侧重于响应本身的信息比如这次请求的 ID、使用的模型版本、以及服务提供商返回的所有原始元数据。它主要用于调试、日志记录和获取请求的上下文信息。usage_metadata消息的使用元数据例如令牌计数。侧重于资源消耗的量化信息即这次请求消耗了多少 Token。它主要用于成本计算、监控和预算控制。1.5.输出解析若只想输出聊天模型返回的结果字符串可以使⽤StrOutputParser输出解析器组件将⼤模型输出结果解析为最可能的字符串。核⼼代码# 定义str字符串输出解析器 from langchain_core.output_parsers import StrOutputParser parser StrOutputParser() print(parser.invoke(result))1.6.链式执行# 定义链 chain model | parser # 执⾏链 response chain.invoke(messages) print(response)2. Runnable 接口2.1 什么是 Runnable官方定义它为使用 LangChain 组件的基础你可以把它想象成我们生活中的USB-C 接口。无论是大模型、解析器、检索器还是复杂的 Agent只要实现了 Runnable 接口它们就拥有了统一的“插拔”标准2.2 为什么需要 Runnable在没有统一标准前调用模型用model.chat()调用解析器用parser.parse()调用检索器用retriever.get()……方法名五花八门Runnable 接口强制规定了一套标准化的方法论。任何一个实现了 Runnable 的组件都天然具备以下四种核心能力能力对应方法作用解释Invoked调用.invoke()最常用。输入一个东西立刻得到一个输出同步执行。Batched批处理.batch()高效处理一批输入比用for循环快得多支持并行。Streamed流式传输.stream()实现 AI 打字机效果数据边生成边输出提升用户体验。Composed组合配合|使用这是 LCEL 的基础允许将自身与另一个 Runnable 串联。2.3 代码印证from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # ChatOpenAI 是 Runnable 接口的实例 model ChatOpenAI(modelgpt-4o-mini) result model.invoke(messages) # 调用 .invoke() # StrOutputParser 也是 Runnable 接口的实例 parser StrOutputParser() parser.invoke(result) # 同样调用 .invoke()3. LCELLangChain 表达式语言如果说 Runnable 是“积木”那么LCEL 就是把这些积木拼在一起的“说明书”和“胶水”3.1 什么是 LCELLCEL 采用声明式的方法从现有的 Runnable 对象构建出新的 Runnable 对象。它不改变组件内部的逻辑只负责定义数据流动的方向它的核心语法极其简单使用|管道运算符3.2 什么是 RunnableSequence# model 和 parser 都是 Runnable chain model | parser # 检查 chain 的类型class langchain_core.runnables.base.RunnableSequence print(type(chain)) # 因为 RunnableSequence 也是 Runnable所以它也有 .invoke() result chain.invoke(messages)当我们通过 LCEL 将两个或更多的 Runnable 串联起来时LangChain 会在底层生成一个新的对象 ——RunnableSequence可运行序列。关键点来了RunnableSequence本身也完整地实现了 Runnable 接口这是一个典型的“组合模式”应用。这意味着你拼出来的这条“链”Chain和你之前用的单个model拥有完全一样的调用姿势也有.invoke()、.batch()等方法。你可以把拼好的链再当作一个零件去拼进更大的链里实现无限嵌套4. 深入管道符三种等价的“组合”方式三种将组件串联的方法在逻辑上完全等价只是编码风格不同序号写法方式代码示例适用场景1管道符最常用chain model | parser日常开发首选Python 重载了|运算符简洁直观。2显式实例化chain RunnableSequence(firstmodel, lastparser)底层实现原理通常用于框架内部或需要动态构建链时。3.pipe() 方法chain model.pipe(parser)函数式编程风格与管道符完全同义在某些 IDE 中自动补全更友好。5. 数据流转的底层逻辑chain model | parser当你在 LCEL 中执行 chain.invoke(messages) 时输入messages消息列表传入 chain。第一步LCEL 将 messages 传给 model大模型执行 model.invoke(messages)。中间产出model 返回一个 AIMessage 对象包含内容、元数据、Token 用量等。第二步LCEL 自动将上一步的产出AIMessage作为参数传给 parser执行 parser.invoke(AIMessage)。最终输出parser 提取出 AIMessage 中的 content 字段返回纯字符串 你好。核心公式上一个 Runnable 的输出类型 下一个 Runnable 的输入类型