python langchain案例

完整功能总结一、整体概述这份代码是一套面向企业级 AI 应用开发的LangChain 标准化实战 Demo 集基于 LangChain1.x 新版本开发兼容 DeepSeek、通义千问等全量 OpenAI 兼容接口模型覆盖模型调用、多模态图文、并发限流、提示词工程、文档分块、文本向量化、面向接口多态7 大核心开发场景完整覆盖 RAG 知识库、文档解析、批量任务、多模型适配等主流业务需求可直接作为项目开发模板。二、7 大 Demo 分项功能详解Demo1ChatOpenAI 基础文本调用最核心基础能力核心组件ChatOpenAI、SystemMessage/HumanMessage对话消息结构功能要点统一封装兼容 OpenAI 格式的各类大模型DeepSeek、千问等仅需修改API_KEY/BASE_URL/MODEL_ID即可切换模型通过 SystemMessage 固定 AI 角色、输出约束HumanMessage 承载用户输入标准化对话入参配置生产级参数temperature0保证输出稳定、超时时间、失败自动重试使用invoke标准同步调用是项目 90% 文本问答、信息提取的基础写法。业务场景文本问答、知识解释、简单文案生成。Demo2多模态图片识别调用核心组件多模态ChatOpenAI、Base64 图片编码、图片 URL 入参功能要点支持两种图片传入方式网络图片 URL、本地图片转 Base64 编码HumanMessage 支持混合图文结构体同时传入图片 文本提问调用逻辑与纯文本完全统一仅扩展入参结构无缝对接千问 VL 等视觉大模型。业务场景文档截图 OCR 识别、图纸解析、图片内容描述、图文混合文档分析。Demo3多线程并发限流控制核心组件threading.BoundedSemaphore信号量、多线程 Worker功能要点解决批量调用 LLM 时请求过载、接口限流报错问题通过信号量设置全局最大并发数同一时间仅允许 N 条 LLM 请求执行封装安全调用函数自动抢占 / 释放资源增加等待超时保护模拟多线程批量任务适配文档批量解析、批量数据抽取场景。业务场景大批量文档并行处理、批量实体抽取、多任务异步 AI 处理。Demo4ChatPromptTemplate 结构化提示词模板核心组件ChatPromptTemplate、模板 partial 固化、结构化 JSON 输出功能要点提示词模板化、可复用通过占位符{变量}动态传入角色、文本、抽取字段partial固化固定不变的规则 / 输出格式仅传入动态业务数据简化调用强制模型输出标准 JSON 结构配合 json.loads 直接转为字典实现结构化信息抽取。业务场景合同 / 文档关键字段提取、表单信息抽取、标准化结构化数据输出。Demo5MarkdownHeaderTextSplitter 文档分层分块核心组件Markdown 标题分割器、Document 文档对象、metadata 元数据解析功能要点按 Markdown 一至六级标题自动切割长文档保留每一块对应的完整标题层级分割后的文档块自带metadata元数据存储各级标题名称封装工具自动提取完整标题链记录段落所属章节解决长文档上下文丢失问题支持二次精细分割是 RAG 知识库构建中文档预处理核心步骤。业务场景技术手册、项目方案、Markdown 知识库拆分为向量入库做预处理。Demo6文本向量化Embedding两种实现方案方案 1OpenAI 官方兼容 Embedding 客户端直接调用模型向量接口批量文本一次性生成向量内置余弦相似度计算函数快速判断文本语义相似度适配千问文本嵌入模型用于知识库检索匹配。方案 2原生 requests 请求本地向量服务embeddingsV2不依赖 LangChain 封装通过 HTTP 请求调用私有化部署向量服务自定义向量维度、输入文本适配本地私有化向量引擎。业务场景RAG 向量库构建、文本语义检索、知识库相似度匹配。Demo7BaseChatModel 抽象基类面向接口多态设计核心组件BaseChatModel抽象接口、抽象基类 ABC、自定义业务接口功能要点LangChain 顶层抽象模型接口所有对话模型ChatOpenAI、Azure 模型等均实现该接口业务代码以BaseChatModel做类型注解实现多态传入任意兼容模型均可正常运行面向接口编程后续切换模型、新增厂商模型无需修改业务逻辑代码解耦、扩展性强。业务场景企业多模型混合调度、支持随时切换不同厂商大模型的通用底层服务。三、代码整体架构与生产价值统一配置层集中管理 API 密钥、模型地址、模型名称环境变量优先读取便于多环境部署由浅入深学习顺序基础调用→多模态→提示词模板→文档分割→向量入库→并发处理→底层抽象设计完全贴合 RAG 应用开发全流程企业落地优势全部代码适配生产环境内置重试、超时、并发限流等异常保护兼容公有云模型DeepSeek、通义千问与私有化本地服务组件高度解耦可单独抽取任意 Demo 模块集成到自有业务系统完整覆盖 RAG 全链路文档分割→文本向量化→向量检索→LLM 问答 / 信息抽取。四、适用开发场景汇总知识库问答系统RAG完整链路开发文档智能解析合同、技术文档关键字自动提取图文混合内容识别、截图 OCR 解析大批量文档并行批量 AI 处理支持多模型灵活切换的通用 AI 底层服务私有化向量服务、本地大模型服务对接开发。 LangChain 功能演示 conda create -n langchain_project python3.11.13 conda activate langchain_project pip install langchain1.3.9 langchain-openai1.1.0 langchain-text-splitters1.1.2 按学习顺序排列。 运行方式 直接运行本文件即可看到所有 demo 的输出。 部分 demo 需要有效的 API Key请设置环境变量或修改下方配置。 import os import threading import time from typing import List # # 配置区请根据实际情况修改 # API_KEY os.getenv(API_KEY, sk-xxxx) BASE_URL os.getenv(BASE_URL, xxxx) MODEL_ID os.getenv(MODEL_ID, xxxx) #官网deepseek模型 # API_KEY os.getenv(API_KEY, sk-xxx) # BASE_URL os.getenv(BASE_URL, https://api.deepseek.com/v1) # MODEL_ID os.getenv(MODEL_ID, deepseek-v4-flash) #官网千问视觉模型可参考官方API调用方式 # API_KEY os.getenv(API_KEY, sk-xxx) # BASE_URL os.getenv(BASE_URL, https://llm-7s1oce6uylaeiab1.cn-beijing.maas.aliyuncs.com/compatible-mode/v1) # MODEL_ID os.getenv(MODEL_ID, qwen3-vl-flash) #官网千问向量化模型可参考官方API调用方式 # API_KEY os.getenv(API_KEY, sk-22195fc1ee84455498b01f7cf8479ce0) # BASE_URL os.getenv(BASE_URL, https://llm-7s1oce6uylaeiab1.cn-beijing.maas.aliyuncs.com/compatible-mode/v1) # MODEL_ID os.getenv(MODEL_ID, qwen3.7-text-embedding) # # 用法 1: ChatOpenAI 基础调用 # 这是最核心、最常用的模式90% # def demo_basic_llm_call(): 项目中的标准用法 1. 创建 ChatOpenAI 实例全局单例 2. 用 SystemMessage HumanMessage 构建消息 3. 调用 llm.invoke(messages) 获取回复 print( * 60) print(【用法 1】ChatOpenAI 基础调用SystemMessage HumanMessage) print( * 60) ChatOpenAI — 来自 langchain_openai LLM 调用客户端封装了 OpenAI 兼容的 API 接口。项目中使用它来对接通义千问、DeepSeek 等任意 OpenAI 兼容模型 不只是 OpenAI 自家的模型。 SystemMessage设定 AI 的身份、行为规则、输出格式约束 例如给AI身份你是一个专业的文档分析助手。请简洁地回答用户的问题。 HumanMessage,用户输入的问题或待处理内容 from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage # 1. 创建 LLM 客户端项目中通常是全局单例只创建一次 model 模型名称/ID指向一个部署的 DeepSeek 模型 api_key 认证密钥用于调用 API的地址 temperature 生成随机性。0.0 表示每次都输出最确定的答案项目里几乎全用 0.0 保证稳定性 timeout 请求超时秒数120 秒后还没返回就放弃 max_retries 请求失败后的最大重试次数 base_url 模型地址 llm ChatOpenAI( modelMODEL_ID, api_keyAPI_KEY, base_urlBASE_URL, temperature0.0, # 项目里几乎都用 0.0保证输出稳定 timeout120, # 超时时间 max_retries3, # 重试次数 ) # 2. 构建消息列表 SystemMessage #给系统的提示词假设AI身份 HumanMessage #这是用户输入的问AI的问题 messages [ SystemMessage(content你是一个专业的文档分析助手。请简洁地回答用户的问题。), HumanMessage(content请用一句话解释什么是 RAG。), ] # 3. 调用 LLM项目中 100% 使用 invoke 同步调用 try: response llm.invoke(messages) 回复: RAG检索增强生成是一种结合信息检索与语言生成的技术通过从外部知识库检索相关文档 来增强大语言模型的回答准确性和知识覆盖面。 print(f 回复: {response.content}\n) except Exception as e: print(f [跳过] API 调用失败: {e}请设置正确的 API_KEY 和 BASE_URL\n) # # 用法 2: 多模态图片识别HumanMessage 传图片 # def demo_multimodal_call(): 将图片以 Base64 或 URL 的形式嵌入 HumanMessage传给多模态模型。 用于 OCR 识别、图片内容理解等场景。 print( * 60) print(【用法 2】多模态图片识别HumanMessage 传图片) print( * 60) import base64 from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm ChatOpenAI( modelMODEL_ID, api_keyAPI_KEY, base_urlBASE_URL, temperature0.0, ) # 读取本地 cat.png 并转为 Base64 image_path os.path.join(os.path.dirname(__file__), cat.png) with open(image_path, rb) as f: b64_data base64.b64encode(f.read()).decode(utf-8) # 将 Base64 图片嵌入 HumanMessage message HumanMessage(content[ { type: image_url, image_url: {url: https://gips0.baidu.com/it/u3560029307,576412274fm3028app3028fJPEGfmtauto?w960h1280}, }, # { # type: image_url, # image_url: {url: fdata:image/png;base64,{b64_data}}, # }, { type: text, #text:SYSTEM_PIC_PROMPT,#一段超级长的提示词 text: 请描述一下图片内容, }, ]) # 调用方式与普通文本完全一样 图片描述:这是一幅充满童趣与梦幻感的3D卡通风格插画画面主体是一个可爱的小男孩骑在一条巨大的金鱼身上整体色调温暖柔和。 try: response llm.invoke([message]) print(f 图片描述: {response.content}\n) except Exception as e: print(f [跳过] {e}\n) # # 用法 3: 并发控制BoundedSemaphore 限制并发数 # def demo_concurrency_control(): 项目中的用法 用 threading.BoundedSemaphore 限制 LLM 并发调用数 避免同时发太多请求把模型服务打满。 print( * 60) print(【用法 3】并发控制BoundedSemaphore 限制并发数) print( * 60) from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage llm ChatOpenAI( modelMODEL_ID, api_keyAPI_KEY, base_urlBASE_URL, temperature0.0, ) 核心并发控制代码 # 模拟并发控制逻辑 # 最多同时 2 个请求 MAX_CONCURRENCY 2 lock threading.BoundedSemaphore(MAX_CONCURRENCY) def safe_llm_invoke(llm, messages): 带并发控制的 LLM 调用与项目中的 llm_invoke 一致 acquired lock.acquire(timeout10) if not acquired: raise TimeoutError(等待 LLM 资源超时) try: time.sleep(15) return llm.invoke(messages) finally: # 用完后释放让其他线程可以调用 lock.release() def worker(task_id: int): messages [ SystemMessage(content请用一句话回答。), HumanMessage(contentf任务 {task_id}: 11 等于几), ] try: resp safe_llm_invoke(llm, messages) print(f 线程 {task_id}: {resp.content}) except Exception as e: print(f 线程 {task_id}: [跳过] {e}) # 启动 4 个线程但同一时间最多 2 个在调用 print( 启动 4 个线程并发上限 2...) threads [threading.Thread(targetworker, args(i,)) for i in range(4)] for t in threads: t.start() for t in threads: t.join() print() # # 用法 4: ChatPromptTemplate结构化提示模板 # def demo_prompt_template(): 项目中的用法 用 ChatPromptTemplate 定义可复用的提示模板 结合 Pydantic 模型实现结构化输出。 print( * 60) print(【用法 4】ChatPromptTemplate结构化提示模板) print( * 60) from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 定义模板项目中用于字段抽取 template ChatPromptTemplate.from_messages([ (system, 你是一个 {role}。请从以下文本中提取所需信息。输出遵循规则{schema}), (human, 文本内容\n{text}\n\n请提取{field_name}), ]) # 2. 静态的固定的参数partial固化固定不变的schema要求的叫预算 fixed_template template.partial(schema{项目名称:str,预算:str,投标截止日期:str}) # 3. 动态参数每次推理只传动态变化的参数虽然抽取的叫预算金额但实际输出的是预算符合schema设定 {项目名称: 智慧园区建设项目, 预算: 500万元, 投标截止日期: 2024年12月31日} messages fixed_template.format_messages( role专业的文档解析助手, text招标项目名称智慧园区建设项目\n预算金额500万元\n投标截止日期2024年12月31日, field_name项目名称、预算金额、投标截止日期, ) llm ChatOpenAI( modelMODEL_ID, api_keyAPI_KEY, base_urlBASE_URL, temperature0.0, ) import json try: response llm.invoke(messages) dic_response json.loads(response.content) class dict print(type(dic_response)) 提取结果: { 项目名称: 智慧园区建设项目, 预算: 500万元, 投标截止日期: 2024年12月31日 } print(f 提取结果: {dic_response}\n) except Exception as e: print(f [跳过] {e}\n) # # 用法 5: MarkdownHeaderTextSplitterMarkdown 分块 # def demo_markdown_splitter(): 项目中的用法 1. 用 MarkdownHeaderTextSplitter 按标题层级宏观分块 2. 从 metadata 中提取标题链parent_title 3. 在每个块内再做精细分割图片、表格、文本 print( * 60) print(【用法 5】MarkdownHeaderTextSplitterMarkdown 按标题分块) print( * 60) from langchain_text_splitters import MarkdownHeaderTextSplitter # 模拟一段 Markdown 文档 md_document # 第一章 项目概述 ## 1.1 项目背景 本项目旨在建设一个智能化的园区管理系统。 ## 1.2 项目目标 实现园区的数字化管理。 # 第二章 技术方案 ## 2.1 系统架构 系统采用微服务架构。 ### 2.1.1 前端技术 使用 React 框架。 ### 2.1.2 后端技术 使用 Python FastAPI。 ## 2.2 部署方案 采用 Docker 容器化部署。 # 1. 配置按哪些标题层级切分 headers_to_split_on [ (#, 标题1), (##, 标题2), (###, 标题3), (####, 标题4), (#####, 标题5), (######, 标题6), ] splitter MarkdownHeaderTextSplitter(headers_to_split_onheaders_to_split_on) docs splitter.split_text(md_document) doc in docs # 代表key:标题1 ## 代表key:标题2 ### 代表key:标题3 所以使用 React 框架。这个doc doc.metadata属性 dict类型 {标题1: 第二章 技术方案, 标题2: 2.1 系统架构, 标题3: 2.1.1 前端技术} doc.page_content str类型 使用 React 框架。 def extract_parent_title(metadata: dict) - List[str]: header_keys sorted([k for k in metadata.keys() if k.startswith(标题)]) return [metadata[k] for k in header_keys] print(f 共切出 {len(docs)} 个块\n) for i, doc in enumerate(docs, 1): parent_title extract_parent_title(doc.metadata) content_preview doc.page_content[:80].replace(\n, ).strip() print(f 块 {i}:) print(f 标题链: { .join(parent_title)}) print(f 内容预览: {content_preview}...) print() # 在复杂场景下可在每个块内再做二次分割图片、表格、文本 # # 用法 6: OpenAIEmbeddings文本向量化 这个怎么用建议参考官网模型API # def demo_embeddings(): OpenAIEmbeddings print( * 60) print(【用法 6】OpenAIEmbeddings文本向量化) print( * 60) try: from openai import OpenAI client OpenAI( api_keyAPI_KEY, base_urlBASE_URL, ) texts [什么是 RAG, 检索增强生成, 今天天气怎么样] response client.embeddings.create( modeltext-embedding-v3, inputtexts, ) vectors [item.embedding for item in response.data] for text, vec in zip(texts, vectors): print(f 文本: {text}) print(f 向量维度: {len(vec)}) print(f 前 5 个值: {vec[:5]}) print() # 计算相似度 query_resp client.embeddings.create( modeltext-embedding-v3, inputRAG 技术, ) query_vec query_resp.data[0].embedding from math import sqrt def cosine_similarity(a, b): dot sum(x * y for x, y in zip(a, b)) norm_a sqrt(sum(x * x for x in a)) norm_b sqrt(sum(x * x for x in b)) return dot / (norm_a * norm_b) for text, vec in zip(texts, vectors): sim cosine_similarity(query_vec, vec) print(f RAG 技术 与 {text} 的相似度: {sim:.4f}) except Exception as e: print(f [跳过] {e}\n) print() import requests import json def demo_embeddingsV2(): all_embeddings [] headers {Content-Type: application/json} payload {model: Qwen-Embedding, input: 什么是 RAG, dimensions: 1024} response requests.post( http://127.0.0.1:38777/v1/embeddings, headersheaders, datajson.dumps(payload), timeout600 ) if 200 response.status_code: pass # # 用法 7: BaseChatModel抽象类型注解 # def demo_base_chat_model(): 项目中的用法 在抽象接口中把 BaseChatModel 作为参数类型注解 这样不管是 ChatOpenAI 还是 AzureChatOpenAI 都能传入。 体现了面向接口编程的思想。 print( * 60) print(【用法 7】BaseChatModel抽象类型注解) print( * 60) from langchain_core.language_models import BaseChatModel from abc import ABC, abstractmethod # 项目中用 BaseChatModel 做类型注解支持多态 class IFieldDiscoverer(ABC): 字段发现器抽象接口 abstractmethod async def discover(self, llm: BaseChatModel, text: str) - list: 传入任何 BaseChatModel 子类都能工作 pass # 实际使用时传入 ChatOpenAI 或 AzureChatOpenAI 都行 class SimpleDiscoverer(IFieldDiscoverer): async def discover(self, llm: BaseChatModel, text: str) - list: # llm 可以是 ChatOpenAI 或 AzureChatOpenAI # 实际调用 ... return [] print( BaseChatModel 作为抽象类型支持 ChatOpenAI 和 AzureChatOpenAI 多态传入\n) # # 主入口运行所有 Demo # if __name__ __main__: print(\n) print(╔══════════════════════════════════════════════════════════╗) print(║ LangChain 功能演示 ║) print(╚══════════════════════════════════════════════════════════╝) print() print(f 当前模型: {MODEL_ID}) print(f API 地址: {BASE_URL}) print(f API Key : {已设置 if API_KEY ! sk-xxx else 未设置请设置 API_KEY 环境变量}) print() #demo_basic_llm_call() # demo_multimodal_call() # demo_concurrency_control() # demo_prompt_template() demo_markdown_splitter() # demo_embeddings() # demo_embeddingsV2() # demo_base_chat_model() print(╔══════════════════════════════════════════════════════════╗) print(║ LangChain 功能演示结束 ║) print(╚══════════════════════════════════════════════════════════╝)