本文将带你了解LangChain生态体系包括环境配置、核心组件关系、Agent构建方法、LCEL表达式语言以及LangSmith监控工具。通过本文你将掌握LangChain的基本使用方法为进一步深入学习大模型开发打下坚实基础。 前置准备环境配置在开始学习之前请确保完成以下环境配置1. Python 版本python --version # 需要 Python 3.10 或更高版本2. 安装依赖# 使用 pip 安装最新版本 pip install langchain langchain-openai langgraph langchain-community # 或使用 uv (推荐) uv pip install langchain langchain-openai langgraph langchain-community # 如需指定版本推荐使用1.0.7或更高版本 pip install langchain1.0.7 langchain-openai1.0.3 langgraph1.0.33. 环境变量配置# 创建 .env 文件 OPENAI_API_KEYsk-your-api-key-here LANGSMITH_API_KEYyour-langsmith-key # 可选,用于监控 LANGSMITH_TRACINGtrue # 可选 # 在代码中加载 from dotenv import load_dotenv import os load_dotenv() # 验证环境变量 required_vars [OPENAI_API_KEY] for var in required_vars: if not os.getenv(var): raise EnvironmentError(f缺少必需的环境变量: {var})4. 依赖版本清单# pyproject.toml 推荐配置 [tool.poetry.dependencies] python ^3.10 langchain ^1.0.7 langchain-openai ^1.0.3 langgraph ^1.0.3 langchain-community ^0.3.0 langchain-core ^1.0.7 langsmith ^0.4.43 python-dotenv ^1.0.0 # requirements.txt 格式 # langchain1.0.7 # langchain-openai1.0.3 # langgraph1.0.3 # langchain-community0.3.0 # langchain-core1.0.7 # langsmith0.4.43 # python-dotenv1.0.0前置知识建议具备以下基础知识✅ Python 基础 (async/await、类型注解、装饰器)✅ LLM 基本概念 (Prompt、Token、Temperature等)✅ API 调用基础✅ JSON 数据格式第1章LangChain 生态全景1.1 架构层次关系LangChain 生态系统目前已形成“多层协同”的架构体系既可支持快速原型开发也可支撑生产级 LLM 应用。整体结构如下层级核心组件职责定位典型场景应用层Deep Agents / LangGraph Projects复杂自治 Agent、长期运行、多 Agent 协作智能助手、自动化任务系统编排层LangGraph状态化流程控制、节点执行、分支循环多 Agent 编排、可视化状态流链路层LangChain / LCEL模型调用、提示管理、工具集成RAG、问答、对话监控层LangSmith调试、观测、评估、成本追踪DevOps、Evals、质量监控1.1.1 LangChain 与 LangGraph 的关系LangChain 专注于 链式逻辑与 Agent 封装LangGraph 专注于 流程编排与状态管理。LangChain用于构建单条或线性 chainPrompt→Model→Tool→Output。LangGraph用于管理含分支、循环、并发的复杂流程可视化、持久化状态。二者可并用LangGraph 中的节点可运行 LangChain 或 LCEL 构造的 chain。图 1-2 LangChain 与 LangGraph 协作关系图1.1.2 如何构建 AgentLangChain 1.0 提供统一的 Agent 构建接口create_agent快速开始创建你的第一个 Agentfrom langchain.agents import create_agent from langchain_openai import ChatOpenAI from langchain_core.tools import tool # 步骤1: 定义工具 tool def get_weather(city: str) - str: 获取指定城市的天气 return f{city}今天天气晴朗温度25°C tool def calculate(expression: str) - str: 计算数学表达式 try: result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)} # 步骤2: 创建 Agent agent create_agent( modelChatOpenAI(modelgpt-4), tools[get_weather, calculate], system_prompt你是一个有帮助的助手可以查询天气和进行计算。 ) # 步骤3: 运行 Agent result agent.invoke({ messages: [(user, 北京天气如何另外帮我算一下 25 * 4)] }) # 查看结果 print(result[messages][-1].content)输出示例北京今天天气晴朗温度25°C。 25 * 4 的计算结果是 100。核心概念create_agent的工作原理关键参数说明参数类型必需说明modelChatModelstr✅toolsList[Tool]✅可用的工具列表system_promptstr❌系统提示词定义 Agent 行为checkpointerCheckpointer❌状态持久化用于多轮对话interrupt_beforeList[str]❌在指定节点前暂停需要人工确认interrupt_afterList[str]❌在指定节点后暂停完整工作流程模型绑定指定使用的 LLM如 GPT-4、Claude 等工具注册提供 Agent 可调用的工具集合提示配置通过system_prompt定义 Agent 的角色和行为决策执行LLM 基于 ReAct 模式自动决定是否调用工具结果返回自动组合工具输出和 LLM 回复监控追踪集成 LangSmith 实现全链路追踪关键特性✅ 官方推荐LangChain 1.0 标准 API✅ 简洁易用统一的接口3步即可创建 Agent✅ 完整功能支持 middleware、cache、checkpointer✅ 自动工具调用LLM 自动判断何时使用哪个工具✅ 多轮对话支持状态持久化实现上下文记忆✅ 长期支持官方维护持续更新1.1.3 LCEL 的定位与作用LCELLangChain Expression Language是 LangChain 的“声明式组合语法”用于 构建可并行、可流式、可追踪的 Runnable 链。核心概念RunnableSequence顺序执行RunnableParallel并行执行支持 async / stream / batch 统一调用可直接嵌入 LangGraph 节点。价值在代码层面构建“数据流管线”如同 Node-RED 或 Airflow 的轻量化实现。1.1.4 LangSmith 的监控职责LangSmith 是 LangChain 官方推出的可观测性与质量评估平台。主要职责 Tracing 追踪 Chain/Graph/Agent 每个调用节点。 Metrics 监控延迟、Token 用量、错误率、成本。 Evaluation 对模型或 Agent 输出进行打分与对比。⚙️ Integration 与 LangChain 、LangGraph 、Deep Agents 原生集成。1.2 核心设计理念1.2.1 Provider-Agnostic 设计LangChain 通过统一接口屏蔽 LLM 提供商差异OpenAI、Anthropic、Cohere、Azure 等以 “Provider 无关” 的方式构建应用。模型切换无需修改上层逻辑。支持跨平台成本追踪与性能比较。1.2.2 Runnable Protocol 统一抽象Runnable 是 LangChain 的核心执行协议一切皆 Runnable。包括 Chain、Agent、Tool、Prompt 均实现该接口。统一执行入口invoke()、ainvoke()、stream()。支持异步、批量、流式、可追踪调用。所有 Runnable 可嵌套、组合、装饰。1.2.3 Middleware-Driven 架构LangChain 支持 Callback / Hook / Tracing 机制可在执行流中插入中间件。常见中间件用途Token 计数与成本监控日志与错误追踪安全审查与访问控制重试与超时控制1.2.4 Production-First 理念LangChain 1.0 及 LangGraph 1.0 发布后生态全面转向 生产级稳定性与可观测性。核心目标包括长期兼容向 2.0 平滑过渡成本可控LangSmith 监控 自动计费模型热替换Provider-agnostic完整 CI/CD 与 Evals 集成1.3 技术选型决策树1.3.1 何时使用 create_agent适用场景单 Agent 执行流程线性需要快速实现 Tool 调用用于 RAG 、问答、助手类场景。1.3.2 何时深入 LangGraph适用场景多 Agent 协作存在分支 / 循环 / 状态管理需可视化、可调试、持久化运行。1.3.3 何时使用 Deep Agents适用场景长期运行、自主决策 Agent复杂任务拆解、子 Agent 管理持续任务执行与周期性触发。1.3.4 何时需要 Middleware适用场景生产环境运行需要日志、指标、安全控制、回调。推荐所有 Chain/Agent 均启用 LangSmith Tracing 自定义 Callback。1.3.5 典型应用场景分析场景推荐技术理由A. 企业文档问答create_agent LCEL快速构建 RAG 问答B. 智能客服系统LangChain Agent Middleware需多轮对话与监控C. 自动化任务管理LangGraph Deep Agents LangSmith复杂 workflow 自治 agentD. 内容摘要或转换LCEL轻量、高并行、可流式本章小结LangChain 生态体系可概括为链式逻辑LangChain → 图式编排LangGraph → 监控评估LangSmith → 自治进化Deep Agents核心理念Provider-AgnosticRunnable 统一抽象Middleware 可插架构Production-First 部署思维设计哲学上从“玩具原型”走向“生产可观测”的工程系统。思考与练习练习 1选择一个业务场景画出其 LangChain 技术选型决策路径。练习 2编写一个 LCEL 例程Prompt → Model → Parser → Tool并标注你会插入哪些 Middleware。练习 3设计一个长期运行 Agent 如 市场监控或自动报告说明如何用 LangGraph LangSmith 实现监控与 Evals。思考题LCEL 在 LangGraph 节点中嵌套使用会带来哪些优势与代价第2章核心抽象Runnable 与 LCEL2.1 Runnable Protocol2.1.1 为什么需要统一抽象在 LangChain 1.0 之前不同组件Prompt、Model、Tool、Chain的调用方式各不相同导致接口不一致学习成本高难以组合缺乏标准化无法统一追踪、监控组合困难不同组件难以嵌套使用Runnable Protocol 解决方案LangChain 1.0 引入 Runnable 作为统一执行协议所有组件均实现该接口from langchain_core.runnables import Runnable # 所有组件均实现 Runnable 接口 class Runnable: def invoke(self, input, configNone): ... # 同步调用 def ainvoke(self, input, configNone): ... # 异步调用 def stream(self, input, configNone): ... # 流式输出 def astream(self, input, configNone): ... # 异步流式 def batch(self, inputs, configNone): ... # 批量处理核心优势实际应用示例from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 所有组件都是 Runnable prompt ChatPromptTemplate.from_template(Tell me a joke about {topic}) model ChatOpenAI() parser StrOutputParser() # 统一的调用方式 result prompt.invoke({topic: AI}) result model.invoke(Tell me a joke) result parser.invoke(some text)2.1.2 核心方法invoke、stream、batchinvoke() - 同步调用最基础的调用方式适用于单次请求from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4) # 同步调用 response model.invoke(What is LangChain?) print(response.content)执行流程stream() - 流式输出适用于需要实时反馈的场景如聊天界面# 流式输出 for chunk in model.stream(Tell me a long story): print(chunk.content, end, flushTrue)流式输出的优势✅ 降低首字延迟TTFT - Time To First Token✅ 提升用户体验实时显示✅ 减少超时风险batch() - 批量处理适用于批量请求场景自动优化并发# 批量处理自动并发优化 inputs [ What is AI?, What is ML?, What is LLM? ] results model.batch(inputs) for result in results: print(result.content)批量处理的优势✅ 自动并发控制✅ 成本追踪聚合✅ 错误处理优化abatch() - 异步批量处理在需要高并发处理大量请求时abatch()比同步batch()性能更好import asyncio from langchain_openai import ChatOpenAI model ChatOpenAI() async def async_batch_example(): inputs [ What is AI?, What is ML?, What is LLM?, What is NLP?, What is DL? ] # 异步批量处理 results await model.abatch(inputs) for i, result in enumerate(results): print(fResult {i1}: {result.content}) # 运行异步任务 asyncio.run(async_batch_example())abatch 与 batch 的对比方法适用场景优势batch()中小批量50实现简单无需async/awaitabatch()大批量50、I/O密集更高并发性能资源利用率高2.1.3 异步方法ainvoke、astream在高并发场景下异步方法可显著提升性能ainvoke() - 异步调用import asyncio from langchain_openai import ChatOpenAI model ChatOpenAI() async def main(): # 异步调用 response await model.ainvoke(What is async programming?) print(response.content) asyncio.run(main())astream() - 异步流式async def stream_example(): async for chunk in model.astream(Tell me a story): print(chunk.content, end, flushTrue) asyncio.run(stream_example())并发性能对比# ❌ 同步方式串行执行慢 def sync_batch(): results [] for query in queries: results.append(model.invoke(query)) return results # ✅ 异步方式并发执行快 async def async_batch(): tasks [model.ainvoke(query) for query in queries] return await asyncio.gather(*tasks)性能对比请求数同步耗时异步耗时性能提升1030s5s6x50150s15s10x100300s25s12x2.1.4 Runnable 类型Lambda、Parallel、Branch、FallbacksRunnableLambda - 自定义函数包装将普通 Python 函数包装为 Runnablefrom langchain_core.runnables import RunnableLambda def uppercase(text: str) - str: return text.upper() # 包装为 Runnable runnable_upper RunnableLambda(uppercase) # 统一调用方式 result runnable_upper.invoke(hello) # HELLORunnableParallel - 并行执行同时执行多个 Runnable结果以字典形式返回from langchain_core.runnables import RunnableParallel parallel RunnableParallel( jokeChatPromptTemplate.from_template(Tell a joke about {topic}) | model, poemChatPromptTemplate.from_template(Write a poem about {topic}) | model ) # 并行执行 result parallel.invoke({topic: AI}) print(result[joke]) print(result[poem])RunnableBranch - 条件分支根据条件选择不同的执行路径from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: len(x) 100, long_text_handler), (lambda x: len(x) 10, medium_text_handler), short_text_handler # 默认分支 ) result branch.invoke(some text)with_fallbacks() - 降级处理主 Runnable 失败时自动切换到备用方案from langchain_openai import ChatOpenAI primary_model ChatOpenAI(modelgpt-4) fallback_model ChatOpenAI(modelgpt-3.5-turbo) # 直接使用 with_fallbacks 方法无需导入额外类 model_with_fallback primary_model.with_fallbacks([fallback_model]) # 如果 GPT-4 失败自动使用 GPT-3.5 result model_with_fallback.invoke(Hello)参数说明 (基于官方API文档验证)必需参数fallbacks:Sequence[Runnable]- 备用 Runnable 序列按顺序尝试可选参数 (仅关键字参数)exceptions_to_handle:Tuple[Type[BaseException], ...]- 需要处理的异常类型元组默认为(Exception,)exception_key:Optional[str]- 可选的键名用于将异常信息传递给备用方案。如为None(默认)异常不传递给备用方案完整参数示例# ✅ 示例1: 只对特定异常类型执行降级 model_with_fallback primary_model.with_fallbacks( fallbacks[fallback_model], # ✅ 官方标准参数fallbacks (复数列表) exceptions_to_handle(TimeoutError, ConnectionError), # ✅ 官方标准参数 ) # ✅ 示例2: 将异常信息传递给备用方案 from langchain_core.runnables import RunnableLambda def handle_with_error_context(inputs): 备用方案可以访问异常信息 if error in inputs: print(fOriginal error: {inputs[error]}) return fallback_model.invoke(inputs[input]) model_with_error_context primary_model.with_fallbacks( fallbacks[RunnableLambda(handle_with_error_context)], # ✅ 使用 fallbacks 参数名 exception_keyerror # ✅ 官方标准参数异常会以 error 键传递 ) # ⚠️ 重要使用 exception_key 时主 Runnable 和所有备用方案都必须接受字典输入 result model_with_error_context.invoke({input: Hello})API 规范总结def with_fallbacks( self, fallbacks: Sequence[Runnable[Input, Output]], # 必需 *, exceptions_to_handle: Tuple[Type[BaseException], ...] (Exception,), # 可选 exception_key: Optional[str] None # 可选 ) - RunnableWithFallbacksT[Input, Output]: ...2.2 LCEL 表达式语言2.2.1 声明式组合理念LCELLangChain Expression Language是一种声明式语法用于组合 Runnable 对象。命令式 vs 声明式# ❌ 命令式手动控制流程 def imperative_chain(input): step1_result prompt.invoke(input) step2_result model.invoke(step1_result) step3_result parser.invoke(step2_result) return step3_result # ✅ 声明式LCEL 管道 chain prompt | model | parser result chain.invoke(input)LCEL 的核心优势2.2.2 管道操作符|与并行{}管道操作符|- 顺序执行将多个 Runnable 串联成管道from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 管道组合 chain ( ChatPromptTemplate.from_template(Tell me about {topic}) | ChatOpenAI() | StrOutputParser() ) # 自动按顺序执行 result chain.invoke({topic: LangChain})执行流程并行字典{}- 并行执行使用字典语法实现并行执行from langchain_core.runnables import RunnablePassthrough chain { context: retriever | format_docs, question: RunnablePassthrough() } | prompt | model # context 和 question 并行处理 result chain.invoke(What is LangChain?)执行流程assign() - 状态更新快捷方式RunnablePassthrough.assign()是 LCEL 中最常用的操作之一用于在链中添加或更新字段from langchain_core.runnables import RunnablePassthrough from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 创建向量检索器 vectorstore Chroma.from_texts( [LangChain是一个AI应用框架, 它支持RAG和Agent], embeddingOpenAIEmbeddings() ) retriever vectorstore.as_retriever() # 使用 assign() 添加检索上下文 chain ( RunnablePassthrough.assign( contextretriever # 添加 context 字段 ) | ChatPromptTemplate.from_template( 基于以下上下文回答问题:/n{context}/n/n问题: {question} ) | ChatOpenAI() | StrOutputParser() ) # 输入只需要 questioncontext 会自动添加 result chain.invoke({question: 什么是LangChain?}) # 内部流程: {question: ...} - {question: ..., context: [...]}assign() 的优势保留原始输入不覆盖已有字段简化代码避免手动构造字典链式组合可以多次调用# 多次 assign 叠加字段 chain ( RunnablePassthrough.assign( contextretriever # 添加检索结果 ) .assign( context_countlambda x: len(x[context]) # 添加统计信息 ) .assign( timestamplambda x: 2025-11-17 # 添加时间戳 ) | prompt | model ) # 输入: {question: ...} # 第一步后: {question: ..., context: [...]} # 第二步后: {question: ..., context: [...], context_count: 3} # 第三步后: {question: ..., context: [...], context_count: 3, timestamp: ...}常见使用场景# 场景1: RAG 添加检索上下文 rag_chain ( RunnablePassthrough.assign(contextretriever) | rag_prompt | model ) # 场景2: 添加多个数据源 multi_source_chain ( RunnablePassthrough.assign( docsdoc_retriever, historyhistory_retriever, metadatametadata_fetcher ) | prompt | model ) # 场景3: 数据转换 transform_chain ( RunnablePassthrough.assign( upper_textlambda x: x[text].upper(), word_countlambda x: len(x[text].split()) ) | processor )2.2.3 组合模式顺序、并行、条件、循环顺序链接# 简单顺序 chain step1 | step2 | step3 # 复杂顺序 chain ( {input: RunnablePassthrough()} | prompt | model | {output: parser, raw: RunnablePassthrough()} )并行执行# 并行获取多个信息 chain RunnableParallel( summarysummarize_chain, keywordsextract_keywords_chain, sentimentsentiment_chain )条件分支from langchain_core.runnables import RunnableBranch # 根据输入长度选择不同处理 chain RunnableBranch( (lambda x: len(x[text]) 1000, long_text_chain), (lambda x: len(x[text]) 100, medium_text_chain), short_text_chain )循环迭代# 使用 RunnableLambda 实现循环 def iterative_refine(input): result input for _ in range(3): result refine_chain.invoke(result) return result chain RunnableLambda(iterative_refine)2.3 高级特性2.3.1 Fallback 降级与 Retry 重试Fallback - 自动降级# 多级降级 chain ( primary_model .with_fallbacks(fallbacks[backup_model_1, backup_model_2]) # ✅ 使用 fallbacks 参数名 ) # 只对特定异常执行降级 chain ( primary_model .with_fallbacks( fallbacks[backup_model_1, backup_model_2], # ✅ 使用 fallbacks 参数名 exceptions_to_handle(TimeoutError, ConnectionError) # ✅ 官方标准参数 ) )降级流程Retry - 自动重试# 直接使用 with_retry 方法无需单独导入 chain ( prompt | model | parser ).with_retry( stop_after_attempt3, # 最大重试次数 wait_exponential_jitterTrue, # 指数退避 随机抖动 retry_if_exception_type(Exception,) # 指定需要重试的异常类型 )参数说明基于官方API文档验证stop_after_attempt最大重试次数默认为 3wait_exponential_jitter是否使用指数退避 随机抖动默认为 Trueretry_if_exception_type需要重试的异常类型元组默认为(Exception,)重试策略示例from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 只对特定异常重试 chain (prompt | model | parser).with_retry( retry_if_exception_type(TimeoutError, ConnectionError), stop_after_attempt5, wait_exponential_jitterTrue ) # 禁用指数退避立即重试 chain (prompt | model | parser).with_retry( stop_after_attempt3, wait_exponential_jitterFalse # 禁用指数退避立即重试 )重试行为指数退避1s → 2s → 4s → 8s最大重试次数可自定义默认3次重试条件可指定异常类型默认所有 Exception2.3.2 Timeout 超时控制重要:RunnableConfig不支持timeout参数。超时控制应在模型层面配置。from langchain_openai import ChatOpenAI # ✅ 正确在模型构造时设置timeout model ChatOpenAI( modelgpt-4, timeout30, # 30秒超时 max_retries2 ) chain prompt | model | parser result chain.invoke(input)超时 降级组合策略from langchain_openai import ChatOpenAI # 主模型严格超时 slow_model ChatOpenAI(modelgpt-4, timeout10) # 降级模型快速响应 fast_model ChatOpenAI(modelgpt-3.5-turbo, timeout5) # 组合超时 降级 chain (prompt | slow_model | parser).with_fallbacks([ prompt | fast_model | parser ])使用 RunnableConfig 配置其他参数from langchain_core.runnables import RunnableConfig # RunnableConfig支持的参数 result chain.invoke( input, configRunnableConfig( max_concurrency5, # 最大并发数 tags[production], # 标签用于监控 metadata{user: alice} # 元数据 ) )2.3.3 缓存与性能优化 提示: 本节介绍 Runnable Protocol 的基础性能API。生产环境的深度性能调优、成本控制、缓存架构等内容详见 第八篇《生产实践》第21章。LLM 缓存from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache # 启用缓存 set_llm_cache(InMemoryCache()) # 相同请求直接返回缓存结果 model.invoke(What is AI?) # 调用 LLM model.invoke(What is AI?) # 返回缓存不调用 LLM批处理优化# 批处理优化使用max_concurrency控制并发 chain prompt | model.with_config({max_concurrency: 10}) # 内部自动合并请求 results chain.batch(inputs)流式优化# 流式传输减少延迟 for chunk in chain.stream(input): print(chunk, end)性能对比特性普通调用优化后性能提升缓存2s50ms40x批处理10s2s5x流式TTFT 2sTTFT 200ms10x本章小结Runnable Protocol 核心要点✅ 统一接口invoke、stream、batch、ainvoke、astream✅ 可组合性Lambda、Parallel、Branch、Fallbacks✅ 可追踪性自动集成 LangSmith✅ 性能优化异步、批处理、缓存LCEL 核心要点✅ 声明式组合|管道、{}并行✅ 自动优化并行执行、流式传输✅ 高级特性Fallback、Retry、Timeout、Cache设计哲学一切皆 Runnable所有组件统一接口声明式组合自动优化执行。思考与练习练习 1基础管道构建一个 LCEL 管道Prompt → Model → Parser实现一个简单的问答系统。练习 2并行处理使用 RunnableParallel 同时生成一个笑话、一首诗和一个故事输入主题为AI。练习 3错误处理实现一个带有 Fallback 和 Retry 的 chain主模型失败时自动切换到备用模型。练习 4性能优化对比同步批处理和异步批处理的性能差异10个请求。思考题什么场景下应该使用 stream 而不是 invokeRunnableBranch 和简单的 if-else 有什么区别如何在 LCEL 中实现循环逻辑如何学习大模型 AI 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取 
长河君 · 2026/10/29 2:05:37 · 阅读 1.2k 小白程序员必备：轻松入门大模型开发，收藏这份LangChain学习指南！
本文将带你了解LangChain生态体系包括环境配置、核心组件关系、Agent构建方法、LCEL表达式语言以及LangSmith监控工具。通过本文你将掌握LangChain的基本使用方法为进一步深入学习大模型开发打下坚实基础。 前置准备环境配置在开始学习之前请确保完成以下环境配置1. Python 版本python --version # 需要 Python 3.10 或更高版本2. 安装依赖# 使用 pip 安装最新版本 pip install langchain langchain-openai langgraph langchain-community # 或使用 uv (推荐) uv pip install langchain langchain-openai langgraph langchain-community # 如需指定版本推荐使用1.0.7或更高版本 pip install langchain1.0.7 langchain-openai1.0.3 langgraph1.0.33. 环境变量配置# 创建 .env 文件 OPENAI_API_KEYsk-your-api-key-here LANGSMITH_API_KEYyour-langsmith-key # 可选,用于监控 LANGSMITH_TRACINGtrue # 可选 # 在代码中加载 from dotenv import load_dotenv import os load_dotenv() # 验证环境变量 required_vars [OPENAI_API_KEY] for var in required_vars: if not os.getenv(var): raise EnvironmentError(f缺少必需的环境变量: {var})4. 依赖版本清单# pyproject.toml 推荐配置 [tool.poetry.dependencies] python ^3.10 langchain ^1.0.7 langchain-openai ^1.0.3 langgraph ^1.0.3 langchain-community ^0.3.0 langchain-core ^1.0.7 langsmith ^0.4.43 python-dotenv ^1.0.0 # requirements.txt 格式 # langchain1.0.7 # langchain-openai1.0.3 # langgraph1.0.3 # langchain-community0.3.0 # langchain-core1.0.7 # langsmith0.4.43 # python-dotenv1.0.0前置知识建议具备以下基础知识✅ Python 基础 (async/await、类型注解、装饰器)✅ LLM 基本概念 (Prompt、Token、Temperature等)✅ API 调用基础✅ JSON 数据格式第1章LangChain 生态全景1.1 架构层次关系LangChain 生态系统目前已形成“多层协同”的架构体系既可支持快速原型开发也可支撑生产级 LLM 应用。整体结构如下层级核心组件职责定位典型场景应用层Deep Agents / LangGraph Projects复杂自治 Agent、长期运行、多 Agent 协作智能助手、自动化任务系统编排层LangGraph状态化流程控制、节点执行、分支循环多 Agent 编排、可视化状态流链路层LangChain / LCEL模型调用、提示管理、工具集成RAG、问答、对话监控层LangSmith调试、观测、评估、成本追踪DevOps、Evals、质量监控1.1.1 LangChain 与 LangGraph 的关系LangChain 专注于 链式逻辑与 Agent 封装LangGraph 专注于 流程编排与状态管理。LangChain用于构建单条或线性 chainPrompt→Model→Tool→Output。LangGraph用于管理含分支、循环、并发的复杂流程可视化、持久化状态。二者可并用LangGraph 中的节点可运行 LangChain 或 LCEL 构造的 chain。图 1-2 LangChain 与 LangGraph 协作关系图1.1.2 如何构建 AgentLangChain 1.0 提供统一的 Agent 构建接口create_agent快速开始创建你的第一个 Agentfrom langchain.agents import create_agent from langchain_openai import ChatOpenAI from langchain_core.tools import tool # 步骤1: 定义工具 tool def get_weather(city: str) - str: 获取指定城市的天气 return f{city}今天天气晴朗温度25°C tool def calculate(expression: str) - str: 计算数学表达式 try: result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)} # 步骤2: 创建 Agent agent create_agent( modelChatOpenAI(modelgpt-4), tools[get_weather, calculate], system_prompt你是一个有帮助的助手可以查询天气和进行计算。 ) # 步骤3: 运行 Agent result agent.invoke({ messages: [(user, 北京天气如何另外帮我算一下 25 * 4)] }) # 查看结果 print(result[messages][-1].content)输出示例北京今天天气晴朗温度25°C。 25 * 4 的计算结果是 100。核心概念create_agent的工作原理关键参数说明参数类型必需说明modelChatModelstr✅toolsList[Tool]✅可用的工具列表system_promptstr❌系统提示词定义 Agent 行为checkpointerCheckpointer❌状态持久化用于多轮对话interrupt_beforeList[str]❌在指定节点前暂停需要人工确认interrupt_afterList[str]❌在指定节点后暂停完整工作流程模型绑定指定使用的 LLM如 GPT-4、Claude 等工具注册提供 Agent 可调用的工具集合提示配置通过system_prompt定义 Agent 的角色和行为决策执行LLM 基于 ReAct 模式自动决定是否调用工具结果返回自动组合工具输出和 LLM 回复监控追踪集成 LangSmith 实现全链路追踪关键特性✅ 官方推荐LangChain 1.0 标准 API✅ 简洁易用统一的接口3步即可创建 Agent✅ 完整功能支持 middleware、cache、checkpointer✅ 自动工具调用LLM 自动判断何时使用哪个工具✅ 多轮对话支持状态持久化实现上下文记忆✅ 长期支持官方维护持续更新1.1.3 LCEL 的定位与作用LCELLangChain Expression Language是 LangChain 的“声明式组合语法”用于 构建可并行、可流式、可追踪的 Runnable 链。核心概念RunnableSequence顺序执行RunnableParallel并行执行支持 async / stream / batch 统一调用可直接嵌入 LangGraph 节点。价值在代码层面构建“数据流管线”如同 Node-RED 或 Airflow 的轻量化实现。1.1.4 LangSmith 的监控职责LangSmith 是 LangChain 官方推出的可观测性与质量评估平台。主要职责 Tracing 追踪 Chain/Graph/Agent 每个调用节点。 Metrics 监控延迟、Token 用量、错误率、成本。 Evaluation 对模型或 Agent 输出进行打分与对比。⚙️ Integration 与 LangChain 、LangGraph 、Deep Agents 原生集成。1.2 核心设计理念1.2.1 Provider-Agnostic 设计LangChain 通过统一接口屏蔽 LLM 提供商差异OpenAI、Anthropic、Cohere、Azure 等以 “Provider 无关” 的方式构建应用。模型切换无需修改上层逻辑。支持跨平台成本追踪与性能比较。1.2.2 Runnable Protocol 统一抽象Runnable 是 LangChain 的核心执行协议一切皆 Runnable。包括 Chain、Agent、Tool、Prompt 均实现该接口。统一执行入口invoke()、ainvoke()、stream()。支持异步、批量、流式、可追踪调用。所有 Runnable 可嵌套、组合、装饰。1.2.3 Middleware-Driven 架构LangChain 支持 Callback / Hook / Tracing 机制可在执行流中插入中间件。常见中间件用途Token 计数与成本监控日志与错误追踪安全审查与访问控制重试与超时控制1.2.4 Production-First 理念LangChain 1.0 及 LangGraph 1.0 发布后生态全面转向 生产级稳定性与可观测性。核心目标包括长期兼容向 2.0 平滑过渡成本可控LangSmith 监控 自动计费模型热替换Provider-agnostic完整 CI/CD 与 Evals 集成1.3 技术选型决策树1.3.1 何时使用 create_agent适用场景单 Agent 执行流程线性需要快速实现 Tool 调用用于 RAG 、问答、助手类场景。1.3.2 何时深入 LangGraph适用场景多 Agent 协作存在分支 / 循环 / 状态管理需可视化、可调试、持久化运行。1.3.3 何时使用 Deep Agents适用场景长期运行、自主决策 Agent复杂任务拆解、子 Agent 管理持续任务执行与周期性触发。1.3.4 何时需要 Middleware适用场景生产环境运行需要日志、指标、安全控制、回调。推荐所有 Chain/Agent 均启用 LangSmith Tracing 自定义 Callback。1.3.5 典型应用场景分析场景推荐技术理由A. 企业文档问答create_agent LCEL快速构建 RAG 问答B. 智能客服系统LangChain Agent Middleware需多轮对话与监控C. 自动化任务管理LangGraph Deep Agents LangSmith复杂 workflow 自治 agentD. 内容摘要或转换LCEL轻量、高并行、可流式本章小结LangChain 生态体系可概括为链式逻辑LangChain → 图式编排LangGraph → 监控评估LangSmith → 自治进化Deep Agents核心理念Provider-AgnosticRunnable 统一抽象Middleware 可插架构Production-First 部署思维设计哲学上从“玩具原型”走向“生产可观测”的工程系统。思考与练习练习 1选择一个业务场景画出其 LangChain 技术选型决策路径。练习 2编写一个 LCEL 例程Prompt → Model → Parser → Tool并标注你会插入哪些 Middleware。练习 3设计一个长期运行 Agent 如 市场监控或自动报告说明如何用 LangGraph LangSmith 实现监控与 Evals。思考题LCEL 在 LangGraph 节点中嵌套使用会带来哪些优势与代价第2章核心抽象Runnable 与 LCEL2.1 Runnable Protocol2.1.1 为什么需要统一抽象在 LangChain 1.0 之前不同组件Prompt、Model、Tool、Chain的调用方式各不相同导致接口不一致学习成本高难以组合缺乏标准化无法统一追踪、监控组合困难不同组件难以嵌套使用Runnable Protocol 解决方案LangChain 1.0 引入 Runnable 作为统一执行协议所有组件均实现该接口from langchain_core.runnables import Runnable # 所有组件均实现 Runnable 接口 class Runnable: def invoke(self, input, configNone): ... # 同步调用 def ainvoke(self, input, configNone): ... # 异步调用 def stream(self, input, configNone): ... # 流式输出 def astream(self, input, configNone): ... # 异步流式 def batch(self, inputs, configNone): ... # 批量处理核心优势实际应用示例from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 所有组件都是 Runnable prompt ChatPromptTemplate.from_template(Tell me a joke about {topic}) model ChatOpenAI() parser StrOutputParser() # 统一的调用方式 result prompt.invoke({topic: AI}) result model.invoke(Tell me a joke) result parser.invoke(some text)2.1.2 核心方法invoke、stream、batchinvoke() - 同步调用最基础的调用方式适用于单次请求from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4) # 同步调用 response model.invoke(What is LangChain?) print(response.content)执行流程stream() - 流式输出适用于需要实时反馈的场景如聊天界面# 流式输出 for chunk in model.stream(Tell me a long story): print(chunk.content, end, flushTrue)流式输出的优势✅ 降低首字延迟TTFT - Time To First Token✅ 提升用户体验实时显示✅ 减少超时风险batch() - 批量处理适用于批量请求场景自动优化并发# 批量处理自动并发优化 inputs [ What is AI?, What is ML?, What is LLM? ] results model.batch(inputs) for result in results: print(result.content)批量处理的优势✅ 自动并发控制✅ 成本追踪聚合✅ 错误处理优化abatch() - 异步批量处理在需要高并发处理大量请求时abatch()比同步batch()性能更好import asyncio from langchain_openai import ChatOpenAI model ChatOpenAI() async def async_batch_example(): inputs [ What is AI?, What is ML?, What is LLM?, What is NLP?, What is DL? ] # 异步批量处理 results await model.abatch(inputs) for i, result in enumerate(results): print(fResult {i1}: {result.content}) # 运行异步任务 asyncio.run(async_batch_example())abatch 与 batch 的对比方法适用场景优势batch()中小批量50实现简单无需async/awaitabatch()大批量50、I/O密集更高并发性能资源利用率高2.1.3 异步方法ainvoke、astream在高并发场景下异步方法可显著提升性能ainvoke() - 异步调用import asyncio from langchain_openai import ChatOpenAI model ChatOpenAI() async def main(): # 异步调用 response await model.ainvoke(What is async programming?) print(response.content) asyncio.run(main())astream() - 异步流式async def stream_example(): async for chunk in model.astream(Tell me a story): print(chunk.content, end, flushTrue) asyncio.run(stream_example())并发性能对比# ❌ 同步方式串行执行慢 def sync_batch(): results [] for query in queries: results.append(model.invoke(query)) return results # ✅ 异步方式并发执行快 async def async_batch(): tasks [model.ainvoke(query) for query in queries] return await asyncio.gather(*tasks)性能对比请求数同步耗时异步耗时性能提升1030s5s6x50150s15s10x100300s25s12x2.1.4 Runnable 类型Lambda、Parallel、Branch、FallbacksRunnableLambda - 自定义函数包装将普通 Python 函数包装为 Runnablefrom langchain_core.runnables import RunnableLambda def uppercase(text: str) - str: return text.upper() # 包装为 Runnable runnable_upper RunnableLambda(uppercase) # 统一调用方式 result runnable_upper.invoke(hello) # HELLORunnableParallel - 并行执行同时执行多个 Runnable结果以字典形式返回from langchain_core.runnables import RunnableParallel parallel RunnableParallel( jokeChatPromptTemplate.from_template(Tell a joke about {topic}) | model, poemChatPromptTemplate.from_template(Write a poem about {topic}) | model ) # 并行执行 result parallel.invoke({topic: AI}) print(result[joke]) print(result[poem])RunnableBranch - 条件分支根据条件选择不同的执行路径from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: len(x) 100, long_text_handler), (lambda x: len(x) 10, medium_text_handler), short_text_handler # 默认分支 ) result branch.invoke(some text)with_fallbacks() - 降级处理主 Runnable 失败时自动切换到备用方案from langchain_openai import ChatOpenAI primary_model ChatOpenAI(modelgpt-4) fallback_model ChatOpenAI(modelgpt-3.5-turbo) # 直接使用 with_fallbacks 方法无需导入额外类 model_with_fallback primary_model.with_fallbacks([fallback_model]) # 如果 GPT-4 失败自动使用 GPT-3.5 result model_with_fallback.invoke(Hello)参数说明 (基于官方API文档验证)必需参数fallbacks:Sequence[Runnable]- 备用 Runnable 序列按顺序尝试可选参数 (仅关键字参数)exceptions_to_handle:Tuple[Type[BaseException], ...]- 需要处理的异常类型元组默认为(Exception,)exception_key:Optional[str]- 可选的键名用于将异常信息传递给备用方案。如为None(默认)异常不传递给备用方案完整参数示例# ✅ 示例1: 只对特定异常类型执行降级 model_with_fallback primary_model.with_fallbacks( fallbacks[fallback_model], # ✅ 官方标准参数fallbacks (复数列表) exceptions_to_handle(TimeoutError, ConnectionError), # ✅ 官方标准参数 ) # ✅ 示例2: 将异常信息传递给备用方案 from langchain_core.runnables import RunnableLambda def handle_with_error_context(inputs): 备用方案可以访问异常信息 if error in inputs: print(fOriginal error: {inputs[error]}) return fallback_model.invoke(inputs[input]) model_with_error_context primary_model.with_fallbacks( fallbacks[RunnableLambda(handle_with_error_context)], # ✅ 使用 fallbacks 参数名 exception_keyerror # ✅ 官方标准参数异常会以 error 键传递 ) # ⚠️ 重要使用 exception_key 时主 Runnable 和所有备用方案都必须接受字典输入 result model_with_error_context.invoke({input: Hello})API 规范总结def with_fallbacks( self, fallbacks: Sequence[Runnable[Input, Output]], # 必需 *, exceptions_to_handle: Tuple[Type[BaseException], ...] (Exception,), # 可选 exception_key: Optional[str] None # 可选 ) - RunnableWithFallbacksT[Input, Output]: ...2.2 LCEL 表达式语言2.2.1 声明式组合理念LCELLangChain Expression Language是一种声明式语法用于组合 Runnable 对象。命令式 vs 声明式# ❌ 命令式手动控制流程 def imperative_chain(input): step1_result prompt.invoke(input) step2_result model.invoke(step1_result) step3_result parser.invoke(step2_result) return step3_result # ✅ 声明式LCEL 管道 chain prompt | model | parser result chain.invoke(input)LCEL 的核心优势2.2.2 管道操作符|与并行{}管道操作符|- 顺序执行将多个 Runnable 串联成管道from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 管道组合 chain ( ChatPromptTemplate.from_template(Tell me about {topic}) | ChatOpenAI() | StrOutputParser() ) # 自动按顺序执行 result chain.invoke({topic: LangChain})执行流程并行字典{}- 并行执行使用字典语法实现并行执行from langchain_core.runnables import RunnablePassthrough chain { context: retriever | format_docs, question: RunnablePassthrough() } | prompt | model # context 和 question 并行处理 result chain.invoke(What is LangChain?)执行流程assign() - 状态更新快捷方式RunnablePassthrough.assign()是 LCEL 中最常用的操作之一用于在链中添加或更新字段from langchain_core.runnables import RunnablePassthrough from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 创建向量检索器 vectorstore Chroma.from_texts( [LangChain是一个AI应用框架, 它支持RAG和Agent], embeddingOpenAIEmbeddings() ) retriever vectorstore.as_retriever() # 使用 assign() 添加检索上下文 chain ( RunnablePassthrough.assign( contextretriever # 添加 context 字段 ) | ChatPromptTemplate.from_template( 基于以下上下文回答问题:/n{context}/n/n问题: {question} ) | ChatOpenAI() | StrOutputParser() ) # 输入只需要 questioncontext 会自动添加 result chain.invoke({question: 什么是LangChain?}) # 内部流程: {question: ...} - {question: ..., context: [...]}assign() 的优势保留原始输入不覆盖已有字段简化代码避免手动构造字典链式组合可以多次调用# 多次 assign 叠加字段 chain ( RunnablePassthrough.assign( contextretriever # 添加检索结果 ) .assign( context_countlambda x: len(x[context]) # 添加统计信息 ) .assign( timestamplambda x: 2025-11-17 # 添加时间戳 ) | prompt | model ) # 输入: {question: ...} # 第一步后: {question: ..., context: [...]} # 第二步后: {question: ..., context: [...], context_count: 3} # 第三步后: {question: ..., context: [...], context_count: 3, timestamp: ...}常见使用场景# 场景1: RAG 添加检索上下文 rag_chain ( RunnablePassthrough.assign(contextretriever) | rag_prompt | model ) # 场景2: 添加多个数据源 multi_source_chain ( RunnablePassthrough.assign( docsdoc_retriever, historyhistory_retriever, metadatametadata_fetcher ) | prompt | model ) # 场景3: 数据转换 transform_chain ( RunnablePassthrough.assign( upper_textlambda x: x[text].upper(), word_countlambda x: len(x[text].split()) ) | processor )2.2.3 组合模式顺序、并行、条件、循环顺序链接# 简单顺序 chain step1 | step2 | step3 # 复杂顺序 chain ( {input: RunnablePassthrough()} | prompt | model | {output: parser, raw: RunnablePassthrough()} )并行执行# 并行获取多个信息 chain RunnableParallel( summarysummarize_chain, keywordsextract_keywords_chain, sentimentsentiment_chain )条件分支from langchain_core.runnables import RunnableBranch # 根据输入长度选择不同处理 chain RunnableBranch( (lambda x: len(x[text]) 1000, long_text_chain), (lambda x: len(x[text]) 100, medium_text_chain), short_text_chain )循环迭代# 使用 RunnableLambda 实现循环 def iterative_refine(input): result input for _ in range(3): result refine_chain.invoke(result) return result chain RunnableLambda(iterative_refine)2.3 高级特性2.3.1 Fallback 降级与 Retry 重试Fallback - 自动降级# 多级降级 chain ( primary_model .with_fallbacks(fallbacks[backup_model_1, backup_model_2]) # ✅ 使用 fallbacks 参数名 ) # 只对特定异常执行降级 chain ( primary_model .with_fallbacks( fallbacks[backup_model_1, backup_model_2], # ✅ 使用 fallbacks 参数名 exceptions_to_handle(TimeoutError, ConnectionError) # ✅ 官方标准参数 ) )降级流程Retry - 自动重试# 直接使用 with_retry 方法无需单独导入 chain ( prompt | model | parser ).with_retry( stop_after_attempt3, # 最大重试次数 wait_exponential_jitterTrue, # 指数退避 随机抖动 retry_if_exception_type(Exception,) # 指定需要重试的异常类型 )参数说明基于官方API文档验证stop_after_attempt最大重试次数默认为 3wait_exponential_jitter是否使用指数退避 随机抖动默认为 Trueretry_if_exception_type需要重试的异常类型元组默认为(Exception,)重试策略示例from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 只对特定异常重试 chain (prompt | model | parser).with_retry( retry_if_exception_type(TimeoutError, ConnectionError), stop_after_attempt5, wait_exponential_jitterTrue ) # 禁用指数退避立即重试 chain (prompt | model | parser).with_retry( stop_after_attempt3, wait_exponential_jitterFalse # 禁用指数退避立即重试 )重试行为指数退避1s → 2s → 4s → 8s最大重试次数可自定义默认3次重试条件可指定异常类型默认所有 Exception2.3.2 Timeout 超时控制重要:RunnableConfig不支持timeout参数。超时控制应在模型层面配置。from langchain_openai import ChatOpenAI # ✅ 正确在模型构造时设置timeout model ChatOpenAI( modelgpt-4, timeout30, # 30秒超时 max_retries2 ) chain prompt | model | parser result chain.invoke(input)超时 降级组合策略from langchain_openai import ChatOpenAI # 主模型严格超时 slow_model ChatOpenAI(modelgpt-4, timeout10) # 降级模型快速响应 fast_model ChatOpenAI(modelgpt-3.5-turbo, timeout5) # 组合超时 降级 chain (prompt | slow_model | parser).with_fallbacks([ prompt | fast_model | parser ])使用 RunnableConfig 配置其他参数from langchain_core.runnables import RunnableConfig # RunnableConfig支持的参数 result chain.invoke( input, configRunnableConfig( max_concurrency5, # 最大并发数 tags[production], # 标签用于监控 metadata{user: alice} # 元数据 ) )2.3.3 缓存与性能优化 提示: 本节介绍 Runnable Protocol 的基础性能API。生产环境的深度性能调优、成本控制、缓存架构等内容详见 第八篇《生产实践》第21章。LLM 缓存from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache # 启用缓存 set_llm_cache(InMemoryCache()) # 相同请求直接返回缓存结果 model.invoke(What is AI?) # 调用 LLM model.invoke(What is AI?) # 返回缓存不调用 LLM批处理优化# 批处理优化使用max_concurrency控制并发 chain prompt | model.with_config({max_concurrency: 10}) # 内部自动合并请求 results chain.batch(inputs)流式优化# 流式传输减少延迟 for chunk in chain.stream(input): print(chunk, end)性能对比特性普通调用优化后性能提升缓存2s50ms40x批处理10s2s5x流式TTFT 2sTTFT 200ms10x本章小结Runnable Protocol 核心要点✅ 统一接口invoke、stream、batch、ainvoke、astream✅ 可组合性Lambda、Parallel、Branch、Fallbacks✅ 可追踪性自动集成 LangSmith✅ 性能优化异步、批处理、缓存LCEL 核心要点✅ 声明式组合|管道、{}并行✅ 自动优化并行执行、流式传输✅ 高级特性Fallback、Retry、Timeout、Cache设计哲学一切皆 Runnable所有组件统一接口声明式组合自动优化执行。思考与练习练习 1基础管道构建一个 LCEL 管道Prompt → Model → Parser实现一个简单的问答系统。练习 2并行处理使用 RunnableParallel 同时生成一个笑话、一首诗和一个故事输入主题为AI。练习 3错误处理实现一个带有 Fallback 和 Retry 的 chain主模型失败时自动切换到备用模型。练习 4性能优化对比同步批处理和异步批处理的性能差异10个请求。思考题什么场景下应该使用 stream 而不是 invokeRunnableBranch 和简单的 if-else 有什么区别如何在 LCEL 中实现循环逻辑如何学习大模型 AI 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取