完整功能总结

一、整体概述

这份代码是一套面向企业级 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("╚══════════════════════════════════════════════════════════╝")