完整功能总结

一、整体概述

这份代码是一套面向企业级 AI 应用开发的LangChain 标准化实战 Demo 集，基于 LangChain1.x 新版本开发，兼容 DeepSeek、通义千问等全量 OpenAI 兼容接口模型，覆盖模型调用、多模态图文、并发限流、提示词工程、文档分块、文本向量化、面向接口多态7 大核心开发场景，完整覆盖 RAG 知识库、文档解析、批量任务、多模型适配等主流业务需求，可直接作为项目开发模板。

二、7 大 Demo 分项功能详解

Demo1：ChatOpenAI 基础文本调用（最核心基础能力）

1. 核心组件：ChatOpenAI、SystemMessage/HumanMessage对话消息结构
2. 功能要点
   • 统一封装兼容 OpenAI 格式的各类大模型（DeepSeek、千问等），仅需修改API_KEY/BASE_URL/MODEL_ID即可切换模型；
   • 通过 SystemMessage 固定 AI 角色、输出约束，HumanMessage 承载用户输入，标准化对话入参；
   • 配置生产级参数：temperature=0保证输出稳定、超时时间、失败自动重试；
   • 使用invoke标准同步调用，是项目 90% 文本问答、信息提取的基础写法。
3. 业务场景：文本问答、知识解释、简单文案生成。

Demo2：多模态图片识别调用

1. 核心组件：多模态ChatOpenAI、Base64 图片编码、图片 URL 入参
2. 功能要点
   • 支持两种图片传入方式：网络图片 URL、本地图片转 Base64 编码；
   • HumanMessage 支持混合图文结构体，同时传入图片 + 文本提问；
   • 调用逻辑与纯文本完全统一，仅扩展入参结构，无缝对接千问 VL 等视觉大模型。
3. 业务场景：文档截图 OCR 识别、图纸解析、图片内容描述、图文混合文档分析。

Demo3：多线程并发限流控制

1. 核心组件：threading.BoundedSemaphore信号量、多线程 Worker
2. 功能要点
   • 解决批量调用 LLM 时请求过载、接口限流报错问题；
   • 通过信号量设置全局最大并发数，同一时间仅允许 N 条 LLM 请求执行；
   • 封装安全调用函数，自动抢占 / 释放资源，增加等待超时保护；
   • 模拟多线程批量任务，适配文档批量解析、批量数据抽取场景。
3. 业务场景：大批量文档并行处理、批量实体抽取、多任务异步 AI 处理。

Demo4：ChatPromptTemplate 结构化提示词模板

1. 核心组件：ChatPromptTemplate、模板 partial 固化、结构化 JSON 输出
2. 功能要点
   • 提示词模板化、可复用，通过占位符{变量}动态传入角色、文本、抽取字段；
   • partial固化固定不变的规则 / 输出格式，仅传入动态业务数据，简化调用；
   • 强制模型输出标准 JSON 结构，配合 json.loads 直接转为字典，实现结构化信息抽取。
3. 业务场景：合同 / 文档关键字段提取、表单信息抽取、标准化结构化数据输出。

Demo5：MarkdownHeaderTextSplitter 文档分层分块

1. 核心组件：Markdown 标题分割器、Document 文档对象、metadata 元数据解析
2. 功能要点
   • 按 Markdown 一至六级标题自动切割长文档，保留每一块对应的完整标题层级；
   • 分割后的文档块自带metadata元数据，存储各级标题名称；
   • 封装工具自动提取完整标题链，记录段落所属章节，解决长文档上下文丢失问题；
   • 支持二次精细分割，是 RAG 知识库构建中文档预处理核心步骤。
3. 业务场景：技术手册、项目方案、Markdown 知识库拆分，为向量入库做预处理。

Demo6：文本向量化（Embedding）两种实现方案

1. 方案 1：OpenAI 官方兼容 Embedding 客户端
   • 直接调用模型向量接口，批量文本一次性生成向量；
   • 内置余弦相似度计算函数，快速判断文本语义相似度；
   • 适配千问文本嵌入模型，用于知识库检索匹配。
2. 方案 2：原生 requests 请求本地向量服务（embeddingsV2）
   • 不依赖 LangChain 封装，通过 HTTP 请求调用私有化部署向量服务；
   • 自定义向量维度、输入文本，适配本地私有化向量引擎。
3. 业务场景：RAG 向量库构建、文本语义检索、知识库相似度匹配。

Demo7：BaseChatModel 抽象基类（面向接口多态设计）

1. 核心组件：BaseChatModel抽象接口、抽象基类 ABC、自定义业务接口
2. 功能要点
   • LangChain 顶层抽象模型接口，所有对话模型（ChatOpenAI、Azure 模型等）均实现该接口；
   • 业务代码以BaseChatModel做类型注解，实现多态：传入任意兼容模型均可正常运行；
   • 面向接口编程，后续切换模型、新增厂商模型无需修改业务逻辑，代码解耦、扩展性强。
3. 业务场景：企业多模型混合调度、支持随时切换不同厂商大模型的通用底层服务。

三、代码整体架构与生产价值

1. 统一配置层：集中管理 API 密钥、模型地址、模型名称，环境变量优先读取，便于多环境部署；
2. 由浅入深学习顺序：基础调用→多模态→提示词模板→文档分割→向量入库→并发处理→底层抽象设计，完全贴合 RAG 应用开发全流程；
3. 企业落地优势
   • 全部代码适配生产环境，内置重试、超时、并发限流等异常保护；
   • 兼容公有云模型（DeepSeek、通义千问）与私有化本地服务；
   • 组件高度解耦，可单独抽取任意 Demo 模块集成到自有业务系统；
   • 完整覆盖 RAG 全链路：文档分割→文本向量化→向量检索→LLM 问答 / 信息抽取。

四、适用开发场景汇总

1. 知识库问答系统（RAG）完整链路开发；
2. 文档智能解析：合同、技术文档关键字自动提取；
3. 图文混合内容识别、截图 OCR 解析；
4. 大批量文档并行批量 AI 处理；
5. 支持多模型灵活切换的通用 AI 底层服务；
6. 私有化向量服务、本地大模型服务对接开发。

""" LangChain 功能演示 =================== conda create -n langchain_project python==3.11.13 conda activate langchain_project pip install langchain==1.3.9 langchain-openai==1.1.0 langchain-text-splitters==1.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( model=MODEL_ID, api_key=API_KEY, base_url=BASE_URL, temperature=0.0, # 项目里几乎都用 0.0，保证输出稳定 timeout=120, # 超时时间 max_retries=3, # 重试次数 ) """# 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( model=MODEL_ID, api_key=API_KEY, base_url=BASE_URL, temperature=0.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/u=3560029307,576412274&fm=3028&app=3028&f=JPEG&fmt=auto?w=960&h=1280"}, }, # { # "type": "image_url", # "image_url": {"url": f"data: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( model=MODEL_ID, api_key=API_KEY, base_url=BASE_URL, temperature=0.0, ) """ 核心并发控制代码 # 模拟并发控制逻辑 # 最多同时 2 个请求 """ MAX_CONCURRENCY = 2 lock = threading.BoundedSemaphore(MAX_CONCURRENCY) def safe_llm_invoke(llm, messages): """带并发控制的 LLM 调用（与项目中的 llm_invoke 一致）""" acquired = lock.acquire(timeout=10) 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(content=f"任务 {task_id}: 1+1 等于几？"), ] 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(target=worker, 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( model=MODEL_ID, api_key=API_KEY, base_url=BASE_URL, temperature=0.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: MarkdownHeaderTextSplitter（Markdown 分块） # ============================================================ """ def demo_markdown_splitter(): """ 项目中的用法： 1. 用 MarkdownHeaderTextSplitter 按标题层级宏观分块 2. 从 metadata 中提取标题链（parent_title） 3. 在每个块内再做精细分割（图片、表格、文本） """ print("=" * 60) print("【用法 5】MarkdownHeaderTextSplitter（Markdown 按标题分块）") 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_on=headers_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_key=API_KEY, base_url=BASE_URL, ) texts = ["什么是 RAG", "检索增强生成", "今天天气怎么样"] response = client.embeddings.create( model="text-embedding-v3", input=texts, ) 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( model="text-embedding-v3", input="RAG 技术", ) 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", headers=headers, data=json.dumps(payload), timeout=600 ) 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("╚══════════════════════════════════════════════════════════╝")