一、为什么需要 LLM 抽象1.1 大模型 API 碎片化的行业痛点在 LangChain 出现之前开发 LLM 应用需要面对严重的接口碎片化问题。不同厂商的大模型 API 在请求格式、参数命名、返回值结构、错误码等方面完全不兼容导致切换模型成本极高从 OpenAI 切换到 Claude 需要重写所有模型调用代码重复造轮子每个项目都需要重新实现重试、限流、日志、Token 统计等通用功能代码可维护性差硬编码的 API 调用散落在代码各处难以统一升级和维护示例不同模型的原生 API 调用对比# OpenAI原生API from openai import OpenAI client OpenAI() response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 你好}], temperature0.7 ) print(response.choices[0].message.content) # 字节跳动豆包原生API from volcenginesdkarkruntime import Ark client Ark() response client.chat.completions.create( modelep-20240515123456789, messages[{role: user, content: 你好}], temperature0.7 ) print(response.choices[0].message.content) # 阿里通义千问原生API from dashscope import Generation response Generation.call( modelqwen-turbo, messages[{role: user, content: 你好}], temperature0.7 ) print(response.output.choices[0].message.content)可以看到三个模型的 API 看起来相似但在客户端初始化、模型名称、返回值结构等方面存在细微差异这些差异会随着模型数量的增加被无限放大。1.2 LangChain 的解决方案统一抽象层LangChain 通过BaseChatModel基类定义了所有聊天模型的统一接口将不同厂商的 API 差异封装在底层为上层应用提供了完全一致的编程体验。LangChain LLM 抽象架构图这种设计带来了三大核心优势模型无关性上层应用代码不依赖任何特定模型切换模型只需要修改一行配置功能复用重试、限流、日志、监控等通用功能可以在抽象层统一实现可扩展性新增模型只需要实现BaseChatModel接口上层应用无需任何修改1.3 四种调用模式的适用场景LangChain 的BaseChatModel支持四种调用模式分别适用于不同的业务场景调用模式方法适用场景特点同步调用invoke()简单请求、低并发场景代码简单阻塞等待结果异步调用ainvoke()高并发服务、批量处理非阻塞支持高并发流式输出stream()聊天机器人、实时交互逐令牌返回结果提升用户体验批量调用batch()批量处理大量请求合并请求提高吞吐量二、核心 API 详解2.1 BaseChatModel 基类BaseChatModel是所有聊天模型的抽象基类定义了所有模型必须实现的核心方法和属性。它继承自Runnable因此天然支持 LCEL 表达式语言可以与其他 Runnable 组件无缝组合。BaseChatModel 核心方法from langchain_core.language_models import BaseChatModel from langchain_core.messages import BaseMessage, AIMessage from typing import List, Optional class BaseChatModel: def invoke( self, input: List[BaseMessage] | str, config: Optional[RunnableConfig] None ) - AIMessage: 同步调用模型返回完整的AI消息 pass async def ainvoke( self, input: List[BaseMessage] | str, config: Optional[RunnableConfig] None ) - AIMessage: 异步调用模型 pass def stream( self, input: List[BaseMessage] | str, config: Optional[RunnableConfig] None ) - Iterator[AIMessageChunk]: 流式调用模型返回消息块迭代器 pass async def astream( self, input: List[BaseMessage] | str, config: Optional[RunnableConfig] None ) - AsyncIterator[AIMessageChunk]: 异步流式调用模型 pass def batch( self, inputs: List[List[BaseMessage] | str], config: Optional[RunnableConfig] None ) - List[AIMessage]: 批量调用模型 pass2.2 消息类型系统LangChain 定义了一套标准化的消息类型系统用于表示对话中的不同角色和内容。所有消息都继承自BaseMessage基类。核心消息类型消息类型角色用途HumanMessage用户表示用户发送的消息AIMessageAI 助手表示 AI 生成的回复消息SystemMessage系统表示系统提示词用于设定 AI 的角色和行为ToolMessage工具表示工具调用的返回结果FunctionMessage函数旧版函数调用消息已被ToolMessage替代消息内容块ContentBlock LangChain 1.2 支持多模态内容消息的content字段可以是字符串或内容块列表。内容块可以是文本、图像、音频、视频等多种类型。from langchain_core.messages import HumanMessage # 纯文本消息 text_message HumanMessage(content你好LangChain) # 多模态消息文本图像 multimodal_message HumanMessage(content[ {type: text, text: 这张图片里有什么}, { type: image_url, image_url: {url: https://example.com/image.jpg} } ])2.3 主流大模型集成LangChain 通过官方 Partner 包提供了对主流大模型的支持所有模型都实现了BaseChatModel接口使用方式完全一致。2.3.1 OpenAI GPT 系列官方推荐OpenAI 是 LangChain 生态支持最完善的模型也是开发和测试的首选。安装依赖poetry add langchain-openai0.2.2from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 初始化模型 llm ChatOpenAI( modelgpt-4o-mini-2024-07-18, # 推荐使用带版本号的模型名称 temperature0, max_tokens1024, timeout30, max_retries2 ) # 调用模型 messages [ SystemMessage(content你是一个专业的Python开发助手), HumanMessage(content什么是装饰器) ] response llm.invoke(messages) print(response.content)2.3.2 字节跳动豆包 4.0国内首选字节跳动豆包是国内性能和效果最好的大模型之一LangChain 通过langchain-volcengine包提供支持。安装依赖poetry add langchain-volcengine0.1.3from langchain_volcengine import ChatVolcEngine from langchain_core.messages import HumanMessage # 初始化模型 # 需要在.env中配置VOLC_ACCESSKEY和VOLC_SECRETKEY llm ChatVolcEngine( modelep-20240515123456789, # 替换为你的模型端点ID temperature0, max_tokens1024 ) # 调用方式与OpenAI完全一致 response llm.invoke([HumanMessage(content什么是装饰器)]) print(response.content)2.3.3 阿里通义千问 3.5阿里通义千问是国内另一个主流大模型LangChain 通过langchain-dashscope包提供支持。安装依赖poetry add langchain-dashscope0.1.6from langchain_dashscope import ChatTongyi from langchain_core.messages import HumanMessage # 初始化模型 # 需要在.env中配置DASHSCOPE_API_KEY llm ChatTongyi( modelqwen-turbo, temperature0, max_tokens1024 ) # 调用方式完全一致 response llm.invoke([HumanMessage(content什么是装饰器)]) print(response.content)2.3.4 本地模型集成Ollama对于需要本地部署的场景Ollama 是最简单的本地大模型运行工具LangChain 通过langchain-ollama包提供支持。安装 Ollama访问Ollama 官网下载并安装运行ollama pull qwen:7b下载 Qwen 2.5 7B 模型安装依赖poetry add langchain-ollama0.2.0from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage # 初始化本地模型 llm ChatOllama( modelqwen:7b, temperature0, max_tokens1024 ) # 调用方式完全一致 response llm.invoke([HumanMessage(content什么是装饰器)]) print(response.content)关键结论所有模型的调用方式完全相同切换模型只需要修改初始化代码上层业务逻辑无需任何改动。2.4 模型参数深度解析LangChain 的BaseChatModel支持一系列通用参数用于控制模型的输出行为。理解这些参数的作用是优化 LLM 应用效果的关键。参数作用取值范围推荐值场景说明temperature控制输出的随机性0-20事实性任务、0.7创意性任务值越高输出越随机、越有创意值越低输出越确定、越一致max_tokens限制模型生成的最大令牌数1 - 模型上下文窗口1024-4096防止生成过长的回复控制成本top_p核采样参数控制词汇选择范围0-10.9值越小输出越集中在高概率词汇上值越大输出越多样frequency_penalty频率惩罚减少重复内容-2.0-2.00.0-1.0值越高越不容易出现重复的短语presence_penalty存在惩罚鼓励生成新主题-2.0-2.00.0-1.0值越高越容易引入新的话题timeout请求超时时间秒正数30-60防止请求长时间阻塞max_retries最大重试次数非负整数2-3自动重试失败的请求参数调优最佳实践事实性任务问答、摘要、代码生成temperature0保证输出的准确性和一致性创意性任务写作、头脑风暴、文案生成temperature0.7-1.0增加输出的多样性永远不要使用temperature2会导致输出完全混乱没有任何意义top_p和temperature不要同时调整通常只调整其中一个即可三、工业级最佳实践简单的模型调用只能用于原型开发要达到生产环境的要求必须实现重试、降级、限流、监控等一系列工业级特性。3.1 模型工厂模式模型工厂模式是管理多个模型的最佳实践它将模型的初始化逻辑封装在工厂类中上层应用通过配置获取模型实例实现了配置与代码的分离。工业级模型工厂实现from typing import Dict, Optional from pydantic import BaseModel, Field from langchain_core.language_models import BaseChatModel from langchain_openai import ChatOpenAI from langchain_ollama import ChatOllama class ModelConfig(BaseModel): 模型配置类 provider: str Field(description模型提供商openai/ollama) model_name: str Field(description模型名称) temperature: float Field(default0.0) max_tokens: int Field(default1024) timeout: int Field(default30) max_retries: int Field(default2) class ModelFactory: 模型工厂类 _instances: Dict[str, BaseChatModel] {} classmethod def get_model(cls, config: ModelConfig) - BaseChatModel: 获取模型实例单例 instance_key f{config.provider}:{config.model_name}:{config.temperature} if instance_key not in cls._instances: if config.provider openai: model ChatOpenAI( modelconfig.model_name, temperatureconfig.temperature, max_tokensconfig.max_tokens, timeoutconfig.timeout, max_retriesconfig.max_retries ) elif config.provider ollama: model ChatOllama( modelconfig.model_name, temperatureconfig.temperature, max_tokensconfig.max_tokens, timeoutconfig.timeout ) else: raise ValueError(f不支持的提供商{config.provider}) cls._instances[instance_key] model return cls._instances[instance_key] # 使用示例 config ModelConfig( provideropenai, model_namegpt-4o-mini, temperature0 ) llm ModelFactory.get_model(config) print(llm.invoke(你好))3.2 重试机制指数退避 抖动使用 LangChain 内置的ModelRetryMiddleware实现工业级重试策略解决 API 不稳定问题。通用实现from langchain_core.middleware import ModelRetryMiddleware from tenacity import stop_after_attempt, wait_exponential_jitter # 创建重试中间件 retry_middleware ModelRetryMiddleware( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential_jitter( multiplier1, # 初始等待1秒 min1, # 最小等待1秒 max10 # 最大等待10秒 ), # 只重试可恢复的异常 retry_exceptions( TimeoutError, ConnectionError, openai.RateLimitError, openai.APIError, openai.APIConnectionError ) ) # 应用到模型 llm_with_retry llm.with_middleware(retry_middleware)3.3 自动降级策略当主模型不可用时自动切换到备用模型保证服务可用性。通用实现# 主模型GPT-4o-mini primary_llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 备用模型本地Qwen 7B fallback_llm ChatOllama(modelqwen:7b, temperature0) # 组合模型主模型失败时自动切换到备用 robust_llm primary_llm.with_fallbacks([fallback_llm])3.4 Token 统计与成本控制使用CostTrackingMiddleware自动统计 Token 使用量和成本。通用实现from langchain_core.middleware import CostTrackingMiddleware cost_tracker CostTrackingMiddleware() llm_with_cost llm.with_middleware(cost_tracker) # 调用模型 response llm_with_cost.invoke(统计Token使用量) # 获取统计结果 print(f总Token数{cost_tracker.total_tokens}) print(f输入Token{cost_tracker.prompt_tokens}) print(f输出Token{cost_tracker.completion_tokens}) print(f总成本${cost_tracker.total_cost:.6f})四、新建项目现有 RAG 项目在整合过程中发现遇到依赖版本问题无法使用langchain1.x以上版本所以决定重新创建新工程。4.1 项目初始化# 创建项目目录 mkdir langchain-2026 cd langchain-2026 # 创建虚拟环境必须隔离系统Python python -m venv .venv # 激活虚拟环境 # Windows PowerShell .venv\Scripts\Activate.ps14.2 项目标准结构langchain-2026-framework/ ├── .env # 环境变量敏感信息 ├── .gitignore # Git忽略文件 ├── requirements.txt # 依赖清单 ├── main.py # 项目入口 ├── config/ # 配置管理 │ ├── __init__.py │ └── settings.py # 全局配置 ├── core/ # 核心业务层 │ ├── __init__.py │ ├── llm_factory.py # 模型工厂核心 │ ├── rag.py # RAG模块后续扩展 │ └── agent.py # Agent模块后续扩展 ├── utils/ # 通用工具 │ ├── __init__.py │ ├── logger.py # 日志系统 │ └── exceptions.py # 自定义异常 └── tests/ # 单元测试 └── test_llm.py # LLM模块测试4.3 依赖安装2026 年 5 月最新稳定版创建requirements.txt写入以下内容无需指定版本自动安装最新兼容版不要用中文# LangChain核心生态2026年5月最新稳定版 langchain langchain-core langchain-openai langchain-ollama langgraph langsmith # 工具依赖 python-dotenv pydantic-settings tenacity aiohttp # 开发依赖 pytest black isort安装依赖pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple4.4 配置管理config/settings.pyfrom pydantic_settings import BaseSettings, SettingsConfigDict from pydantic import Field from pathlib import Path class Settings(BaseSettings): model_config SettingsConfigDict( env_file.env, env_file_encodingutf-8, case_sensitiveFalse, extraignore ) # 项目基础配置 app_name: str Field(defaultLangChain 2026开发框架, description应用名称) debug: bool Field(defaultTrue, description调试模式) # 大模型API配置 doubao_api_key: str Field(default, description豆包API密钥) doubao_base_url: str Field(defaulthttps://ark.cn-beijing.volces.com/api/v3, description豆包API地址) openai_api_key: str Field(default, descriptionOpenAI API密钥) # 默认模型配置 default_provider: str Field(defaultdoubao, description默认模型提供商) default_model: str Field(defaultdoubao-seed-2-0-code-xxxx, description默认模型名称) llm_temperature: float Field(default0.1, description温度参数) llm_max_tokens: int Field(default1024, description最大生成长度) llm_timeout: int Field(default60, description超时时间秒) llm_max_retries: int Field(default3, description最大重试次数) # LangSmith可观测性配置 langchain_tracing_v2: bool Field(defaultFalse, description是否开启LangSmith追踪) langchain_api_key: str Field(default, descriptionLangSmith API密钥) langchain_project: str Field(defaultlangchain-2026-framework, descriptionLangSmith项目名称) # 日志配置 log_level: str Field(defaultINFO, description日志级别) log_dir: Path Field(defaultPath(./logs), description日志目录) # 全局单例 settings Settings()4.5 日志系统utils/logger.pyimport logging from logging.handlers import RotatingFileHandler from pathlib import Path from config.settings import settings def setup_logger(): 配置全局日志系统 # 创建日志目录 settings.log_dir.mkdir(exist_okTrue) # 获取根日志器 logger logging.getLogger() logger.setLevel(settings.log_level) logger.handlers.clear() # 日志格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s, datefmt%Y-%m-%d %H:%M:%S ) # 控制台处理器 console_handler logging.StreamHandler() console_handler.setFormatter(formatter) console_handler.setLevel(logging.INFO) # 文件处理器 file_handler RotatingFileHandler( filenamesettings.log_dir / app.log, maxBytes10*1024*1024, backupCount5, encodingutf-8 ) file_handler.setFormatter(formatter) file_handler.setLevel(logging.DEBUG) # 添加处理器 logger.addHandler(console_handler) logger.addHandler(file_handler) # 第三方库日志级别 logging.getLogger(httpx).setLevel(logging.WARNING) logging.getLogger(openai).setLevel(logging.WARNING) logging.getLogger(langchain).setLevel(logging.INFO) logger.info(✅ 日志系统初始化完成) return logger # 全局日志器 logger setup_logger()4.6 自定义异常utils/exceptions.pyfrom utils.logger import logger class FrameworkBaseException(Exception): 框架基础异常类 def __init__(self, message: str, error_code: int 1000): self.message message self.error_code error_code super().__init__(self.message) logger.error(f[{self.error_code}] {self.message}) class ModelLoadError(FrameworkBaseException): 模型加载失败异常 def __init__(self, model_name: str, details: str ): message f模型加载失败{model_name} if details: message f详细信息{details} super().__init__(message, error_code1001) class GenerationError(FrameworkBaseException): 生成失败异常 def __init__(self, details: str ): message 回答生成失败 if details: message f详细信息{details} super().__init__(message, error_code1002)4.7 核心模型工厂core/llm_factory.pyfrom typing import Dict, Optional, Iterator, AsyncIterator from langchain_core.language_models import BaseChatModel from langchain_openai import ChatOpenAI from langchain_ollama import ChatOllama from langchain_core.runnables import RunnableRetry from tenacity import stop_after_attempt, wait_exponential_jitter from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from config.settings import settings from utils.logger import logger from utils.exceptions import ModelLoadError, GenerationError class LLMFactory: 大模型工厂类2026最新版 支持豆包、OpenAI、本地Ollama模型 内置重试、降级、Token统计、流式输出 _instances: Dict[str, BaseChatModel] {} classmethod def get_llm( cls, provider: str None, model_name: str None, temperature: float None, max_tokens: int None ) - BaseChatModel: 获取模型实例单例模式 :param provider: 模型提供商doubao/openai/ollama :param model_name: 模型名称 :param temperature: 温度参数 :param max_tokens: 最大生成长度 :return: LangChain ChatModel实例 # 使用默认配置 provider provider or settings.default_provider model_name model_name or settings.default_model temperature temperature if temperature is not None else settings.llm_temperature max_tokens max_tokens or settings.llm_max_tokens # 生成唯一实例键 instance_key f{provider}:{model_name}:{temperature}:{max_tokens} if instance_key not in cls._instances: try: # 创建基础模型 base_llm cls._create_base_llm(provider, model_name, temperature, max_tokens) # 添加重试机制LangChain 2026标准写法 llm_with_retry base_llm.with_retry( retry_if_exception_type(TimeoutError, ConnectionError, Exception), wait_exponential_jitterTrue, stop_after_attemptsettings.llm_max_retries, exponential_jitter_params{initial: 1, max: 10} ) cls._instances[instance_key] llm_with_retry logger.info(f✅ 模型初始化成功{provider} {model_name}) except Exception as e: raise ModelLoadError(f{provider} {model_name}, str(e)) from e return cls._instances[instance_key] classmethod def _create_base_llm( cls, provider: str, model_name: str, temperature: float, max_tokens: int ) - BaseChatModel: 创建基础模型实例 if provider doubao: # 豆包API完全兼容OpenAI格式 return ChatOpenAI( api_keysettings.doubao_api_key, base_urlsettings.doubao_base_url, modelmodel_name, temperaturetemperature, max_tokensmax_tokens, timeoutsettings.llm_timeout ) elif provider openai: return ChatOpenAI( api_keysettings.openai_api_key, modelmodel_name, temperaturetemperature, max_tokensmax_tokens, timeoutsettings.llm_timeout ) elif provider ollama: return ChatOllama( modelmodel_name, temperaturetemperature, max_tokensmax_tokens, timeoutsettings.llm_timeout * 2 ) else: raise ValueError(f不支持的模型提供商{provider}) classmethod def generate( cls, prompt: str | list[BaseMessage], provider: str None, model_name: str None, temperature: float None, max_tokens: int None ) - str: 同步生成文本 :param prompt: 提示词字符串或消息列表 :return: 生成的文本 try: llm cls.get_llm(provider, model_name, temperature, max_tokens) # 处理字符串输入 if isinstance(prompt, str): messages [HumanMessage(contentprompt)] else: messages prompt response llm.invoke(messages) # 统计Token使用量 if response.usage_metadata: logger.debug( fToken统计 | 总{response.usage_metadata[total_tokens]} | f输入{response.usage_metadata[input_tokens]} | f输出{response.usage_metadata[output_tokens]} ) return response.content.strip() except Exception as e: raise GenerationError(str(e)) from e classmethod def stream_generate( cls, prompt: str | list[BaseMessage], provider: str None, model_name: str None, temperature: float None, max_tokens: int None ) - Iterator[str]: 流式生成文本 try: llm cls.get_llm(provider, model_name, temperature, max_tokens) if isinstance(prompt, str): messages [HumanMessage(contentprompt)] else: messages prompt for chunk in llm.stream(messages): if chunk.content: yield chunk.content except Exception as e: logger.error(f流式生成失败{str(e)}) yield 抱歉生成回答时发生错误请稍后重试。 classmethod async def astream_generate( cls, prompt: str | list[BaseMessage], provider: str None, model_name: str None, temperature: float None, max_tokens: int None ) - AsyncIterator[str]: 异步流式生成文本 try: llm cls.get_llm(provider, model_name, temperature, max_tokens) if isinstance(prompt, str): messages [HumanMessage(contentprompt)] else: messages prompt async for chunk in llm.astream(messages): if chunk.content: yield chunk.content except Exception as e: logger.error(f异步流式生成失败{str(e)}) yield 抱歉生成回答时发生错误请稍后重试。4.8 环境变量配置.env# 豆包API推荐中文效果好 DOUBAO_API_KEY你的豆包API密钥 DOUBAO_BASE_URLhttps://ark.cn-beijing.volces.com/api/v3 # OpenAI API可选 # OPENAI_API_KEY你的OpenAI API密钥 # LangSmith可观测性可选调试用 # LANGCHAIN_TRACING_V2true # LANGCHAIN_API_KEY你的LangSmith API密钥 # LANGCHAIN_PROJECTlangchain-2026-framework # 调试模式 DEBUGtrue4.9 项目入口与验证main.pyfrom dotenv import load_dotenv load_dotenv() # 必须在导入其他模块之前加载环境变量 from core.llm_factory import LLMFactory from langchain_core.messages import SystemMessage, HumanMessage import asyncio def test_basic_generate(): 测试基础生成功能 print( 测试基础生成功能 ) response LLMFactory.generate(用3句话解释什么是大语言模型) print(f回答{response}\n) def test_stream_generate(): 测试流式生成功能 print( 测试流式生成功能 ) print(回答, end, flushTrue) for chunk in LLMFactory.stream_generate(写一首关于夏天的七言绝句): print(chunk, end, flushTrue) print(\n) def test_system_prompt(): 测试系统提示词 print( 测试系统提示词 ) messages [ SystemMessage(content你是一个专业的Python开发助手只回答Python相关问题), HumanMessage(content什么是装饰器) ] response LLMFactory.generate(messages) print(f回答{response}\n) pytest.mark.asyncio async def test_astream_generate(): 测试异步流式生成 print( 测试异步流式生成 ) print(回答, end, flushTrue) async for chunk in LLMFactory.astream_generate(什么是异步编程): print(chunk, end, flushTrue) print(\n) if __name__ __main__: print( LangChain 2026开发框架启动\n) # 运行所有测试 test_basic_generate() test_stream_generate() test_system_prompt() # 运行异步测试 asyncio.run(test_astream_generate()) print( 所有测试通过框架搭建完成)4.10 测试