文章目录

1.代码调用Qwen云端大模型

  • 在Qwen：https://bailian.console.aliyun.com/cn-beijing?tab=model#/api-key 创建apikey
  • 通过pip为python程序提供openAI库：pip install openai
  • 通过代码测试（可从Qwen官网获取测试代码）

2.OpenAI库的基础使用

OpenAI 库是 OpenAI 官方推出的 Python SDK，核心作用是让开发者能简单、高效地调用 OpenAI 的各类 API（如 GPT 聊天、DALL・E 绘图、语音转文字等），无需手动处理 HTTP 请求、身份验证等底层细节。

由于其发布较早且比较易用，现如今许多模型服务商（如阿里云百炼平台）均兼容 OpenAI SDK 的调用。

2.1 获取客户端对象

from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://llm-elbcdb80b00imo4b.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)

主要用如上2个参数：

  • api_key：模型服务商提供的APIKEY密钥
  • base_url：模型服务商的API接入地址
    • 主要基于此参数来切换不同的模型服务商（如openAI、阿里云、腾讯云）

2.2 调用模型

response = client.chat.completions.create(
    model="qwen3.7-max",
    messages=[
        {"role": "system", "content": "你是一个python编程专家"},
        {"role": "assistant", "content": "我是一个python编程专家，请问有什么能帮助你的吗"},
        {"role": "user", "content": "帮我用for循环打印1到5"},
    ]
)

client.chat.completions.createh创建ChatCompletion对象

主要参数：

  • model：选择所用的模型
  • messages：提供给模型的消息，类型：list，可以包含多个字典信息，每个字典信息包含2个key（role、content）
    • system：全局人设、规则、约束，模型全程遵循。
    • assistant：模型上一轮输出的历史回复，用来拼接完整对话上下文，让模型记住之前聊了什么。
    • user：本次用户最新提问：for循环出1到5的数字，是模型当前要处理的需求。
  • stream：流式输出开关
  • extra_body：拓展参数，{“thinking_enable”: True} 是Qwen特有扩展参数，开启深度思考展示

2.3 获取结果

print(response.choices[0].message.content)

3.OpenAI的流式输出

#开启流式输出
stream=True

# 判空
for chunk in response:
    if not chunk.choices:
        continue
    print(chunk.choices[0].delta.content, end="", flush=True)

4.OpenAI库附带历史消息调用模型

在调用模型时通过在messages的list内，组织历史消息提供给模型。

5.大模型prompt工程指南

5.1 提示词概念

提示工程（Prompt Engineering），也称为 In-Context Prompting，是指在不更新模型权重的情况下如何与大模型交互以引导其行为以获得所需结果的方法。

在人工智能领域，Prompt 指的是用户给大型语言模型发出的指令。

  • 例如，“讲个笑话”、“用 Python 编个贪吃蛇游戏”、“写封情书” 等。
  • 虽然看似简单，但实际上，Prompt 的设计对于模型的结果影响很大。
  • 因此如何设计 prompt，进而与模型更好的交互，是研究人员必备的必不可少的技能（提示工程）。

5.2 提示词技巧

1. 详细的描述

摒弃模糊简短指令，把需求、细节、边界、目标完整写清楚。

模糊示例：帮我写 Python 代码；

详细示例：写一段 Python for 循环代码，打印 1 到 5 的数字，附带单行注释，代码可直接运行，不添加多余文字说明。

2. 让模型充当角色

通过 system 提示给模型固定专业身份，限定输出风格、专业视角。

示例：你是资深 Python 编程专家，回答简洁专业，只输出完整可运行代码，不闲聊、不拓展无关内容。

3. 使用分隔符区分输入内容

用引号、、---等符号分割指令、参考素材、用户问题，避免文本混淆，让模型区分不同模块信息。 示例： 用户问题总共有几个宠物？历史数据 ```

小明 2 只狗，小红 3 只猫

4. 对任务指定步骤

要求模型分步拆解、按流程完成任务，降低逻辑、计算类错误。

示例：请分两步回答，第一步统计所有宠物数量，第二步输出最终数字，不要直接给出答案。

5. 提供示例

输入 1 组或多组标准「输入 - 标准答案」样例，引导模型模仿固定格式、逻辑输出。

示例：

例子 1：

用户：2 只狗 + 1 只猫

答案：3 只

请参照上面格式回答：2 只狗 + 3 只猫

5.3 提示词优化案例介绍

1.实战案例背景

当前金融领域信息化发展的时代，金融数据大量激增，许多投资者和研究者试图通过对这些数据进行深度分析而获得一些有效的决策和帮助，尽可能减少决策失误带来的损失。

所以，针对金融数据的分析方法研究是目前十分有益且热门的话题。

当前案例主要有三大业务场景实现：

  • 基于大模型完成：金融文本分类
  • 基于大模型完成：金融文本信息抽取
  • 基于大模型完成：金融文本匹配

大模型选择：Deepseek 在线大模型（deepseek-v4-pro）

采用方法：基于Few-Shot + Zero-Shot的思想，设计prompt（提示词），进而应用大模型完成相应的任务。

2.Zero-Shot思想（零样本）

在这里插入图片描述

3.Few-Shot思想（少样本）

在这里插入图片描述

4.金融文本分类-实战案例

prompt设计：

对于大模型来讲，prompt 的设计非常重要，一个明确的 prompt 能够帮助我们更好从大模型中获得我们想要的结果。

在该任务的 prompt 设计中，我们主要考虑 2 点：

  • 需要向模型解释什么叫作「文本分类任务」
  • 需要让模型按照我们指定的格式输出

为了让模型知道什么叫做「文本分类」，我们借用 FewShot 的方式，给模型展示一些正确的例子：

User: “今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。” 是 [’ 新闻报道 ', ’ 公司公告 ', ’ 财务公告 ‘,’ 分析师报告 '] 里的什么类别？

Bot: 新闻报道

User: “本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。” 是 [’ 新闻报道 ', ’ 公司公告 ', ’ 财务公告 ‘,’ 分析师报告 '] 里的什么类别？

Bot: 财务报告

其中，User 代表我们输入给模型的句子，Bot 代表模型的回复内容。

注意：上述例子中 Bot 的部分也是由人工输入的，其目的是希望看到在看到类似 User 中的句子时，模型应当做出类似 Bot 的回答。

import os
from openai import OpenAI

# 获取client对象，OpenAI类对象
client = OpenAI(
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://llm-elbcdb80b00imo4b.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)

# 示例数据
example_data = {
    "新闻报道": "今日 A 股大盘小幅收涨，消费板块全线走高，国际原油价格受中东局势影响出现短期波动，多家机构预判下周市场流动性将持续宽松。",
    "财务报告": "2025 年度年报披露：公司全年营收 126.8 亿元，同比增长 8.2%，归母净利润 15.3 亿元，资产负债率维持 42%，经营性现金流净额同比提升 11%。",
    "公司公告": "本公司于 2026 年 7 月 5 日召开董事会，审议通过高管人事调整议案，聘任张某为首席财务官，相关任免文件将于三日内对外公示。",
    "分析师报告": "行业调研结论：储能行业未来三年复合增速有望达 25%，龙头企业具备成本壁垒，建议中长期持有板块头部标的，目标估值上调 18%。",
}

# 分类列表
example_types = ["新闻报道", "财务报告", "公司公告", "分析师报告"]

# 问题
questions = [
    "本地财经快讯：央行宣布下调中期借贷便利利率 0.1 个百分点，将直接降低企业融资成本，如何归类？",
    "2025 全年财报显示企业净利润同比下滑 6%，存货周转天数上升，该文本属于哪一类别？",
    "上市公司发布董事会决议，计划新增两条生产线，此内容对应哪类文档？",
    "券商研报指出新能源车企毛利率将持续修复，推荐布局整车龙头，判断文本类型。",
    "周末去哪里爬山风景更好，推荐一条短途徒步路线？",
]

# 构建提示词
messages = [
    {"role": "system", "content": "你是金融专家，熟悉文本分类['新闻报道', '财务报告', '公司公告', '分析师报告']，如果不知道的就说不清楚,知道的直接输出类别"}
]
for key, value in example_data.items():
    messages.append({"role": "user", "content": value})
    messages.append({"role": "assistant", "content": key})


# 处理结果
# print(response.choices[0].message.content)
for v in questions:
    messages.append({"role": "user", "content": v})
    print(v)
    # 调用模型
    response = client.chat.completions.create(
        model="qwen3.7-max",
        messages=messages,
        stream=False  # 开启了流式输出的功能
    )
    print(response.choices[0].message.content)
    messages.pop()
5.JSON数据格式(结构化)

JSON 全称 JavaScript Object Notation（JavaScript 对象标记语言），是一种轻量级、纯文本、跨语言通用的数据交换格式，独立于编程语言，用于程序之间传递、存储结构化数据。

JSON有2种结构：json对象和json数组

Json对象：

{
    "key": "value",
    "key": value
}
#key必须是字符串
#value可以是：数字、字符串、列表、Json对象或Json数组

Json数组：

[
    {"key": "value"},
    {"key": "value"}
]
#一堆Json对象的组合体

Python 中使用 Json 主要完成：

  • 将 Python 字典、列表转换为 Json 字符串
  • 读取 Json 字符串，转换为 Python 字典或列表

主要使用 Python 内置的 json 库

  • json.dumps(字典或列表, ensure_ascii=False)

    ：将字典或列表转换为 Json 字符串

    • ensure_ascii 参数确保中文能正常显示
    • 返回值：Json 字符串

  • json.loads(json字符串)

    ：将 Json 字符串转换为 Python 字典或列表

    • 返回值：Python 字典 或 Python 列表
6.金融文本信息抽取-实战案例

prompt设计：

在该任务的 prompt 设计中，我们主要考虑 2 点：

  • 需要向模型解释什么叫作「信息抽取任务」
  • 需要让模型按照我们指定的格式（json）输出

为了让模型知道什么叫做「信息抽取」，我们借用 FewShot 的方式，先给模型展示几个正确的例子：

User:‘2023-01-10，股市震荡。股票古哥 - D [EOOE] 美股今日开盘价 100 美元，一度飙升至 105 美元，随后回落至 98 美元，最终以 102 美元收盘，成交量达到 520000。’。提取上述句子中 “金融” (’ 日期 ‘,’ 股票名称 ‘,’ 开盘价 ‘,’ 收盘价 ‘,’ 成交量 ‘) 类型的实体，并按照 JSON 格式输出，上述句子中没有的信息用 [’ 原文中未提及 '] 来表示，多个值之间用 ‘,’ 分隔。

Bot: {’ 日期 ‘: [‘2023-01-10’],’ 股票名称 ‘: [’ 古哥 - D [EOOE] 美股 ‘],’ 开盘价 ': ['100 美元 '], ’ 收盘价 ': ['102 美元 '], 成交量 ': [‘520000’]} …

其中，User 代表我们输入给模型的句子，Bot 代表模型的回复内容。

注意：上述例子中 Bot 的部分也是由人工输入的，其目的是希望看到在看到类似 User 中的句子时，模型应当做出类似 Bot 的回答。

import json
import os
from openai import OpenAI

# 获取client对象，OpenAI类对象
client = OpenAI(
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://llm-elbcdb80b00imo4b.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)

schema = ['日期', '股票名称', '开盘价', '收盘价', '成交量']
examples_data = [  # 示例数据
    {
        "content": "2023-01-10，股市震荡。股票强大科技A股今日开盘价100人民币，一度飙升至105人民币，随后回落至98人民币，最终以102人民币收盘，成交量达到520000。",
        "answers": {
            "日期": "2023-01-10",
            "股票名称": "强大科技A股",
            "开盘价": "100人民币",
            "收盘价": "102人民币",
            "成交量": "520000"
        }
    },
    {
        "content": "2024-05-16，股市利好。股票英伟达美股今日开盘价105美元，一度飙升至109美元，随后回落至100美元，最终以116美元收盘，成交量达到3560000。",
        "answers": {
            "日期": "2024-05-16",
            "股票名称": "英伟达美股",
            "开盘价": "105美元",
            "收盘价": "116美元",
            "成交量": "3560000"
        }
    }
]

questions = [  # 提问问题
    "2025-06-16，股市利好。股票传智教育A股今日开盘价66人民币，一度飙升至70人民币，随后回落至65人民币，最终以68人民币收盘，成交量达到123000。",
    "2025-06-06，股市利好。股票黑马程序员A股今日开盘价200人民币，一度飙升至211人民币，随后回落至201人民币，最终以206人民币收盘。"
]

messages = [
    {"role": "system", "content": f"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未描述"}
]

for exmaple in examples_data:
    messages.append({"role": "user", "content": exmaple['content']})
    messages.append({"role": "assistant", "content": json.dumps(exmaple['answers'], ensure_ascii=False)})


for q in questions:
    response = client.chat.completions.create(
        model="qwen3.7-max",
        messages=messages + [{"role": "user", "content": q}],
        stream=False  # 开启了流式输出的功能
    )
    print(q)
    print(response.choices[0].message.content)
7.金融文本匹配-实战案例

prompt设计：

在该任务的 prompt 设计中，我们主要考虑 2 点：

  • 需要向模型解释什么叫作「文本匹配任务」
  • 需要让模型按照我们指定的格式输出

为了让模型知道什么叫做「文本匹配任务」，我们借用 FewShot 的方式，先给模型展示几个正确的例子：

User:

句子一：公司 ABC 发布了季度财报，显示盈利增长。

句子二：财报披露，公司 ABC 利润上升

Bot: 是

User:

句子一：黄金价格下跌，投资者抛售。

句子二：外汇市场交易额创下新高

Bot: 不是 …

其中，User 代表我们输入给模型的句子，Bot 代表模型的回复内容。

注意：上述例子中 Bot 的部分也是由人工输入的，其目的是希望看到在看到类似 User 中的句子时，模型应当做出类似 Bot 的回答。

import os
from openai import OpenAI

# 获取client对象，OpenAI类对象
client = OpenAI(
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://llm-elbcdb80b00imo4b.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)

examples_data = {
    "是": [
        ("公司ABC发布了季度财报，显示盈利增长。", "财报披露，公司ABC利润上升。"),
        ("公司ITCAST发布了年度财报，显示盈利大幅度增长。", "财报披露，公司ITCAST更赚钱了。")
    ],
    "不是": [
        ("黄金价格下跌，投资者抛售。", "外汇市场交易额创下新高。"),
        ("央行降息，刺激经济增长。", "新能源技术的创新。")
    ]
}

questions = [
    ("利率上升，影响房地产市场。", "高利率对房地产有一定的冲击。"),
    ("油价大幅度下跌，能源公司面临挑战。", "未来智能城市的建设趋势越加明显。"),
    ("股票市场今日大涨，投资者乐观。", "持续上涨的市场让投资者感到满意。")
]

"""
{"role": "system",      "content": f"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："},
{"role": "user",       "content": "句子1：[公司ABC发布了季度财报，显示盈利增长。]句子2：[财报披露，公司ABC利润上升。]"},
{"role": "assistant",   "content": "是"},
{"role": "user",       "content": "句子1：[公司ITCAST发布了年度财报，显示盈利大幅度增长。]句子2：[财报披露，公司ITCAST更赚钱了。]"},
{"role": "assistant",   "content": "是"},
{"role": "user",       "content": "句子1：[黄金价格下跌，投资者抛售。]句子2：[外汇市场交易额创下新高。]"},
{"role": "assistant",   "content": "不是"},
{"role": "user",       "content": "句子1：[央行降息，刺激经济增长。]句子2：[新能源技术的创新。]"},
{"role": "assistant",   "content": "不是"},

{"role": "user",       "content": f"按照上述示例，回答这2个句子的情况。句子1：[...]，句子2：[...]"}
"""

messages = [
    {"role": "system", "content": f"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断它们是否匹配，回答是或不是，请参考如下示例："}
]

for key, value in examples_data.items():
    for x in value:
        messages.append({"role": "user", "content": f"句子1：[{x[0]}]，句子2：[{x[1]}]"})
        messages.append({"role": "assistant", "content": key})

    for q in questions:
        response = client.chat.completions.create(
            model="qwen3.7-max",
            messages=messages + [{"role": "user", "content": f"句子1：[{q[0]}]，句子2：[{q[1]}]"}],
            stream=False
        )
        print(q)
        print(response.choices[0].message.content)

6.RAG

6.1 通用基础大模型存在的问题

  1. LLM 的知识不是实时的，模型训练好后不具备自动更新知识的能力，会导致部分信息滞后。（知识过时）
  2. LLM 领域知识是缺乏的，大模型的知识来源于训练数据，这些数据主要来自公开的互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识。（领域知识缺乏）
  3. 幻觉问题，LLM 有时会在回答中生成看似合理但实际上是错误的信息。（幻觉）
  4. 数据安全性。（数据安全）

在这里插入图片描述

6.2 RAG工作原理

在这里插入图片描述

简单来说，RAG工作分为两条线：离线准备线/在线准备线

在这里插入图片描述

RAG 标准流程由索引（Indexing）、检索（Retriever）和生成（Generation）三个核心阶段组成。

一、索引阶段

通过处理多种来源多种格式的文档提取其中文本，将其切分为标准长度的文本块（chunk），并进行嵌入向量化（embedding），向量存储在向量数据库（vector database）中。

  1. 加载文件
  2. 内容提取
  3. 文本分割，形成 chunk
  4. 文本向量化
  5. 存入向量数据库

二、检索阶段

用户输入的查询（query）被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块。

  1. query 向量化
  2. 在文本向量中匹配出与问句向量相似的 top_k 个

三、生成阶段

检索到的相关文本与原始查询共同构成提示词（Prompt），输入大语言模型（LLM），生成精确且具备上下文关联的回答。

  1. 匹配出的文本作为上下文和问题一起添加到 prompt 中
  2. 提交给 LLM 生成答案

RAG的核心价值就是解决大模型存在的：领域知识缺乏、知识实时性、降低幻觉、无需重新训练模型

7.向量的基础概念

RAG 流程中，向量库是一个重要的节点。

  • 离线流程：知识和信息 → 向量嵌入（向量化）→ 存入向量库
  • 在线流程：用户的提问 → 向量嵌入（向量化）→ 在向量库中匹配

在这里插入图片描述

7.1 基础概念：

向量就是文本的 “数学身份证”：它把一段文字的语义信息，转换成一串固定长度的数字列，让计算机能 “看懂” 文字的含义并做相似度计算。

简单来说，就是让计算机更方便的理解不同的文本内容，是否表述的是一个意思。

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

总结：

向量（Vector）就是文本的 “数学身份证”

它把一段文字的语义信息，转换成一串固定长度的数字列表，让计算机能 “看懂” 文字的含义并做相似度计算。

  • 向量的计算（文本嵌入过程），可借助文本嵌入模型实现，如 text-embedding-v1
  • 向量的匹配通过算法实现，如余弦相似度
  • 向量的维度表示一段文本在多个抽象语义特征方面的强度
    • 维度数代表模型用多少个抽象语义特征来描述文本
    • 维度越多，做语义匹配越精准
    • 但性能压力也会增大

7.2 余弦相似度算法（扩展）

在这里插入图片描述

在这里插入图片描述

8.Langchain

8.1 简介

LangChain 由 Harrison Chase 创建于 2022 年 10 月，它是围绕 LLMs（大语言模型）建立的一个框架。

LangChain 自身并不开发 LLMs，它的核心理念是为各种 LLMs 实现通用的接口，把 LLMs 相关的组件 “链接” 在一起，简化 LLMs 应用的开发难度，方便开发者快速地开发复杂的 LLMs 应用。

在这里插入图片描述

8.2 安装

pip install langchain langchain-community langchain-ollama dashscope chromadb

  • langchain：核心包

  • langchain-community：社区支持包，提供了更多的第三方模型调用。（已废弃）

  • langchain-ollama：Ollama 支持包，支持调用 Ollama 托管部署的本地模型，不需要可以不安装

  • dashscope：阿里通义千问的python sdk包。（未过期，官方持续更新）

  • chromadb：轻量向量数据库（后续使用）

9.Models（模型）

现在市面上的模型多如牛毛，各种各样的模型不断出现，LangChain 模型组件提供了与各种模型的集成，并为所有模型提供一个精简的统一接口。

LangChain 目前支持三种类型的模型：LLMs（大语言模型）、Chat Models (聊天模型)、Embeddings Models (嵌入模型)。

  • LLMs：是技术范畴的统称，指基于大参数量、海量文本训练的 Transformer 架构模型，核心能力是理解和生成自然语言，主要服务于文本生成场景。
  • 聊天模型：是应用范畴的细分，是专为对话场景优化的 LLMs，核心能力是模拟人类对话的轮次交互，主要服务于聊天场景。
  • 文本嵌入模型：文本嵌入模型接收文本作为输入，得到文本的向量。

LangChain 支持的三类模型，它们的使用场景不同，输入和输出不同，开发者需要根据项目需要选择相应。

通义千问系列主要来自于：langchain-community包（已废弃），改用：langchain_openai

pip install langchain_openai

9.1 langchain调用大语言模型

#Qwen云端大语言模型不直接提供调用，Tongyi包已过期，如下替换方案：
import os
# 替换废弃的 Tongyi
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.7,
    # 可选：开启通义深度思考，等价原生dashscope能力
    extra_body={
        "enable_thinking": False
    }
)

# 完全和你原来一样：直接传字符串调用
res = llm.invoke("你是谁？你能做什么")
print(res.content)

9.2 langchain调用聊天模型

import os
from langchain_community.chat_models.tongyi import ChatTongyi  #废弃警告
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatTongyi(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    stream=True
)
# 消息
# messages = [
#     SystemMessage("你是一个边塞诗人"),
#     HumanMessage("写一首唐诗"),
#     AIMessage("锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦"),
#     HumanMessage("根据上一个回复的格式，再写一首唐诗。")
# ]

# 消息简写
messages = [
    ("system", "你是一个边塞诗人"),
    ("human", "写一首唐诗"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦"),
    ("human", "根据上一个回复的格式，再写一首唐诗。")
]
# invoke 一次性返回完整结果
# res = llm.invoke(messages)
# print(res.content)

# stream 逐段流式输出
res = llm.stream(messages)
for chunk in res:
    print(chunk.content, end="", flush=True)

上述方式已废弃，下面为替换方案：

import os
# 替换废弃 ChatTongyi
from langchain_openai import ChatOpenAI
# 消息类可以继续保留，元组简写完全兼容
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get("qwen-api-key"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    temperature=0.7,
    # 通义特有深度思考参数，对应原ChatTongyi能力
    extra_body={
        "enable_thinking": False
    }
)

# 【完全不变】元组简写消息列表，langchain原生支持 ("system"/"human"/"ai", 内容)
messages = [
    ("system", "你是一个边塞诗人"),
    ("human", "写一首唐诗"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦"),
    ("human", "根据上一个回复的格式，再写一首唐诗。")
]

# 一次性完整返回（和原来用法完全一样）
# res = llm.invoke(messages)
# print(res.content)

# 流式输出，直接取 chunk.content
res = llm.stream(messages)
for chunk in res:
    # chunk 是 AIMessageChunk，直接 .content
    if chunk.content:
        print(chunk.content, end="", flush=True)

消息简写的区别及好处：

在这里插入图片描述

9.3 langchain调用嵌入模型

Embeddings Models 嵌入模型的特点：将字符串作为输入，返回一个浮点数的列表（向量）。

在 NLP 中，Embedding 的作用就是将数据进行文本向量化。

import os
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-v4",  # 通义官方云端嵌入模型
    api_key=os.environ.get("qwen-api-key"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    check_embedding_ctx_length=False  # 核心修复，避免token数组报错
)

# 单文本向量化（query）
query_vec = embeddings.embed_query("什么是RAG检索增强生成")
print(f"向量维度：{len(query_vec)}")
print(query_vec)

# 批量文档向量化（documents）
docs = [
    "嵌入模型把文本转换为浮点向量，用于相似度检索",
    "RAG分为索引、检索、生成三个核心阶段"
]
doc_vecs = embeddings.embed_documents(docs)
print(f"批量向量数量：{len(doc_vecs)}")
print(doc_vecs)

10.Prompt（提示词模板）

10.1 langchain的PromptTemplate模板

提示词优化在模型应用中非常重要，LangChain 提供了 PromptTemplate 类，用来协助优化提示词。

PromptTemplate 表示提示词模板，可以构建一个自定义的基础提示词模板，支持变量的注入，最终生成所需的提示词。

标准写法：

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

llm = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
#定义模板
prompt_template = PromptTemplate.from_template(
     "我的邻居姓{lastname}，刚生了{gender}，帮忙起名字，仅返回一个。"
)
#给模板注入变量
prompt_text = prompt_template.format(lastname="林", gender="女儿")
res = llm.invoke(prompt_text)
print(res.content)

基于chain链的写法：

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

llm = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
#定义模板
prompt_template = PromptTemplate.from_template(
     "我的邻居姓{lastname}，刚生了{gender}，帮忙起名字，仅返回一个。"
)
chain = prompt_template | llm
res = chain.invoke({"lastname": "林", "gender": "女儿"})
print(res.content)

基于 PromptTemplate 类可以得到提示词模板，支持基于模板注入变量得到最终提示词。

  • zero-shot（零样本）思想下，可以基于 PromptTemplate 直接完成。仅任务描述 + 本次动态变量，无示范案例。
  • few-shot（少样本）思想下，需要更换为 FewShotPromptTemplate。任务描述 + 多组示范样例 + 本次动态变量。

10.2 langchain的FewShotPromptTemplate模板

# 少样本的提示模板
from langchain_core.prompts import import FewShotPromptTemplate
FewShotPromptTemplate(
    examples=None,
    example_prompt=None,
    prefix=None,
    suffix=None,
    input_variables=None
)

参数：

  • examples：示例数据，list，内套字典
  • example_prompt：示例数据的提示词模板
  • prefix：组装提示词，示例数据前内容
  • suffix：组装提示词，示例数据后内容
  • input_variables：列表，注入的变量列表
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate

llm = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 提示词模板
example_template = PromptTemplate.from_template("单词：{word}， 反义词：{antonym}")

# 示例数据
example_data = [
    {"word": "大", "antonym": "小"},
    {"word": "上", "antonym": "下"},
]

few_shot_prompt = FewShotPromptTemplate(
    example_prompt=example_template,
    examples=example_data,
    prefix="给出给定词的反义词，有如下示例：",
    suffix="基于示例告诉我：{input_word} 的反义词?",
    input_variables=['input_word']
)

# 获得最终提示词
prompt_text = few_shot_prompt.invoke({"input_word": "左"}).to_string()
print(prompt_text)

res = llm.invoke(prompt_text)
print(res.content)

10.3 模板类的invoke和format方法

在 PromptTemplate（通用提示词模板）和 FewShotPromptTemplate（FewShot 提示词模板）的使用中，我们使用了如下：

模板对象的 format 方法：

prompt_template = PromptTemplate.from_template(
    "我的邻居姓{lastname}，刚生了{gender}，你帮我起个名字，简单回答。"
)
# 调用.format方法注入信息即可
prompt_text = prompt_template.format(lastname="张", gender="女儿")

model = Tongyi(model="qwen-max")
res = model.invoke(input=prompt_text)
print(res)

模板对象的 invoke 方法：

few_shot_template = FewShotPromptTemplate(
    example_prompt=example_template,    # 示例数据的模板
    examples=examples_data,             # 示例的数据（用来注入动态数据的），list内套
    prefix="告知我单词的反义词，我提供如下的示例：",  # 示例之前的提示
    suffix="基于前面的示例告知我，{input_word}的反义词是？", # 示例之后的提示
    input_variables=['input_word']       # 声明在前缀或后缀中所需要注入的变量名
)
prompt_text = few_shot_template.invoke(input={"input_word": "左"}).to_string()
print(prompt_text)

PromptTemplate、FewShotPromptTemplate、ChatPromptTemplate（后续学习）都拥有 format 和 invoke 这 2 类方法。

在这里插入图片描述

format和invoke的区别：

在这里插入图片描述

10.4 langchain的ChatPromptTemplate模板

  • PromptTemplate：通用提示词模板，支持动态注入信息。

  • FewShotPromptTemplate：支持基于模板注入任意数量的示例信息。

  • ChatPromptTemplate：支持注入任意数量的历史会话信息。

通过from_messages方法，从列表中获取多轮次会话作为聊天的基础模板

  • PS：前面 PromptTemplate | FewShotPromptTemplate 类用的from_template仅能接入一条消息，而from_messages可以接入一个 list 的消息。
  • (PromptTemplate | FewShotPromptTemplate).from_template()：只能写单段纯文本，无角色区分，适合单轮零样本任务
  • ChatPromptTemplate.from_messages()：接收消息元组列表，区分system/ai/human多角色，专门做多轮对话、携带历史聊天记录。
from langchain_core.prompts import ChatPromptTemplate

ChatPromptTemplate.from_messages(
    [
        ("system", "………"),
        ("ai", "………"),
        ……,
        ("human", "………")
    ]
)

历史会话信息并不是静态的（固定的），而是随着对话的进行不停地积攒，即动态的。

所以，历史会话信息需要支持动态注入。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts import MessagesPlaceholder

chat_template = ChatPromptTemplate.from_messages(
    [
        ("system", "………"),
        ("ai", "………"),
        MessagesPlaceholder("history"),
        ("human", "………")
    ]
)

history_data = [
    ("human", "..."),
    ("ai", "..."),
    ("human", "..."),
    ("ai", "...")
]

prompt_value = chat_template.invoke({"history": history_data})

MessagesPlaceholder 作为占位，提供 history 作为占位的 key，基于 invoke 动态注入历史会话记录，必须是 invoke，format 无法注入

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

chat_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个伟大的诗人"),
        MessagesPlaceholder("history"),
        ("human", "请再写一首唐诗")
    ]
)

history_data = [
    ("human", "你来写一首唐诗"),
    ("ai", "大漠孤烟直，长河落日圆。"),
    ("human", "好诗，再来一个"),
    ("ai", "春风又绿江南岸，明月何时照我还。")
]

prompt_text = chat_template.invoke({"history": history_data})
print(prompt_text)
res = llm.invoke(prompt_text)
print(res.content)

11.chain链

11.1 基础使用

「将组件串联，上一个组件的输出作为下一个组件的输入」是 LangChain 链（尤其是 | 管道链）的核心工作原理，这也是链式调用的核心价值：实现数据的自动化流转与组件的协同工作，如下。

chain = prompt_template | model

核心前提：只有Runnable子类对象才能入链（Callable、Mapping 接口子类对象也可加入，后续使用场景不多）。

我们目前所学习到的组件，均是 Runnable 接口的子类，如下类的继承关系：

在这里插入图片描述

在这里插入图片描述

11.2 | 运算符重载（扩展）

暂时

11.3 Runnable接口

LangChain 中的绝大多数核心组件都继承了 Runnable 抽象基类（位于 langchain_core.runnables.base）。

# 代码
chain = prompt | model

chain 变量是RunnableSequence（RunnableSerializable子类）类型

而得到这个类型的原因就是 Runnable 基类内部对__or__魔术方法的改写。

同时，在后面继续使用|添加新的组件，依旧会得到RunnableSequence，这就是链的基础架构。

def __or__(
    self,
    other: Runnable[Output, Other]
    | Callable[[Iterator[Output]], Iterator[Other]]
    | Callable[[AsyncIterator[Output]], AsyncIterator[Other]]
    | Callable[[Output], Other]
    | Mapping[str, Runnable[Output, Any] | Callable[[Output], Any] | Any],
) -> RunnableSerializable[Input, Any]:
    """Runnable "or" operator.

        Compose this `Runnable` with another object to create a
        `RunnableSequence`.

        Args:
            other: Another `Runnable` or a `Runnable`-like object.

        Returns:
            A new `Runnable`.
        """
    return RunnableSequence(self, coerce_to_runnable(other))

11.4 langchain组件StrOutputParser解析器

有如下代码，想要以第一次模型的输出结果，第二次去询问模型：

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate


model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告知名字无需其它内容"
)

chain = prompt | model | model
res = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res.content)

  • 链的构建完全符合要求（参与的组件）

  • 但是运行报错

    报错信息：ValueError: Invalid input type <class ‘langchain_core.messages.ai.AIMessage’>. Must be a PromptValue, str, or list of BaseMessages.

chain = prompt | model | model

错误的主要原因：

  • prompt的结果是promptValue类型，输入给了model
  • model的输出结果是：AIMessage

模型：ChatOpenAI --> BaseChatModel 源码中关于input的类型：

@override
    def invoke(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AIMessage:
        config = ensure_config(config)
        return cast(
            "AIMessage",
            cast(
                "ChatGeneration",
                self.generate_prompt(
                    [self._convert_input(input)],
                    stop=stop,
                    callbacks=config.get("callbacks"),
                    tags=config.get("tags"),
                    metadata=config.get("metadata"),
                    run_name=config.get("run_name"),
                    run_id=config.pop("run_id", None),
                    **kwargs,
                ).generations[0][0],
            ).message,
        )

        def _convert_input(self, model_input: LanguageModelInput) -> PromptValue:
            if isinstance(model_input, PromptValue):
                return model_input
            if isinstance(model_input, str):
                return StringPromptValue(text=model_input)
            if isinstance(model_input, Sequence):
                return ChatPromptValue(messages=convert_to_messages(model_input))
            msg = (  # type: ignore[unreachable]
                f"Invalid input type {type(model_input)}. "
                "Must be a PromptValue, str, or list of BaseMessages."
            )
            raise ValueError(msg)

需要做类型转换，可以借助Langchain内置的解析器：StrOutputParser字符串输出解析器。

StrOutputParser 是 LangChain 内置的简单字符串解析器

  • 可以将 AIMessage 解析为简单的字符串，符合了模型 invoke 方法要求（可传入字符串，不接收 AIMessage 类型）
  • 是 Runnable 接口的子类（可以加入链）
parser = StrOutputParser()
chain = prompt | model | parser | model

import os
from langchain_core.messages import AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
print(type(model))

prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告知名字无需其它内容"
)

chain = prompt | model | parser | model

res: AIMessage = chain.invoke({"lastname": "林", "gender": "女儿"})
print(res.content)

11.5 langchain组件的JsonOutputParser解析器

chain = prompt | model | parser | model | parser

在前面我们完成了这样的需求去构建多模型链，不过这种做法并不标准，因为：

上一个模型的输出，没有被处理就输入下一个模型。

正常情况下我们应该有如下处理逻辑：

invoke | stream 初始输入 → 提示词模板 → 模型 → 数据处理 → 提示词模板 → 模型 → 解析器 → 结果

即：

  • 上一个模型的输出结果，应该作为提示词模版的输入，构建下一个提示词，用来二次调用模型。

所以，我们需要完成：

将模型输出的 AIMessage → 转为字典 → 注入第二个提示词模板中，形成新的提示词 (PromptValue 对象)，由于StrOutputParser解析器无法完成，所有需要JsonOutputParser解析器。

import os
from langchain_core.messages import AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser

json_parser = JsonOutputParser()
str_parser = StrOutputParser()
model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
print(type(model))

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，并封装JSON格式返回给我，"
    "要求key是name，value就是起的名字，请严格遵守格式要求"
)

second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

chain = first_prompt | model | json_parser | second_prompt | model | str_parser

res: str = chain.invoke({"lastname": "林", "gender": "女儿"})
print(res)

11.6 自定义函数加入chain链

RunnableLambda 类是 LangChain 内置的，将普通函数等转换为 Runnable 接口实例，方便自定义函数加入 chain。

语法：

RunnableLambda(函数对象或lambda匿名函数)

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

str_parser = StrOutputParser()
model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
print(type(model))

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告诉我名字，不要额外信息"
)

second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

# 自定义处理函数
def myFun(input_text):
    return {"name": input_text}
# 将自定义函数包装为Runnable组件
my_func = RunnableLambda(myFun)

# 构建执行链
chain = first_prompt | model | my_func | second_prompt | model | str_parser

#交给模型执行
res: str = chain.invoke({"lastname": "林", "gender": "女儿"})
print(res)

chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser

跳过RunnableLambda类，直接让函数加入链也是可以的。

因为Runnable接口类在实现__or__的时候，支持Callable接口的实例。

  • 函数就是Callable接口的实例
# `__or__`方法源码片段
def __or__(
    self,
    other: Runnable[Any, Other]
        | Callable[[Iterator[Any]], Iterator[Other]]
        | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
        | Callable[[Any], Other]
        | Mapping[str, Runnable[Any, Other]] | Callable[[Any], Other] | Any,
) -> RunnableSerializable[Input, Other]:

如上代码示例，|符号（底层是调用__or__）组链，是支持函数加入的。

其本质是将函数自动转换为 RunnableLambda

import os

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

str_parser = StrOutputParser()
model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
print(type(model))

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告诉我名字，不要额外信息"
)

second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

# 自定义处理函数
def myFun(input_text):
    return {"name": input_text}
# 将自定义函数包装为Runnable组件
#my_func = RunnableLambda(myFun)

# 构建执行链
#chain = first_prompt | model | my_func | second_prompt | model | str_parser

# 不使用RunnableLambda包装
#chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser
chain = first_prompt | model | (myFun) | second_prompt | model | str_parser

#交给模型执行
res: str = chain.invoke({"lastname": "林", "gender": "女儿"})
print(res)

12.Memory记忆

12.1 临时记忆

如果想要封装历史记录，除了自行维护历史消息外，也可以借助 LangChain 内置的历史记录附加功能。

LangChain 提供了 History 功能，帮助模型在有历史记忆的情况下回答。

  • 基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链（新 Runnable 实例）
  • 基于InMemoryChatMessageHistory为历史记录提供内存存储（临时用）
from langchain_core.runnables.history import RunnableWithMessageHistory

# 通过RunnableWithMessageHistory获取一个新的带有历史记录功能的chain
conversation_chain = RunnableWithMessageHistory(
    some_chain,                # 被附加历史消息的Runnable，通常是chain
    None,                      # 获取指定会话ID的历史会话的函数
    input_messages_key="input",        # 声明用户输入消息在模板中的占位符
    history_messages_key="chat_history" # 声明历史消息在模板中的占位符
)

# 获取指定会话ID的历史会话记录函数
chat_history_store = {}  # 存放多个会话ID所对应的历史会话记录
# 函数传入为会话ID（字符串类型）
# 函数要求返回BaseChatMessageHistory的子类
# BaseChatMessageHistory类专用于存放某个会话的历史记录
# InMemoryChatMessageHistory是官方自带的基于内存存放历史记录的类
def get_history(session_id):
    if session_id not in chat_history_store:
        # 返回一个新的实例
        chat_history_store[session_id] = InMemoryChatMessageHistory()
    return chat_history_store[session_id]

import os

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnableLambda

str_parser = StrOutputParser()
model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt

prompt = PromptTemplate.from_template(
    "你需要根据对话历史回应用户问题。对话历史:{chat_history}。用户当前输入: {input}, 请给出回应"
)

base_chain = prompt | RunnableLambda(print_prompt) | model | StrOutputParser()

chat_history_store = {}  # 存放多个会话ID所对应的历史会话记录
def get_history(session_id):
    if session_id not in chat_history_store:
        # 存入新的实例
        chat_history_store[session_id] = InMemoryChatMessageHistory()
    return chat_history_store[session_id]

# 通过RunnableWithMessageHistory获取一个新的带有历史记录功能的chain
conversation_chain = RunnableWithMessageHistory(
    base_chain,    # 被附加历史消息的Runnable，通常是chain
    get_history,   # 获取历史会话的函数
    input_messages_key="input",         # 声明用户输入消息在模板中的占位符
    history_messages_key="chat_history" # 声明历史消息在模板中的占位符
)

if __name__ == '__main__':
    # 如下固定格式，配置当前会话的ID
    session_config = {"configurable": {"session_id": "user_001"}}

    print(conversation_chain.invoke({"input": "小明有一只猫"}, session_config))
    print(conversation_chain.invoke({"input": "小刚有两只狗"}, session_config))
    print(conversation_chain.invoke({"input": "共有几只宠物?"}, session_config))

12.2 持久化记忆

FileChatMessageHistory 类实现，核心思路：

  • 基于文件存储会话记录，以 session_id 为文件名，不同 session_id 有不同文件存储消息

继承 BaseChatMessageHistory 实现如下 3 个方法：

  • add_messages：同步模式，添加消息
  • messages：同步模式，获取消息
  • clear：同步模式，清除消息

如下面代码，官方在 BaseChatMessageHistory 类的注释中提供了一个基于文件存储的示例代码。

import json, os
from langchain_core.messages import messages_from_dict, message_to_dict

class FileChatMessageHistory(BaseChatMessageHistory):
    storage_path: str
    session_id: str

    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(
                os.path.join(self.storage_path, self.session_id),
                "r",
                encoding="utf-8",
            ) as f:
                messages_data = json.load(f)
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        all_messages = list(self.messages) # Existing messages
        all_messages.extend(messages) # Add new messages

        serialized = [message_to_dict(message) for message in all_messages]
        file_path = os.path.join(self.storage_path, self.session_id)
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump(serialized, f)

    def clear(self) -> None:
        file_path = os.path.join(self.storage_path, self.session_id)
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump([], f)

修改一下：file_chat_message_history.py

import json, os
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import messages_from_dict, message_to_dict, BaseMessage

"""
   基于本地JSON文件持久化存储会话历史的自定义历史记录类
   继承LangChain标准基类 BaseChatMessageHistory，必须实现规定抽象方法
"""
class FileChatMessageHistory(BaseChatMessageHistory):

    # message_to_dict: 单个消息对象（BaseMessage类实例） -> 字典
    # messages_from_dict: [字典、字典……] -> [消息、消息……]

    def __init__(self, session_id, storage_path):
        """
           构造方法：初始化会话ID、存储目录、完整文件路径，自动创建文件夹
           :param session_id: 唯一会话标识，作为存储文件名
           :param storage_path: 存放会话文件的根目录
        """
        self.session_id = session_id   # 会话id
        self.storage_path = storage_path  #存储的文件路径
        self.file_path = os.path.join(self.storage_path, self.session_id)  # 完整的文件路径
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)  #文件不存在则创建

    # @property 装饰器将messages方法变成成员属性用
    @property
    def messages(self) -> list[BaseMessage]:
        """
            【必须重写的核心属性】读取当前会话全部历史消息
            1. 父类BaseChatMessageHistory规定抽象属性messages，子类必须实现
            2. 从本地JSON文件读取序列化字典，还原成BaseMessage消息对象列表
            3. 文件不存在时返回空列表，代表暂无历史对话
            :return: 消息对象列表 [HumanMessage, AIMessage, ...]
        """
        try:
            with open(self.file_path,"r",encoding="utf-8",) as f:
                messages_data = json.load(f) # 加载数据，转为消息
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def add_messages(self, messages) -> None:
        """
           【必须重写】追加多条消息到当前会话历史并持久化写入文件
           :param messages: 待新增的BaseMessage消息列表
        """
        all_messages = list(self.messages) # 存在历史消息
        all_messages.extend(messages) # 添加新的消息

        # 将消息转为字典
        serialized = [message_to_dict(message) for message in all_messages]
        # 写入文件
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(serialized, f)

    def clear(self) -> None:
        """【必须重写】清空当前会话所有历史消息"""
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnableLambda
from file_chat_message_history import FileChatMessageHistory

str_parser = StrOutputParser()
model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

def print_prompt(full_prompt):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt

prompt = PromptTemplate.from_template(
    "你需要根据对话历史回应用户问题。对话历史:{chat_history}。用户当前输入: {input}, 请给出回应"
)

base_chain = prompt | RunnableLambda(print_prompt) | model | StrOutputParser()

def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")

# 通过RunnableWithMessageHistory获取一个新的带有历史记录功能的chain
conversation_chain = RunnableWithMessageHistory(
    base_chain,    # 被附加历史消息的Runnable，通常是chain
    get_history,   # 获取历史会话的函数
    input_messages_key="input",         # 声明用户输入消息在模板中的占位符
    history_messages_key="chat_history" # 声明历史消息在模板中的占位符
)


if __name__ == '__main__':
    # 如下固定格式，配置当前会话的ID
    session_config = {"configurable": {"session_id": "user_001"}}

    print(conversation_chain.invoke({"input": "小明有一只猫"}, session_config))
    print(conversation_chain.invoke({"input": "小刚有两只狗"}, session_config))
    #print(conversation_chain.invoke({"input": "共有几只宠物?"}, session_config))

13.文档加载器

13.1 Documnet loaders文档加载器

文档加载器提供了一套标准接口，用于将不同来源（如 CSV、PDF 或 JSON 等）的数据读取为 LangChain 的文档格式。这确保了无论数据来源如何，都能对其进行一致性处理。

文档加载器（内置或自行实现）需实现BaseLoader接口。

Class Document，是 LangChain 内文档的统一载体，所有文档加载器最终返回此类的实例。

一个基础的 Document 类实例，基于如下代码创建：

from langchain_core.documents import Document

document = Document(
    page_content="Hello, world!", metadata={"source": "https://example.com"}
)

可以看到，Document 类其核心记录了：

  • page_content：文档正文内容
  • metadata：文档元数据（字典格式，存放来源、页码、时间等附加信息）

不同的文档加载器可能定义了不同的参数，但是其都实现了统一的接口（方法）。

  • load()：一次性加载全部文档
  • lazy_load()：延迟流式传输文档，对大型数据集很有用，避免内存溢出。
from langchain_community.document_loaders.csv_loader import CSVLoader

loader = CSVLoader(
    ...  # 初始化参数
)

# 一次性加载全部文档
documents = loader.load()

# 对于大数据集，分段返回文档
for document in loader.lazy_load():
    print(document)

13.2 CSVLoader加载器

LangChain 内置了许多文档加载器，详细参见官方文档：

https://docs.langchain.com/oss/python/integrations/document_loaders

我们简单的学习如下几个常用的文档加载器：

  • CSVLoader
  • JSONLoader
  • PDFLoader
from langchain_community.document_loaders.csv_loader import CSVLoader

loader = CSVLoader(file_path="./xxx.csv", encoding="utf-8")
data = loader.load()
print(data)

自定义 CSV 文件的解析和加载：

loader = CSVLoader(
    file_path="./xxx.csv",
    csv_args={
        "delimiter": ",",    # 指定分隔符
        "quotechar": '"',    # 指定字符串的引号包裹，如果内容包含了分隔符，比如：,  则需要用quotechar指定的符号包裹起来
        # 字段列表（无表头使用，有表头勿用，否则会读取首行做为数据）
        "fieldnames": ["name", "age", "gender"],
    },
    encoding="utf-8"
)
data = loader.load()
print(data)

示例：

name,age,gender
张三,18,男
张三2,1,男
张三3,2,男
张三4,3,女
张三5,4,男
张三6,5,男
张三7,6,女
张三8,7,男

from langchain_community.document_loaders.csv_loader import CSVLoader

# 一次性读取
loader = CSVLoader(file_path="./data/test2.csv", encoding="utf-8")
data = loader.load()
print(data)

# 分段读取
for document in loader.lazy_load():
    print(type(document))
    print(document.page_content, end="\n\n")

name,age,gender,desc
张三,18,男,"篮球,羽毛球"
张三2,1,男,"篮球,羽毛球"
张三3,2,男,"篮球,羽毛球"
张三4,3,女,"篮球,羽毛球"
张三5,4,男,"篮球,羽毛球"
张三6,5,男,"篮球,羽毛球"
张三7,6,女,"篮球,羽毛球"
张三8,7,男,"篮球,羽毛球"

from langchain_community.document_loaders.csv_loader import CSVLoader

loader = CSVLoader(
    file_path="./data/test2.csv",
    encoding="utf-8",
    csv_args={
        "delimiter": ",",
        "quotechar": '"',   # 由于单元格内容包含分隔符，为了不被识别为分隔符，则用"包裹起来
        "fieldnames": ["name", "age", "gender", "desc"],  #表头
    },
)
data = loader.load()
print(data)

for document in loader.lazy_load():
    print(type(document))
    print(document.page_content, end="\n\n")

13.3 JSONLoader加载器

JSONLoader 用于将 JSON 数据加载为 Document 类型对象。

使用 JSONLoader 需要额外安装依赖：pip install jq

jq 是一个跨平台的 json 解析工具，LangChain 底层对 JSON 的解析就是基于 jq 工具实现的。

将 JSON 数据的信息抽取出来，封装为 Document 对象，抽取的时候依赖jq_schema语法。

单JSON对象：

{
  "name": "周杰伦",
  "age": 11,
  "hobby": ["唱", "跳", "RAP"],
  "other": {
    "addr": "深圳",
    "tel": "12332112321"
  }
}

jq 语法规则：

  • . 表示整个 JSON 对象（根）
  • .[] 表示数组
  • .name 表示抽取周杰伦
  • .hobby 表示抽取爱好数组
  • .hobby[1] 或 .hobby.[1] 表示抽取第二项「跳」
  • .other.addr 表示抽取地址深圳

JSON数组：

[
  {"name": "周杰伦", "age": 11, "gender": "男"},
  {"name": "蔡依临", "age": 12, "gender": "女"},
  {"name": "王力宏", "age": 11, "gender": "男"}
]

jq 语法规则：

  • .[] 遍历数组，得到 3 个独立字典

  • .[].name 抽取数组内所有对象的 name 字段，得到 3 个姓名信息

了解 jq 的基本抽取规则后，即可使用 JSONLoader 加载 JSON 文件。

from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="xxx.json",    # json文件路径
    jq_schema=".",           # jq 抽取语法
    text_content=False,       # 抽取内容是否为纯字符串，默认True
    json_lines=True,         # 是否为JsonLines格式文件（每行一个独立json对象）
)

JsonLines 文件示例

文件内每行单独一条 JSON，整体不是标准 JSON 数组：

{"name": "周杰伦", "age": 11, "gender": "男"}
{"name": "蔡依临", "age": 12, "gender": "女"}
{"name": "王力宏", "age": 11, "gender": "男"}

示例：

# 读取json对象的单个key
from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/one.json",    # json文件路径
    jq_schema=".name",           # jq 抽取语法
    text_content=False,       # 抽取内容是否为纯字符串，默认True
    json_lines=False,         # 是否为JsonLines格式文件（每行一个独立json对象）
)

data = loader.load()
print(data[0].page_content)

# 读取json对象的多个key
from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/one.json",    # json文件路径
    jq_schema="{name, age, hobby:.hobby[2], addr: .other.addr}",           # jq 抽取语法，提取多个字段为字典
    text_content=False,       # 抽取内容是否为纯字符串，默认True
    json_lines=False,         # 是否为JsonLines格式文件（每行一个独立json对象）
)

data = loader.load()
print(data[0].page_content)

# 读取json数组
from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/one.json",    # json文件路径
    jq_schema=".[]",           # jq 抽取语法
    text_content=False,       # 抽取内容是否为纯字符串，默认True
    json_lines=False,         # 是否为JsonLines格式文件（每行一个独立json对象）
)

data = loader.load()
for document in data:
    print(document.page_content)

# 读取jsonlines
from langchain_community.document_loaders import JSONLoader

loader = JSONLoader(
    file_path="./data/one.json",    # json文件路径
    jq_schema=".[]",           # jq 抽取语法
    text_content=False,       # 抽取内容是否为纯字符串，默认True
    json_lines=True,         # 是否为JsonLines格式文件（每行一个独立json对象）
)

data = loader.load()
for document in data:
    print(document)

13.4 TextLoader加载器

除了前文学习的三个 Loader 以外，还有一个基础的加载器：TextLoader

作用：读取文本文件（如 .txt），将全部内容放入一个 Document 对象中。

from langchain_community.document_loaders import TextLoader

loader = TextLoader(
    "xxx.txt",
    encoding="utf-8",
)
docs = loader.load()
print(docs)
print(len(docs))  # 结果为1

如果文档很大，加载一个Document对象中是否不太合适？那就可以使用文档分割器：RecursiveCharacterTextSplitter

RecursiveCharacterTextSplitter，递归字符文本分割器，主要用于按自然段落分割大文档。

是 LangChain 官方推荐的默认字符分割器。它在保持上下文完整性和控制片段大小之间实现了良好平衡，开箱即用效果佳。

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. 加载txt文档
loader = TextLoader(
    "./data/Python 基础语法教程.txt",
    encoding="utf-8",
)
docs = loader.load()

# 2. 初始化递归文本分割器
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,          # 单个片段最大字符数
    chunk_overlap=50,        # 相邻片段之间重叠字符数
    # 文本分段依据，从左到右优先级依次降低
    separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
    # 用于统计文本长度的函数，默认len计算字符长度
    length_function=len,
)

# 3. 对Document列表执行分割
split_docs = splitter.split_documents(docs)
for doc in split_docs:
    print("====================================")
    print(doc, end="\n\n")

13.5 PyPDFLoader加载器

LangChain 内支持许多 PDF 的加载器，我们选择其中的 PyPDFLoader 使用。

PyPDFLoader 加载器依赖 pypdf 库，需要先安装依赖：pip install pypdf

PyPDFLoader 使用简单，以下代码加载 PDF 文字内容：

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(
    file_path="",        # 必填，PDF文件路径
    mode='page',         # 读取模式，两种可选：
                         # page：每页生成一个独立Document
                         # single：整个PDF合并为一个Document
    password='password'  # 加密PDF填写打开密码，无密码可省略
)

from langchain_community.document_loaders import PyPDFLoader

#============非加密================
loader = PyPDFLoader(
    file_path="./data/pdf.pdf",        # 必填，PDF文件路径
    mode='single',         # 读取模式，两种可选：
                         # page：每页生成一个独立Document
                         # single：整个PDF合并为一个Document
    #password='password'  # 加密PDF填写打开密码，无密码可省略
)
data = loader.load()
print(data)

#=============加密================
loader2 = PyPDFLoader(
    file_path="./data/加密后.pdf",        # 必填，PDF文件路径
    mode='single',         # 读取模式，两种可选：
                         # page：每页生成一个独立Document
                         # single：整个PDF合并为一个Document
    password='123456'  # 加密PDF填写打开密码，无密码可省略
)
data2 = loader2.load()
print(data2)

14.Vector stores向量存储

基于 LangChain 的向量存储，存储嵌入数据，并执行相似性搜索。

在这里插入图片描述

上图是典型向量存储应用，也就是标准 RAG 检索增强生成流程。

开发核心两大模块：

  1. 文本向量化（前文已学习）
  2. 向量存储操作，包含三类核心功能：
    • 存入向量：add_docment
    • 删除向量：delete
    • 向量相似度检索：similarity_search

14.1 内置向量存储 InMemoryVectorStore（内存临时向量库）：

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings

# 初始化内存向量库，绑定嵌入模型
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings())

# 批量添加文档，自定义每条文档唯一id
vector_store.add_documents(documents=[doc1, doc2], ids=["id1", "id2"])

# 根据id删除指定向量文档
vector_store.delete(ids=["id1"])

# 相似度检索：查询文本，返回Top4最相似文档
similar_docs = vector_store.similarity_search("your query here", 4)

source,info
SB程序员，Python 是世界上最好的编程语言
xx教育，股票代码 002032
SB程序员，LangChain 极大地方便了大模型开发
SB程序员，AI 和 Python 是下一个十年的风口
xx教育，Python 学起来很简单的
SB程序员，学习 Python 键盘敲烂月薪过万
SB程序员，努力带来成就，Python 助力辉煌
SB程序员，学习 Python 的时候也要记得好好休息打打篮球
SB程序员，明天晚上吃啥子呀
SB程序员，如何快速减肥呢

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader
import os

loader = CSVLoader(
    file_path="./data/test2.csv",
    encoding="utf-8",
    source_column="source"
)

documents = loader.load()
print(documents)

# 初始化内存向量库，绑定嵌入模型
vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(
        dashscope_api_key=os.environ.get('qwen-api-key')
    )
)

# 批量添加文档，自定义每条文档唯一id
vector_store.add_documents(
    documents=documents,
    ids=["id" + str(i) for i in range(1, len(documents) + 1)]
)

# 根据id删除指定向量文档
vector_store.delete(ids=["id5"])

# 相似度检索：查询文本，返回Top4最相似文档
similar_docs = vector_store.similarity_search("python", 3)
for doc in similar_docs:
    print(doc)

14.2 外部持久向量存储 Chroma（本地磁盘持久化）

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma

vector_store = Chroma(
    collection_name="example_collection",    # 向量集合名称
    embedding_function=DashScopeEmbeddings(),  # 嵌入模型
    persist_directory="./chroma_langchain_db" # 本地持久化文件夹，省略则仅内存
)

使用Chroma数据库需要安装如下得包：

pip install langchain-chroma chromadb

  • chromadb：Chroma 向量数据库底层核心库，提供持久化向量存储、相似度检索能力。

  • langchain-chroma：LangChain 官方适配层，提供 Chroma 向量存储类，用于在 LangChain 中操作 Chroma。

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import CSVLoader
import os
from langchain_chroma import Chroma

loader = CSVLoader(
    file_path="./data/test2.csv",
    encoding="utf-8",
    source_column="source"
)
documents = loader.load()
print(documents)

# 初始化向量库，绑定嵌入模型
vector_store = Chroma(
    collection_name="example_collection",    # 向量集合名称
    embedding_function=DashScopeEmbeddings(
        dashscope_api_key=os.environ.get('qwen-api-key')
    ),  # 嵌入模型
    persist_directory="./data/chroma_langchain_db" # 本地持久化文件夹，省略则仅内存
)

# 批量添加文档，自定义每条文档唯一id
# vector_store.add_documents(
#     documents=documents,
#     ids=["id" + str(i) for i in range(1, len(documents) + 1)]
# )

# 根据id删除指定向量文档
# vector_store.delete(ids=["id5"])

# 相似度检索：查询文本，返回Top4最相似文档
similar_docs = vector_store.similarity_search("python", 3)
for doc in similar_docs:
    print(doc)

15.基于向量检索并构建提示词

"""
提示词: 用户的提问 + 向量库中检索到的参考资料
"""
from langchain_core.runnables import RunnableLambda
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问: {input}")
    ]
)

vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(
        model="text-embedding-v4",
        dashscope_api_key=os.environ.get('qwen-api-key')
    )
)

# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

# 检索向量库
result = vector_store.similarity_search(input_text, k=2)
reference_text = "["
for doc in result:
    reference_text += doc.page_content
reference_text += "]"
print("向量库检索结果：" + reference_text)


def print_prompt(prompt):
    print(prompt.to_string())
    print("="*20)
    return prompt

# 构建chain
chain = prompt | RunnableLambda(print_prompt) | model | StrOutputParser()

res = chain.invoke({"input": input_text, "context": reference_text})
print(res)

16.RunnablePassthrough得使用

作用：将向量检索入chain

"""
提示词: 用户的提问 + 向量库中检索到的参考资料
"""
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
from langchain_openai import ChatOpenAI


model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ.get('qwen-api-key'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供的已知参考资料为主，简洁和专业的回答用户问题。参考资料:{context}。"),
        ("user", "用户提问: {input}")
    ]
)

vector_store = InMemoryVectorStore(
    embedding=DashScopeEmbeddings(
        model="text-embedding-v4",
        dashscope_api_key=os.environ.get('qwen-api-key')
    )
)

def print_prompt(prompt):
    print(prompt.to_string())
    print("="*20)
    return prompt

def format_str(docs):
    reference_text = "["
    for doc in docs:
        reference_text += doc.page_content
    reference_text += "]"
    return reference_text


# 准备一下资料（向量库的数据）
# add_texts 传入一个 list[str]
vector_store.add_texts(["减肥就是要少吃多练", "在减脂期间吃东西很重要,清淡少油控制卡路里摄入并运动起来", "跑步是很好的运动哦"])

input_text = "怎么减肥？"

# langchain中向量存储对象，有一个方法：as_retriever，可以返回一个Runnable接口的检索器实例
retriever = vector_store.as_retriever(search_kwargs={"k": 2})

'''
retriever:
  - 输入: 用户的提问 str
  - 输出: 向量库的检索结果 list[Document]
prompt:
  - 输入: 用户的提问 + 向量库的检索结果 dict
  - 输出: 完整的提示词 PromptValue
'''

# chain
chain = (
    {"input": RunnablePassthrough(), "context": retriever | RunnableLambda(format_str)} | prompt | RunnableLambda(print_prompt) | model | StrOutputParser()
)
# input_text会提供给retriever和RunnablePassthrough，retriever作为chain得第一个组件
res = chain.invoke(input_text)
print(res)
# 人工智能 # 学习
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

【安心陪诊 Agent】HarmonyOS 数据存储设计：Preferences 与 RDB 怎么选

avatar AI Agent技术社区
cover

2026年最高频AI Agent框架技术深度解析：从选型到实战

avatar AI Agent技术社区

AI测试专栏第2期：零基础看懂AI项目架构！大模型、RAG、Agent、向量库通俗拆解

AI测试必须校验输出稳定性、推理速度、多轮一致性、幻觉概率。向量库存语义、相似度。四、AI Agent：2026最火、薪资最高的测试赛道 Agent不是简单聊天，是能自主思考、自主规划、自主调用工具、自主执行任务的智能体。六、传统测试如何快速对标切入 - 自动化测试 → 做多轮对话、Prompt场景、链路回归- 性能测试 → 测推理耗时、Token吞吐、向量检索并发- DFX测试 → 测断库、断网

avatar AI Agent技术社区