课程地址

第1章 前置准备

阿里云百炼网址
注册->登录->支付宝实名认证->申请key

查看免费额度

1.1 代码调用云端大模型

通过代码调用阿里云百炼平台上的模型很简单，仅需要：

• 阿里云百炼平台上，创建API-KEY
• 通过pip为Python程序提供OpenAI库：pip install openai -i https://pypi.tuna.tsinghua.edu.cn/simple
• 编写代码测试（可从百炼平台官网复制代码，直接执行）

复制代码 替换自己申请的key

from openai import OpenAI
import os

client = OpenAI(
    # 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
    api_key="sk-83c1ae70cf0b449c87f377a75e07a59c",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

messages = [{"role": "user", "content": "你是谁"}]
completion = client.chat.completions.create(
    model="qwen3-max",  # 您可以按需更换为其它深度思考模型
    messages=messages,
    extra_body={"enable_thinking": True},
    stream=True
)
is_answering = False  # 是否进入回复阶段
print("\n" + "=" * 20 + "思考过程" + "=" * 20)
for chunk in completion:
    delta = chunk.choices[0].delta
    if hasattr(delta, "reasoning_content") and delta.reasoning_content is not None:
        if not is_answering:
            print(delta.reasoning_content, end="", flush=True)
    if hasattr(delta, "content") and delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20)
            is_answering = True
        print(delta.content, end="", flush=True)

1.2 使用环境变量保护APIKEY

OPENAI_API_KEY ：用于openai库，记录APIKEY
DASHSCOPE_API_KEY： 用于langchain库（后续学习）
win+s->高级系统设置

1.3 Ollama简介

Ollama官方模型库

1.4 Windows/Mac系统部署Ollama

• 进入官网点击Download按钮下载安装
• 安装完成后，通过命令：ollama run 模型名称
• 即可运行对应的模型，并在命令行内做交互

第2章 OPENAI库基础使用

2.1 OpenAI基础使用

1. 获取客户端对象

from openai import OpenAI
client = OpenAI(
	api_key="你的key"
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

主要参数：

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

2. 调用模型

from openai.types.chat_completion import ChatCompletion
response= client.chat.completions.create(
    model="qwen3-max",  # 您可以按需更换为其它深度思考模型
    messages=[
    	{"role": "system", "content": "你是一个python编程专家"},
    	{"role": "assistant", "content": "我是一个python编程专家。请问有什么可以帮助您的吗"},
    	{"role": "user", "content": "for循环输出1到5的数字"}
    ],
)

client.chat.completions.create创建ChatCompletion对象
主要参数：

• model:选择所用模型
• message:提供给模型的信息
  • 类型：list，可以包含多个字典消息
  • 每个字典消息包含两个key
    • role：角色
    • content：内容
• system角色:设定助手的整体性为、角色和规则，为对话提供上下文框架（如指定助手身份、回答风格、核心要求），是全局的背景设定，影响后续所有交互
• assistant角色:代表AI助手的回答，可以在代码中认为设定
• user角色:代表用户，发送问题、指令或需求

3. 处理结果

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

总代码如下：

from openai import OpenAI
# 1. 获取client对象，OpneAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 2. 调用模型
response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是一个Python编程专家，并且不说废话简单回答"},
        {"role":"assistant","content":"好的，我是编程专家，并且话不多，你要问什么？"},
        {"role":"user","content":"输出1-10的数字，使用Python代码"},
    ]
)
# 3. 处理结果
print(response.choices[0].message.content)

2.2 流式输出

• 在client.chat.completions.create()调用模型的时候设置参数：stream=True
• for循环response对象，并在循环内输出内容

from openai import OpenAI
# 1. 获取client对象，OpneAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 2. 调用模型
response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是一个Python编程专家，并且话非常多"},
        {"role":"assistant","content":"好的，我是编程专家，并且话非常多，你要问什么？"},
        {"role":"user","content":"输出1-10的数字，使用Python代码"},
    ],
    stream=True # 开启了流式输出的功能
)
# 3. 处理结果
# print(response.choices[0].message.content)
for chunk in response:
    print(
        chunk.choices[0].delta.content,
        end=' ', # 每一段之间以空格分隔
        flush=True # 立即刷新缓冲区
    )

2.3 附带历史消息调用模型

调用模型传入的参数message，其要求事list对象，即表明其支持非常多的消息在内，我们可以基于此，将历史消息填入，让模型知晓对话的上下文，更好的回答

from openai import OpenAI
# 1. 获取client对象，OpneAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 2. 调用模型
response = client.chat.completions.create(
    model="qwen3-max",
    messages=[
        {"role":"system","content":"你是一个AI助理，回答很简洁"},
        {"role":"user","content":"小明有2条宠物狗"},
        {"role":"assistant","content":"好的"},
        {"role": "user", "content": "小明有3条宠物猫"},
        {"role": "assistant", "content": "好的"},
        {"role":"user","content":"总共有几个宠物？"}
    ],
    stream=True # 开启了流式输出的功能
)
# 3. 处理结果
# print(response.choices[0].message.content)
for chunk in response:
    print(
        chunk.choices[0].delta.content,
        end=' ', # 每一段之间以空格分隔
        flush=True # 立即刷新缓冲区
    )

第3章 提示词工程

3.1 大模型prompt工程指南

• 技巧1：详细的描述
• 技巧2：让模型充当某个角色
• 技巧3：使用分隔符标明输入的不同部分
• 技巧4：对任务指定步骤
• 技巧5：提供例子
• 技巧6：使用参考文本作答

3.2 提示词优化案例介绍和零样本少样本思想

基于Few-Shot + Zero-Shot的思想
Zero-Shot：这种能力被称为“零样本”学习，因为模型在训练时从未见过测试集中的新类别，在模型训练和提示词优化中均有体现

• 在模型训练中
  • 已知马(四角兽)、虎(有条纹)、熊猫(黑白色)的特征，但未训练过斑马的数据(不认识)
  • 告知模型：斑马是四角兽、有黑白色的条纹
  • 模型可以在已知数据中进行推理，从而识别斑马
• 在提示词优化中
  • Zero-Shot思想用于基于已训练的能力，不提供任何示例，仅通过语言描述任务的要求、目标和约束，让模型直接生成结果

Few-Shot：是指少样本学习，当模型在学习了一定类别的大量数据后，对于新的类别，只需要少量的样本就能快速学习，对应的有one-shot learning，单样本学习，也算样本少到为一的情况下的一种Few-Shot learning

• 在模型训练中(相似度判断方法)
  • 基于少量企鹅样本并结合相似度判断，推论未知图片内涵“企鹅”
• 在提示词优化中
  • 主要用于基于少量示例，让模型参考示例回答

3.3 提示词实战案例

from openai import OpenAI
# 1.获取client对象，OpenAI类对象
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
examples_data = {#示例数据
    '新闻报道':'今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。',
    '财务报告':'本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。',
    '公司公告':'本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力',
    '分析师报告':'最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势'
}
# 分类列表
examples_types = ['新闻报道','财务报道','公司公告','分析师报告']
#提问数据
questions = [
    "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。",
    "ABC公司今日发布公告称，已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉，此次收购将进一步巩固ABC公司在行业中的地位，并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏",
    "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。",
    "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会",
    "小明喜欢小新哟"
]

messages = [
    {"role":"system","content":"你是金融专家，将文本分类为[‘新闻报道’，‘财务报道’，‘公司公告’，‘分析师报告’]，不清楚的分类为‘不清楚类别’下面有示例："},
]
for key,value in examples_data.items():
    messages.append({"role":"user","content":value})
    messages.append({"role":"assistant","content":key})

# 向模型提问
for q in questions:
    response = client.chat.completions.create(
        model="qwen3-max",
        messages=messages + [{"role":"user","content":f'按照示例，回答这段文本的分类类别：{q}'}],
    )
    print(response.choices[0].message.content)

3.4 Json数据格式

1. 概念

• Json对象
  • key必须是字符串
  • value可以是：
    • 数字
    • 字符串
    • 列表
    • Json对象或Json数组
• Json数组
  • 即一堆Json对象的组合体

2. python中使用Json主要完成：

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

3. 主要使用Python内置的json库

• json.dumps(字典或列表,ensure_ascii=False)：将字典或列表转换为Json字符串
  • ensure_ascii参数确保中文能正常显示
  • 返回值：Json字符串
• json.loads(json字符串)：将json字符串转换为python字典或列表
  • 返回值：Python字典或Python列表

import json
d = {
    "name":"周杰伦",
    "age":20,
    "gender":"男"
}
print(str(d)) # 单引号
s = json.dumps(d,ensure_ascii=False)
print(s) # 双引号

l = [
        {
            "name":"周杰伦",
            "age":20,
            "gender":"男"
        },
        {
            "name":"蔡依林",
            "age":18,
            "gender":"女"
        },
        {
            "name":"小明",
            "age":11,
            "gender":"男"
        }
    ]
print(json.dumps(l, ensure_ascii=False))

json_str = '{"name": "周杰伦", "age": 20, "gender": "男"}'
json_arr = '[{"name": "周杰伦", "age": 20, "gender": "男"}, {"name": "蔡依林", "age": 18, "gender": "女"}, {"name": "小明", "age": 11, "gender": "男"}]'
print(json.loads(json_str))
print(type(json.loads(json_str)))
print(json.loads(json_arr))
print(type(json.loads(json_arr)))

3.5 提示词优化案例——金融文本信息提取

from openai import OpenAI
import json
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
schema = ['日期','开盘价','收盘价','成交量']
example_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-=6-06,股市利好。股票黑马程序员A股今日开盘价200人民币，一度飙升至211人民币，随后回落至201人民币，最终以206人民币收盘"
]
messages = [
    {"role":"system","content":"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下实例"}
]
for example in example_data:
    messages.append({"role":"user","content":example["content"]})
    messages.append({"role":"assistant","content":json.dumps(example["answers"],ensure_ascii=False)})

for q in questions:
    response = client.chat.completions.create(
        model="qwen3-max",
        messages=messages + [{"role":"user","content":f'按照上述的示例，现在抽取这个句子的信息：{q}'}]
    )
    print(response.choices[0].message.content)

3.6 提示词优化案例——金融文本匹配

from openai import OpenAI
client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
example_data = {
    "是":[
        ("公司ABC发布了季度财报，显示盈利增长","财报披露，公司ABC利润上升"),
        ("公司ITCAST发布了年度财报，显示盈利大幅度增长","财报披露，公司ITCAST更赚钱了")
    ],
    "不是":[
        ("黄金价格下跌，投资者抛售","外汇市场交易の创下新高"),
        ("央行降息，刺激经济增长","新能源技术的创新")
    ]
}
question = [
    ("利率上升，影响房地产市场","高利率对房地产有一定的冲击"),
    ("油价大幅度下跌，能源公司面临挑战","未来智能城市的建设趋势越加明显"),
    ("股票市场今日大涨，投资者乐观","持续上涨的市场让投资者感到满意")
]
messages = [
    {"role":"system","content":"你帮我完成文本匹配，我给你2个句子，被[]包围，你判断他们是否匹配，回答是或不是，请参考如下实例："}
]
for key,value in example_data.items():
    for t in value:
        messages.append({"role": "user", "content": f'句子1：[{t[0]}]，句子2：[{t[1]}]'})
        messages.append({"role":"assistant","content":key})
for q in question:
    response = client.chat.completions.create(
        model="qwen3-max",
        messages=messages + [{"role":"user","content":f'句子1：[{q[0]}]，句子2：[{q[1]}]'}],
    )
    print(response.choices[0].message.content)

第4章 RAG开发

4.1 LangChain简介

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

LangChain主要功能

• Prompts：优化提示词（提示词工程）
• Models：调用各类模型
• History：管理会话历史记录（记忆）
• Indexs：管理和分析各类文档
• Chains：构建功能的执行链条
• Agent：构建智能体

LangChain是一个开发LLM相关业务功能的集大成者，是一个python的第三方库，提供了各种功能的API
提供：

• 提示词优化的相关API
• 调用各类模型的功能API
• 会话记忆的相关功能API
• 各类文档管理分析的功能API
• 构建Agent智能体的相关功能API
• 各类功能链执行的能力

pip install langchain langchain-community langchain-ollama dashscope chromadb

4.2 RAG介绍

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

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

解决：

• 领域知识和私有数据
• 实时数据
• 减少生成不确定性
• 增强数据安全

RAG即检索增强生成，为大模型提供了从特定数据源检索到的信息，以此来修正和补充生成的答案。可以总结为一个公式：RAG = 检索技术 + LLM提示

RAG标准流程

• 索引阶段，通过处理多种来源多种格式的文档提取其中文本，将其切分为标准长度的文本块(chunk)，并进行嵌入向量化(embedding)，向量存储在向量数据库(vector database)中
• 检索结算，用户输入的查询(query)被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块
• 生成阶段，检索到的相关文本与原始查询共同构成提示词(prompt)，输入大语言模型(LLM)，生成精确且具备上下文关联的回答

4.3 向量的基础概念

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

• 离线流程：知识和信息->向量嵌入(向量化)->存入向量库
• 在线流程：用户的提问->向量嵌入(向量化)->在向量库中匹配

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

2. 文本嵌入模型(如text-embedding-v1)通过深度学习等技术，从文本提取语义特征并映射为固定长度的数字序列
3. 在向量匹配的过程中，如何识别2段文本是否表述相似的含义，主要可以通过如余弦相似度等算法来完成。比如（如下数据不是真实向量）
A： “如何快速学打篮球”[0.2, 0.5, 0.8]
B： “打篮球怎么学得快”[0.18, 0.52, 0.79]
C： “运动后吃什么好呢”[0.9, 0.1, 0.2]
通过余弦相似度算法可以计算得到：A和B相似度0.999789，A和C相似度0.361446
由此可通过精确的数学计算，去匹配2段文本是否描述同一个意思，提高语义匹配的效率和精度。
4. 如何更为精准的完成语义匹配，生成向量的维度是一个很重要的指标。如text-embedding-v1模型，可以生成1536维的向量（一段文本固定得到1536个数字序列），比较实用。

• 1536个数字表示，这段文本在1536个主题（抽象的语义特征）方向上的得分（强度）
  
• 生成向量的维度越多，就更好的记录文本的语义特征，做语义匹配会更加精准。
• 更多的向量会在计算、存储和匹配过程中，带来更大的压力。

4.4 余弦相似度

向量的数字序列，共同决定了响铃在高位空间中的方向和长度，而余弦相似度主要就是撇除长度的影响，得到方向的夹角，夹角越小越相似，即方向相同

以一维为例：

二维：

余弦相似度主要匹配的就是：同向（无所谓长度）

4.5 LangChain调用大语言模型

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

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

from langchain_community.llms.tongyi import Tongyi
# 不用qwen3-max，因为qwen3-max是聊天模型，qwen-max是大语言模型
model = Tongyi(model="qwen-max")
res = model.invoke(input="你是谁呀？能做什么？")
print(res)

4.6 LangChain模式的流式输出

from langchain_community.llms.tongyi import Tongyi
model = Tongyi(model="qwen-max")
res = model.stream(input="你是谁呀能做什么？")
for chunk in res:
    print(chunk, end=" ", flush=True)

模型对象有2个方法去调用模型

• invoke，调用模型，一次性返回完整结果
• stream，调用模型，逐段流式输出

4.7 LangChain调用聊天模型

• AIMessage:就是AI输出的消息，可以是针对问题的回答. (OpenAI库中的assistant角色）
• HumanMessage:人类消息就是用户信息，由人给出的信息发送给LLMs的提示信息，比如“实现一个快速排序方法”.(OpenAI库中的user角色）
• SystemMessage:可以用于指定模型具体所处的环境和背景，如角色扮演等。你可以在这里给出具体的指示，比如“作为一个代码专家”，或者“返回json格式”. (OpenAI库中的system角色）

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage,AIMessage,SystemMessage
model = ChatTongyi(model="qwen3-max")
message = [
    SystemMessage(content="你是一个边塞诗人"),
    HumanMessage(content="写一首唐诗"),
    AIMessage(content="锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
    HumanMessage(content="按照你上一个回复的格式，再写一首唐诗")
]
res = model.stream(input=message)
for chunk in res:
    print(chunk.content,end=" ",flush=True)

4.8 LangChain消息的简写形式

from langchain_community.chat_models.tongyi import ChatTongyi
model = ChatTongyi(model="qwen3-max")
message = [
    ("system","你是一个边塞诗人"),
    ("human","写一首唐诗"),
    ("ai","锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。"),
    ("human","按照你上一个回复的格式，再写一首唐诗.")
]
res = model.stream(input=message)
for chunk in res:
    print(chunk.content,end=" ",flush=True)

• 使用类对象的方式如下：
  message = [
  SystemMessage(content=“你是一个边塞诗人”),
  HumanMessage(content=“写一首唐诗”),
  AIMessage(content=“锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。”),
  HumanMessage(content=“按照你上一个回复的格式，再写一首唐诗”)
  ]

  • 是静态的，一步到位，直接就得到了Message类的类对象

• 简写形式如下：
  message = [
  (“system”,“你是一个边塞诗人”),
  (“human”,“写一首唐诗”),
  (“ai”,“锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。”),
  (“human”,“按照你上一个回复的格式，再写一首唐诗.”)
  ]

  • 是动态的，需要在运行时由LangChain内部机制转换为Message类对象

• 好处就在于，简写形式避免导包、写起来更简单，更重要的是支持：
  message = [
  (“system”,“今天的天气是{weather}”),
  (“human”,“我的名字是：{name}”),
  (“ai”,“欢迎{lastname}先生”),
  ]

  • 由于是动态的，需要转换步骤，所以简写形式支持内部填充{变量}占位，可在运行是填充具体值

4.9 LangChain调用嵌入模型

from langchain_community.embeddings import DashScopeEmbeddings

# 创建模型对象 不传model默认用的是text-embedding-v1
model = DashScopeEmbeddings()

# 不用invoke stream
print(model.embed_query("我喜欢你"))
print(model.embed_documents(["我喜欢你", "我稀饭你", "晚上吃啥"]))

目前掌握的LangChain API如下

方式	LLMs大语言模型	聊天模型	文本嵌入模型
阿里云千问	from langchain_community.llms.tongyi import Tongyi	from langchain_community.chat_models.tongyi import ChatTongyi	from langchain_community.embeddings import DashScopeEmbeddings
Ollama本地模型	from langchain_ollama import OllamaLLM	from langchain_ollama import ChatOllama	from langchain_ollama import OllamaEmbeddings
方法	invoke批量/stream流式	invoke批量/stream流式	embed_query单次转换/embed_documents批量转换

4.10 LangChain通用提示词模板

提示词优化在模板中非常重要，LangChain提供了PromptTemplate类，用来协助优化提示词。PromptTemplate表示提示词模板，可以构建一个自定义的基础题诗词模板，支持变量的注入，最终生成所需的提示词

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi
# Zero-shot
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)
model = Tongyi(model="qwen-max")
chain = prompt_template | model
res = chain.invoke(input={"lastname":"曲","gender":"儿子"})
print(res)

4.11 FewShot提示词模板

参数：

• examples:示例数据，list，内套字典
• example_prompt:示例数据的提示词模板
• prefix:组装提示词，示例数据前的内容
• suffix:组装提示词，示例数据后的内容
• input_variables:列表，注入的变量列表

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_community.llms.tongyi import Tongyi
# 示例模板
example_template = PromptTemplate.from_template("单词：{word}，反义词：{antonym}")
# 示例的动态数据注入
examples_data = [
    {"word":"大","antonym":"小"},
    {"word":"上","antonym":"下"}
]
few_shot_template = FewShotTemplate = 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)
model = Tongyi(model="qwen-max")
print(model.invoke(input=prompt_text))

4.12 模板类的format和invoke方法

format和invoke的区别在于：

区别	format	invoke
功能	纯字符串替换，解析占位符生成提示词	Runnable接口标准方法，解析占位符生成提示词
返回值	字符串	PromptValue类对象
传参	.format(k=v,k=v,…)	invoke({“k”:v,k":v,…})
解析	支持解析{}占位符	支持解析{}占位符和MessagePlaceholder结构化占位符

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate, ChatPromptTemplate
'''
PromptTemplate -> StringPromptTemplate -> BasePromptTemplate -> RunnableSerializable -> Runnable
FewShotPromptTemplate -> StringPromptTemplate -> BasePromptTemplate -> RunnableSerializable -> Runnable
ChatPromptTemplate -> BaseChatPromptTemplate -> BasePromptTemplate -> RunnableSerializable -> Runnable
'''
template = PromptTemplate.from_template("我的邻居是：{lastname}，最喜欢：{hobby}")
res = template.format(lastname="张达明",hobby="钓鱼")
print(res)
print(type(res))

res1 = template.invoke({"lastname":"周杰伦","hobby":"唱歌"})
print(res1)
print(type(res1))

4.13 ChatPromptTemplate的使用

1. PromptTemplate:通用提示词模板，支持动态注入信息
2. FewShotPromptTemplate:支持基于模板注入任意数量的示例信息
3. ChatPromptTemplate:支持注入任意数量的历史会话信息

• 通过from_messages方法，从列表中获取多轮次会话作为聊天的基础模板
  • PS:前面PromptTemplate类用的from_template仅能接入一条消息，而from_messages可以接入一个list的消息

历史会话信息并不是静态的（固定的），而是随着对话进行不停的积攒，即动态的，所以，理事会话信息需要支持动态注入

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

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system","你是一个边塞诗人，可以做诗"),
        MessagesPlaceholder("history"),
        ("human","请再来一首唐诗")
    ]
)
history_data = [
    ("human","你来写一首唐诗"),
    ("ai","床前明月光，疑是地上霜。举头望明月，低头思故乡。"),
    ("human","好诗，再来一个"),
    ("ai","锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦")
]
# StringPromptValue  to_string()
prompt_text = chat_prompt_template.invoke({"history":history_data}).to_string()
model = ChatTongyi(model="qwen3-max")
res = model.invoke(input=prompt_text)
print(res.content)

4.14 Chain的基础使用

[将组件串联，上一个组件的输出作为下一个组件的输入]是LangChain链（尤其是 | 管道链）的核心工作原理，这也是链式调用的核心价值：实现数据的自动化流转与组件的协同工作，如下：
chain = prompt_template | model
核心前提是：即Runnable子类对象才能入链（以及Callable、Mapping接口子类对象也可加入），如下继承关系：

执行流程：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system","你是一个边塞诗人，可以做诗"),
        MessagesPlaceholder("history"),
        ("human","请再来一首唐诗")
    ]
)
history_data = [
    ("human","你来写一首唐诗"),
    ("ai","床前明月光，疑是地上霜。举头望明月，低头思故乡。"),
    ("human","好诗，再来一个"),
    ("ai","锄禾日当午，汗滴禾下土。谁知盘中餐，粒粒皆辛苦。")
]
model = ChatTongyi(model="qwen3-max")
# 组成链
chain = chat_prompt_template | model

# 通过链去调用invoke或stream
# res = chain.invoke({"history":history_data})
# print(res.content)

res1 = chain.stream({"history":history_data})
for chunk in res1:
    print(chunk.content,end=" ",flush=True)

4.15 [扩展]或运算符的重写

class Test(object):
    def __init__(self,name):
        self.name = name

    def __or__(self,other):
        return MySequence(self,other)

    def __str__(self):
        return self.name

class MySequence(object):
    def __init__(self,*args):
        self.sequence = []
        for arg in args:
            self.sequence.append(arg)

    def __or__(self,other):
        self.sequence.append(other)
        return self

    def run(self):
        for i in self.sequence:
            print(i)

if __name__ == "__main__":
    a = Test("a")
    b = Test("b")
    c = Test("c")
    d = a | b | c
    d.run()

4.16 StrOutputParser字符串输出解析器

chain = prompt | model | model
错误的主要原因：

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

解决：
需要做类型转换，可以借助LangChain内置的解析器

• StrOutputParser 字符串输出解析器

parser = StrOutputParser()
chain = prompt | model | parser | model

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

parser = StrOutputParser()
model = ChatTongyi(model="qwen3-max")

prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告知我名字无需其他内容"
)
chain = prompt | model | parser | model | parser
res = chain.invoke({"lastname":"张","gender":"女儿"})
print(res)

4.17 JsonOutputParser和多模型执行链

所以，我们需要完成：
将模型输出的AIMessage -> 转为字典 -> 注入第二个提示词模板，形成新的提示词（PromptValue对象）
StrOutputParser不满足（AIMessage -> Str）
更换JsonOutputParser（AIMessage -> Dict(JSON)）

from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate

# 创建所需的解析器
str_parser = StrOutputParser()
json_parser = JsonOutputParser()

# 创建模型
model = ChatTongyi(model="qwen3-max")

# 创建第一个提示词模板
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 = chain.stream({"lastname":"曲","gender":"儿子"})

for chunk in res:
    print(chunk,end=" ",flush=True)

4.18 自定义函数加入链

前文我们根据JsonOutputParser完成了多模型执行链条的构建

• 除了JsonOutputParser这类固定功能的解析器外
• 我们可以自己编写Lambda匿名函数来完成自定义逻辑的数据转换，想怎么转就怎么转，更自由

想完成这个功能可以基于RunnableLambda类实现
RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain
语法：RunnableLambda（函数对象或lambda匿名函数）

chain = first_prompt | model | (lambda ai_msg:{“name”:ai_msg.content}) | second_prompt | model | str_parser
跳过RunnableLambda，直接让函数加入链也是可以的。因为Runnable接口类在实现__or__的时候，支持Callable接口的实例

• 函数就是Callable接口的实例，|符号（底层是调用__or__）组链，是支持函数加入的。其本质是将函数自定义转换为RunnableLambda

from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
model = ChatTongyi(model="qwen3-max")
str_parser = StrOutputParser()

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅告知我名字，不要有额外信息"
)
second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)
# 函数入参：AIMessage -> dict
# my_func = RunnableLambda(lambda ai_msg:{"name":ai_msg.content})

# chain = first_prompt | model | my_func | second_prompt | model | str_parser
chain = first_prompt | model | (lambda ai_msg:{"name":ai_msg.content}) | second_prompt | model | str_parser
for chunk in chain.stream({"lastname":"张","gender":"女儿"}):
    print(chunk,end=" ",flush=True)

4.19 Memory临时会话记忆

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

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

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


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


base_chain = prompt | print_prompt | model | str_parser


store = {}      # key就是session，value就是InMemoryChatMessageHistory类对象
# 实现通过会话id获取InMemoryChatMessageHistory类对象
def get_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()

    return store[session_id]

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)


if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

4.20 Memory长期会话记忆

FileChatMessageHistory类实现，核心思路：

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

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

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

官方提供的：

import os, json
from typing import Sequence

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory


# message_to_dict:单个消息对象(BaseMessage类实例) -> 字典
# messages_from_dict：[字典、字典、...] -> [消息、消息、...]
# AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类

class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self,session_id,storage_path):
        self.session_id=session_id        # 会话id
        self.storage_path=storage_path    # 不同会话id的存储文件，所在的文件夹路径
        self.file_path = os.path.join(self.storage_path,self.session_id)   # 完整的文件路径
        # 确保文件夹是存在的
        os.makedirs(os.path.dirname(self.file_path),exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage])->None:
        # Sequence序列 实例list、tuple
        all_message = list(self.messages) # 已有的消息列表
        all_message.extend(messages) # 新的和已有的融合成一个list
        # 将数据同步写入到本地文件中
        # 类对象写入文件 -> 一堆二进制
        # 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件）
        # 官方message_to_dict:单个消息对象(BaseMessage类实例) -> 字典
        # new_message = []
        # for message in all_message:
        #     d = message_to_dict(message)
        #     new_message.append(d)
        # 列表推导式 等价于上面四句
        new_message = [message_to_dict(message) for message in all_message] # message_to_dict:单个消息对象(BaseMessage类实例) -> 字典
        # 将数据写入文件
        with open(self.file_path,'w',encoding='utf-8') as f:
            json.dump(new_message,f) # 字典或者列表转换为Json字符串

    @property # @property装饰器将message方法变成成员属性
    def messages(self) -> list[BaseMessage]:
        # 当前文件内：list[字典]
        try:
            with open(self.file_path,'r',encoding='utf-8') as f:
                message_data = json.load(f) # 返回值就是：list[字典]
                return messages_from_dict(message_data) # [字典、字典、...] -> [消息、消息、...]
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path,'w',encoding='utf-8') as f:
            json.dump([],f)


model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()

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

base_chain = prompt | print_prompt | model | str_parser

# 实现通过会话id获取InMemoryChatMessageHistory类对象
def get_history(session_id):
    return FileChatMessageHistory(session_id,'./chat_history')

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)

if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

4.21 CSVLoader 文档加载器

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

from lanchain_core.documents import Document
document = Document(
	page_content="hello,world!",metadata={"source":"https://example.com"}
)

可以看到Document类核心记录了

• page_content：文档内容
• metadata：文档元数据（字典）

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

• load():一次性加载全部文档
• lazy_load():延迟流式传输文档，对大型数据集很有用，避免内存溢出

LangChain内置了许多文档加载器，详细参见官方文档：
简单学习以下几种常见的文档加载器

• CSVLoader
• JSONLoader
• PDFLoader

from langchain_community.document_loaders import CSVLoader

loader = CSVLoader(
    file_path="./data/stu.csv",
    csv_args={
        "delimiter": ",", # 指定分隔符
        "quotechar": '"' , # 指定带有分隔符文本的引号包围是单引号还是双引号
        "fieldnames": ['a','b','c','d'] # 如果数据原本有表头，就不要下面的代码，如果没有可以使用
    },
    encoding="utf-8"
)
# 批量加载 .load() -> [Document,Document,...]
# documents = loader.load()
# for document in documents:
#     print(document)

# 懒加载 .lazy_load() -> 迭代器[Document]
for document in loader.lazy_load():
    print(document)

4.22 JSONLoader

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

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

• . 表示整个JSON对象(根)
• []表示数组
• . name表示抽取周杰伦
• . hobby表示抽取爱好数组
• . hobby[1]或 . hobby . [1]表示抽取跳
• . other . addr表示抽取地址深圳

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

• . [] . 得到3个字典
• . [] . name表示抽取全部的name，即得到3个name信息

from langchain_community.document_loaders import JSONLoader

# loader = JSONLoader(
#     file_path="./data/stu.json",
#     jq_schema=".other.addr"
# )
# loader = JSONLoader(
#     file_path="./data/stus.json",
#     jq_schema=".[].name",
#     text_content=False, # 告知JSONLoader 我抽取的不是字符串
# )
loader = JSONLoader(
    file_path="./data/stu_json_lines.json",
    jq_schema=".name",
    text_content=False, # 告知JSONLoader 我抽取的不是字符串
    json_lines=True,  # 告知JSONLoader 这是一个JSONLines文件（每一行都是一个独立的标准JSON）
)
print(loader.load())

4.23 TextLoader和文档分割器

读取文本文件（如.txt），将全部内容放入一个Document对象中
RecursiveCharacterTextSplitter，递归字符文本分割器，主要用于按自然段落分割大文档，是LangChain官方推荐的默认字符分割器

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader("./data/Python基础语法.txt", encoding="utf-8")

docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, # 分段的最大字符数
    chunk_overlap=50, # 分段之间允许重叠字符数
    separators=["\n\n","\n","。","，","！","？",",",".","!","?"," ",""], # 文本自然段落分隔的依据符号
    length_function=len, # 统计字符的依据函数
)
split_docs = splitter.split_documents(docs)
for doc in split_docs:
    print("="*20)
    print(doc)
    print("=" * 20)
print(len(split_docs))

4.24 PyPDFLoader

pip install pypdf

from langchain_community.document_loaders import PyPDFLoader

# loader = PyPDFLoader(
#     file_path="./data/pdf1.pdf",
#     mode="single", # 默认是page模式，每个页面形成一个Document文档对象 single模式：不管有多少页，只返回1个Document对象
# )
loader = PyPDFLoader(
    file_path="./data/pdf2.pdf",
    mode="single", # 默认是page模式，每个页面形成一个Document文档对象 single模式：不管有多少页，只返回1个Document对象
    password="itheima"
)
i = 0
for doc in loader.lazy_load():
    i += 1
    print(doc)
    print("="*20,i)

-	CSVLoader	JsonLoader	TextLoader	PyPDFLoader
参数	file_path=“”,csv_args={“delimiter”:“,”,“quotechar”:‘"’,“fieldnames”:[“name”,“age”,“gender”],source_column=“”,}	file_path=“”,jq_schema=“”,text_content=False,json_lines=True,	“文件路径”,encoding=“utf-8”	file_path=“”,model=‘page’,password=‘password’
加载器	load(),load_lazy()	load()	load() + RecursiveCharacterTextSplitter()	load(),load_lazy()

4.25 VectorStores向量存储

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

如图，这部分开发主要涉及到：

• 如何文本转向量（千问已经学习）
• 创建向量存储，基于向量存储完成
  • 存入向量 -> add_documents
  • 删除向量 -> delete
  • 向量检索 -> similarity_search

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

# Chroma 向量数据库（轻量级）

vector_store = Chroma(
    collection_name="test",  # 当前向量存储起个名字，类似数据库表名称
    embedding_function=DashScopeEmbeddings(),  # 嵌入模型
    persist_directory="./chroma_db",  # 指定数据存放的文件夹
)

loader = CSVLoader(
    file_path="./data/info.csv",
    encoding="utf-8",
    source_column="source" # 指定本条数据的来源是哪里
)

documents = loader.load()
print(documents[0])

# 向量存储的新增、删除、检索
vector_store.add_documents(
    documents=documents, # 被添加的文件，类型list[Document]
    ids=[f'id{i}' for i in range(1, len(documents) + 1)], # 给添加的文档提供id(字符串) list[str]
)
vector_store.delete(["id1", "id2"])
result = vector_store.similarity_search(
    "Python是不是简单易学呀",3,filter={"source":"黑马程序员"}
)
print(result)

4.26 基于向量检索构建提示词

'''
提示词：用户的提问+向量库中检索到的参考资料
'''
from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate(
    [
        ("system","以我提供的已知参考资料为主，简洁和专业的回答用户问题，参考资料：{context}."),
        ("user","用户提问:{input}")
    ]
)
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))

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

input_text = "怎么减肥？"

# 检索向量库
result = vector_store.similarity_search(input_text,2)
reference_text = "["

for doc in result:
    reference_text += doc.page_content
reference_text += "]"

def print_prompt(prompt):
    print(prompt.to_string())
    print("="*20)
    return prompt
# chain
chain = prompt | print_prompt | model | StrOutputParser()

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

4.27 RunnablePassthrough的使用

让向量检索加入链？使用RunnablePassthrough类

'''
提示词：用户的提问+向量库中检索到的参考资料
'''
from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate(
    [
        ("system","以我提供的已知参考资料为主，简洁和专业的回答用户问题，参考资料：{context}."),
        ("user","用户提问:{input}")
    ]
)
vector_store = InMemoryVectorStore(embedding=DashScopeEmbeddings(model="text-embedding-v4"))

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

input_text = "怎么减肥？"

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

def format_func(docs:list[Document]):
    if not docs:
        return "无参考资料"
    format_str = "["
    for doc in docs:
        format_str += doc.page_content
    format_str += "]"

    return format_str


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

chain = (
    {"input":RunnablePassthrough(),"context":retriever | format_func} | prompt | print_prompt | model | StrOutputParser()
)
'''
retriever:
    - 输入：用户的提问       str
    - 输出：向量库的检索结果  list[Document]
    
prompt:
    - 输入：用户的提问 + 向量库的检索结果   dict
    - 输出：完整的提示词                  PromptValue
'''
res = chain.invoke(input_text)
print(res)

第5章 RAG项目

gitee地址

5.1 RAG项目案例介绍

RAG即检索、增强和生成，其主要分为2条线：

• 离线处理：向私有知识库（向量存储）源源不断添加私有知识文档
  • 向知识库添加来自未来的知识文档（基于模型训练完成时间）
  • 向模型添加私有知识文档
  • 给出模型参考资料，规避模型幻觉（一本正经的胡说八道）
• 在线处理：用户提问会先基于私有知识库做检索，获取参考资料，同步组装新提示词询问大模型获取结果

本次项目以“某东商品衣服”为例，以衣服属性构建本地知识。使用者可以自由更新本地知识，用户问题的答案也是基于本地知识生成的

5.2 文本上传WEB服务

离线流程：

app_file_uploader

'''
基于Streamlit完成WEB网页上传服务
pip install streamlit

cmd 到项目文件夹下 streamlit run 文件名

'''
import streamlit as st

# 添加网页标题
st.title("知识库更新服务")

# file_uploader
uploader_file = st.file_uploader(
    "请上传TXT文件",
    type=['txt'],
    accept_multiple_files=False, # False表示仅接受一个我呢见的上传
)
if uploader_file is not None:
    # 提取文件的信息
    file_name = uploader_file.name
    file_type = uploader_file.type
    file_size = uploader_file.size / 1024 # KB

    st.subheader(f'文件名：{file_name}')
    st.write(f'格式：{file_type} | 大小：{file_size:.2f} KB')

    # get_value -> bytes -> decode('utf-8')
    text = uploader_file.getvalue().decode('utf-8')
    st.write(text)

5.3 md5工具函数开发

def check_md5(md5_str:str):
    """检查传入的md5字符串是否已经被处理过了
        return False(md5未处理过)  True(已经处理过，已有记录)
    """
    if not os.path.exists(config.md5_path):
        # if进入表示文件不存在，那肯定没有处理过这个md5了
        open(config.md5_path, 'w',encoding="utf-8").close()
        return False
    else:
        for line in open(config.md5_path,'r',encoding="utf-8").readlines():
            line = line.strip() # 处理字符串前后的空格和回车
            if line == md5_str:
                return True
        return False

def save_md5(md5_str:str):
    """将传入的md5字符串，记录到文件内保存"""
    with open(config.md5_path,'a',encoding="utf-8") as f:
        f.write(md5_str + '\n')

def get_string_md5(input_str:str,encoding='utf-8'):
    """将传入的字符串转为md5字符串"""

    # 将字符串转换为bytes字节数组
    str_bytes = input_str.encode(encoding=encoding)
    # 创建md5对象
    md5_obj = hashlib.md5() # 得到md5对象
    md5_obj.update(str_bytes) # 更新内容（传入即将要转换的字节数组）
    md5_hex = md5_obj.hexdigest() # 得到md5的十六进制字符串
    return md5_hex

5.4 知识库更新服务

class KnowledgeBaseService(object):
    def __init__(self):
        # 如果文件夹不存在则创建，如果存在则跳过
        os.makedirs(config.persist_directory, exist_ok=True)

        self.chroma = Chroma(
            collection_name=config.collection_name, # 数据库的表名
            embedding_function=DashScopeEmbeddings(model="text-embedding-v4"),
            persist_directory=config.persist_directory, # 数据库本地存储文件
        ) # 向量存储的实例 Chroma向量库对象
        self.spliter = RecursiveCharacterTextSplitter(
            chunk_size=config.chunk_size, # 分割后的文本最大长度
            chunk_overlap=config.chunk_overlap, # 连续文本段之间的字符重叠数量
            separators=config.separators, # 自然段落划分的符号
            length_function=len, # 使用python自带的len函数做长度统计的依据
        ) # 文本分割器的对象

    def upload_by_str(self,data:str,filename):
        """将传入的字符串，进行向量化，存入向量数据库"""
        # 先得到传入字符串的md5值
        md5_hex = get_string_md5(data)

        if check_md5(md5_hex):
            return "[跳过]内容已经存在知识库中"
        if len(data) > config.max_split_char_number:
            knowledge_chunks:list[str] = self.spliter.split_text(data)
        else:
            knowledge_chunks = [data]

        metadata = {
            "source":filename,
            "create_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "operator":"lhl"
        }
        self.chroma.add_texts( # 内容就加载到向量库中
            # iterable -> list -> tuple
            knowledge_chunks,
            metadatas = [metadata for _ in knowledge_chunks],
        )
        save_md5(md5_hex)
        return "[成功]内容已经成功载入向量库中"

5.5 在线流程向量存储服务代码

from langchain_chroma import Chroma
import config_data as config
class VectorStoreService(object):
    def __init__(self, embedding):
        """
        :param embedding: 嵌入模型的传入
        """
        self.embedding = embedding
        self.vector_store = Chroma(
            collection_name=config.collection_name,
            embedding_function=self.embedding,
            persist_directory=config.persist_directory,
        )

    def get_retriever(self):
        """返回向量检索器，方便加入链"""
        return self.vector_store.as_retriever(search_kwargs={"k":config.similarity_threshold})

if __name__ == "__main__":
    from langchain_community.embeddings import DashScopeEmbeddings
    retriever = VectorStoreService(DashScopeEmbeddings(model="text-embedding-v4")).get_retriever()
    res = retriever.invoke("我的体重180斤，尺码推荐")
    print(res)

5.6 rag服务核心代码开发

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document
import config_data as config
from vector_store import VectorStoreService
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

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


class RagService(object):
    def __init__(self):
        self.vector_service = VectorStoreService(
            embedding=DashScopeEmbeddings(model=config.embedding_model_name),
        )
        self.prompt_template = ChatPromptTemplate.from_messages(
            [
                ("system","以我提供的已知参考资料为主，"
                 "简洁和专业的回答用户问题，参考资料：{context}。"),
                ("user","请回答用户提问：{input}")
            ]
        )
        self.chat_model = ChatTongyi(model=config.chat_model_name)
        self.chain = self.__get_chain()

    def __get_chain(self):
        """获取最终的执行链"""
        retriever = self.vector_service.get_retriever()

        def format_document(docs:list[Document]):
            if not docs:
                return "无相关参考资料"
            formatted_str = ""
            for doc in docs:
                formatted_str += f"文档片段：{doc.page_content}\n文档元数据：{doc.metadata}\n\n"
            return formatted_str

        chain = (
            {
                "input":RunnablePassthrough(),
                "context":retriever | format_document
            } | self.prompt_template | print_prompt | self.chat_model | StrOutputParser()
        )
        return chain
if __name__ == '__main__':
    res = RagService().chain.invoke("我体重180斤，尺码推荐")
    print(res)

5.7 历史会话记录功能的实现

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

# 实现通过会话id获取InMemoryChatMessageHistory类对象
def get_history(session_id):
    return FileChatMessageHistory(session_id,'./chat_history')


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self,session_id,storage_path):
        self.session_id=session_id        # 会话id
        self.storage_path=storage_path    # 不同会话id的存储文件，所在的文件夹路径
        self.file_path = os.path.join(self.storage_path,self.session_id)   # 完整的文件路径
        # 确保文件夹是存在的
        os.makedirs(os.path.dirname(self.file_path),exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage])->None:
        # Sequence序列 实例list、tuple
        all_message = list(self.messages) # 已有的消息列表
        all_message.extend(messages) # 新的和已有的融合成一个list
        # 将数据同步写入到本地文件中
        # 类对象写入文件 -> 一堆二进制
        # 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件）
        # 官方message_to_dict:单个消息对象(BaseMessage类实例) -> 字典
        # new_message = []
        # for message in all_message:
        #     d = message_to_dict(message)
        #     new_message.append(d)
        # 列表推导式 等价于上面四句
        new_message = [message_to_dict(message) for message in all_message]
        # 将数据写入文件
        with open(self.file_path,'w',encoding='utf-8') as f:
            json.dump(new_message,f)

    @property # @property装饰器将message方法变成成员属性
    def messages(self) -> list[BaseMessage]:
        # 当前文件内：list[字典]
        try:
            with open(self.file_path,'r',encoding='utf-8') as f:
                message_data = json.load(f) # 返回值就是：list[字典]
                return messages_from_dict(message_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path,'w',encoding='utf-8') as f:
            json.dump([],f)

5.8 聊天页面开发

import time
import streamlit as st
from rag import RagService
import config_data as config

# 标题
st.title("智能客服")
st.divider() # 分隔符

if "message" not in st.session_state:
    st.session_state["message"] = [{"role":"assistant","content":"你好，有什么可以帮助你？"}]

if "rag" not in st.session_state:
    st.session_state["rag"] = RagService()

for message in st.session_state["message"]:
    st.chat_message(message["role"]).write(message["content"])

# 在页面最下方提供用户输入栏
prompt = st.chat_input()

if prompt:
    # 在页面输出客户的提问
    st.chat_message("user").write(prompt)

    st.session_state["message"].append({"role":"user","content":prompt})
    ai_res_list = []
    with st.spinner("AI思考中......"):
        # res = st.session_state["rag"].chain.invoke({"input":prompt},config.session_config)
        # st.chat_message("assistant").write(res)
        res_stream = st.session_state["rag"].chain.stream({"input": prompt}, config.session_config)
        # yield
        def capture(generator, cache_list):
            for chunk in generator:
                cache_list.append(chunk)
                yield chunk

        st.chat_message("assistant").write_stream(capture(res_stream,ai_res_list))
        st.session_state["message"].append({"role": "assistant", "content": "".join(ai_res_list)})

第6章 Agent智能体

6.1 Agent智能体介绍

Agent是一种能够自主规划、决策、执行任务的组件，核心是让大语言模型(LLM)根据任务需求，选择并调用工具，完成单靠模型自身无法解决的复杂问题

核心特点：

• 目标驱动：围绕用户的具体任务目标展开工作
• 工具调用能力：能连接外部工具，弥补LLM的局限性
• 自主决策与迭代：不需要人工干预，能根据工具返回的结果，判断是否需要调用工具，或直接生成最终答案

以电商商品问答为例：

普通chain	Agent
执行流程固定，按预设步骤运行	执行流程动态，根据任务和结果自主调整
工具调用路径写死在代码里	工具选择由LLM思考决定
适合简单、标准化任务	适合复杂、多步骤、需要决策的任务

Agent智能体 = 大语言模型（大脑）+工具集（手脚）+决策逻辑（思维）
会做事（影响现实世界）的智能助手

6.2 Agent智能体初体验

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="查询天气")
def get_weather() -> str:
    return "晴天"

agent = create_agent(
    model = ChatTongyi(model="qwen3-max"),  # 智能体的大脑LLM
    tools=[get_weather],   # 向智能体提供工具列表
    system_prompt="你是一个聊天助手，可以回答用户问题。"
)
res = agent.invoke(
    {
        "messages":[
            {"role":"user","content":"明天深圳的天气如何？"}
        ]
    }
)
for msg in res["messages"]:
    print(type(msg).__name__,msg.content)
    
"""
HumanMessage 明天深圳的天气如何？
AIMessage 
ToolMessage 晴天
AIMessage 明天深圳的天气是晴天。记得做好防晒措施，祝你有愉快的一天！
"""

6.3 Agent的流式输出

通过create_agent方法可以创建Agent对象，其也是Runnable接口的子类实现，所以也拥有：

• invoke：执行，一次性得到完整结果
• stream：执行，流式得到结果

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool


@tool(description="获取股价，传入股票名称，返回字符串信息")
def get_price(name:str)->str:
    return f'股票{name}的价格是20元'

@tool(description="获取股票信息，传入股票名称，返回字符串信息")
def get_info(name:str)->str:
    return f'股票{name}，是一家A股上市公司，专注于IT职业教育。'

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_price,get_info],
    system_prompt="你是一个智能助手，可以回答股票相关问题，记住请告知我思考过程，让我知道你为什么调用某工具"
)

res = agent.stream(
    {
        "messages":[
            {"role":"user","content":"传智教育股价多少，并介绍一下"}
        ]
    },
    stream_mode="values"
)
for chunk in res:
    latest_message = chunk["messages"][-1]
    if latest_message.content:
        print(type(latest_message).__name__,latest_message.content)
    try:
        if latest_message.tool_calls:
            print(f"工具调用：{[tc['name'] for tc in latest_message.tool_calls]}")
    except AttributeError as e:
        pass

6.4 Agent的ReAct行动框架

Agent ReAct是大模型智能体的核心思考与行动框架，全程Reasoning + Acting(推理 + 行动)，是让Agent像人类一样【思考问题->制定策略->执行行动->验证结果】的关键逻辑

一个典型的ReAct范式的Agent如图所示：

• 思考Reasoning：分析问题，判断现有信息是否足够，明确下一步
  • 即模型决策是否需要调用外部工具获取更多信息用来回答
• 行动Action：执行思考阶段指定的策略
  • 即基于模型决策结果，调用工具获取信息
• 观察Obsercation：获取行动的结果，提取有效信息
  • 即获取工具返回值即判断工具是否正常工作为下一轮思考提供信息
• （再）思考->（再）行动->（再）观察->循环往复直到结束

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool


@tool(description="获取体重，返回值是整数，范围千克")
def get_weight()->int:
    return 60

@tool(description="返回身高，返回值是整数，单位是厘米")
def get_height()->int:
    return 168

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weight,get_height],
    system_prompt="""你是严格遵守ReAct框架的智能体，必须按【思考->行动->观察->再思考】的流程解决问题，
    且**每轮仅能思考并调用1个工具**，禁止单次调用多个工具。
    并告知我你的思考过程。工具的调用原因，按思考、行动、观察三个结构告知我"""
)

res = agent.stream(
    {
        "messages":[
            {"role":"user","content":"计算我的BMI"}
        ]
    },
    stream_mode="values"
)
for chunk in res:
    latest_message = chunk["messages"][-1]
    if latest_message.content:
        print(type(latest_message).__name__,latest_message.content)
    try:
        if latest_message.tool_calls:
            print(f"工具调用：{[tc['name'] for tc in latest_message.tool_calls]}")
    except AttributeError as e:
        pass

6.5 Agent的middleware中间件

中间件的作用是对智能体的每一步工作进行控制和自定义的执行
作用场景：

• 日志记录、分析、调试
• 转换提示词、工具选择
• 重试、备用、提前终止等逻辑控制
• 安全防护、个人身份检测等

LangChain中内置了一些基础的中间件，地址
中间件通过Hooks钩子来实现拦截，自定义中间件可以简单的使用装饰器来定义
节点式钩子（执行点顺序拦截）：

• before_agent：agent执行之前拦截
• after_agent：agent执行后拦截
• before_model：模型执行前拦截
• after_model：模型执行后拦截

针对工具和模型的包装式钩子：

• wrap_model_call：每个模型调用时候拦截
• wrap_tool_call：每个工具调用时候拦截

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent, after_agent, before_model, after_model, wrap_model_call, wrap_tool_call
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
from langgraph.runtime import Runtime


@tool(description="查询天气，传入城市名称字符串，返回字符串天气信息")
def get_weather(city:str) -> str:
    return f"{city}天气:晴天"

"""
1. agent执行前
2. agent执行后
3. model执行前
4. model执行后
5. 工具执行中
6. 模型执行中
"""

@before_agent
def log_before_agent(state:AgentState, runtime:Runtime)->None:
    # agent执行前会调用这个函数并传入state和runtime两个对象
    print(f"[before agent]agent启动，并附带{len(state['messages'])}消息")

@after_agent
def log_after_agent(state:AgentState, runtime:Runtime)->None:
    print(f"[after agent]agent结束，并附带{len(state['messages'])}消息")

@before_model
def log_before_model(state:AgentState, runtime:Runtime)->None:
    print(f"[before model]模型即将调用，并附带{len(state['messages'])}消息")

@after_model
def log_after_model(state:AgentState, runtime:Runtime)->None:
    print(f"[after model]模型调用结束，并附带{len(state['messages'])}消息")

@wrap_model_call
def model_call_hook(request,handler):
    print("模型调用啦")
    return handler(request)

@wrap_tool_call
def monitor_tool(request,handler):
    print(f"工具执行：{request.tool_call['name']}")
    print(f"工具执行传入参数：{request.tool_call['args']}")
    return handler(request)

agent = create_agent(
    model = ChatTongyi(model="qwen3-max"),  # 智能体的大脑LLM
    tools=[get_weather],   # 向智能体提供工具列表
    middleware=[log_before_agent,log_after_agent,log_before_model,log_after_model,model_call_hook,monitor_tool],
)
res = agent.invoke({"messages":[{"role":"user","content":"深圳今天的天气如何呀，如何穿衣"}]})
print("*"*20+"\n",res)

6.6 Agent智能体项目介绍

6.7 [Agent项目]日志和路径工具开发

import logging
import os
from datetime import datetime
from path_tool import get_abs_path

# 日志保存的根目录
LOG_ROOT = get_abs_path("logs")

# 确保日志的目录存在
os.makedirs(LOG_ROOT, exist_ok=True)

# 日志的格式配置
DEFAULT_LOG_FORMAT = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s"
)

def get_logger(
        name:str = "agent",
        console_level:int = logging.INFO,
        file_level:int = logging.DEBUG,
        log_file = None,
) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 避免重复添加handler
    if logger.handlers:
        return logger
    # 控制台Handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(console_level)
    console_handler.setFormatter(DEFAULT_LOG_FORMAT)

    logger.addHandler(console_handler)

    # 文件Handler
    if not log_file:
        log_file = os.path.join(LOG_ROOT, f"{name}_{datetime.now().strftime('%Y%m%d')}.log")

    file_handler = logging.FileHandler(log_file,encoding="utf-8")
    file_handler.setLevel(file_level)
    file_handler.setFormatter(DEFAULT_LOG_FORMAT)

    logger.addHandler(file_handler)
    return logger

# 快捷获取日志器
logger = get_logger()

if __name__ == '__main__':
    logger.info("信息日志")
    logger.error("错误日志")
    logger.warning("警告日志")
    logger.debug("调试日志")

"""
为整个工程提供统一的绝对路径
"""

import os

def get_project_root() -> str:
    """
    获取工程所在的根目录
    :return: 字符串根目录
    """
    # 当前文件的绝对路径
    current_dir = os.path.abspath(__file__)
    # 获取工程的根目录，先获取文件所在的文件夹绝对路径
    current_dir = os.path.dirname(current_dir)
    # 获取工程根目录
    project_root = os.path.dirname(current_dir)
    return project_root

def get_abs_path(relative_path:str) -> str:
    """
    传递相对路径，得到绝对路径
    :param relative_path: 相对路径
    :return: 绝对路径
    """
    project_root = get_project_root()
    return os.path.join(project_root, relative_path)

if __name__ == '__main__':
    print(get_abs_path("config/config.txt"))

6.8 [Agent项目]配置工具、文件工具和提示词加载工具开发

"""
yaml
k: v
"""
import yaml
from path_tool import get_abs_path

def load_rag_config(config_path: str=get_abs_path("config/rag.yml"),encoding:str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

def load_chroma_config(config_path: str=get_abs_path("config/chroma.yml"),encoding:str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

def load_prompts_config(config_path: str=get_abs_path("config/prompts.yml"),encoding:str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

def load_agent_config(config_path: str=get_abs_path("config/agent.yml"),encoding:str="utf-8"):
    with open(config_path, "r", encoding=encoding) as f:
        return yaml.load(f, Loader=yaml.FullLoader)

rag_conf = load_rag_config()
chroma_conf = load_chroma_config()
prompts_conf = load_prompts_config()
agent_conf = load_agent_config()

if __name__ == '__main__':
    print(rag_conf["chat_model_name"])

import os
import hashlib
from logger_handler import logger
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader

def get_file_hex(filepath: str):  # 获取文件的md5的十六进制字符串
    if not os.path.exists(filepath):
        logger.error(f"[md5计算]文件{filepath}不存在")
        return

    if not os.path.isfile(filepath):
        logger.error(f"[md5计算]路径{filepath}不是文件")
        return

    md5_obj = hashlib.md5()
    chunk_size = 4096 # 4KB分片，避免文件过大爆内存
    try:
        with open(filepath, "rb") as f: # 必须二进制读取
            while chunk := f.read(chunk_size):
                md5_obj.update(chunk)
            """
            chunk = f.read(chunk_size)
            while chunk:
                md5_obj.update(chunk)
                chunk = f.read(chunk_size)
            """
            md5_hex = md5_obj.hexdigest()
            return md5_hex
    except Exception as e:
        logger.error(f"计算文件{filepath}md5失败，{str(e)}")
        return None

def listdir_with_allowed_type(path: str, allowed_types: tuple[str]):  # 返回文件夹内的文件列表（允许的文件后缀）
    files = []

    if not os.path.isdir(path):
        logger.error(f"[listdir_with_allowed_type]{path}不是文件夹")
        return allowed_types

    for f in os.listdir(path):  # os.listdir列出文件夹中的文件
        if f.endswith(allowed_types):
            files.append(os.path.join(path, f))

    return tuple(files)

def pdf_loader(filepath: str, password: None) -> list[Document]:
    return PyPDFLoader(filepath, password).load()

def txt_loader(filepath: str) -> list[Document]:
    return TextLoader(filepath).load()

from config_handler import prompts_conf
from path_tool import get_abs_path
from logger_handler import logger

def load_system_prompts():
    try:
        system_prompt_path = get_abs_path(prompts_conf["main_prompt_path"])
    except KeyError as e:
        logger.error(f"[load_system_prompts]在yaml配置项中没有main_prompt_path配置项")
        raise e

    try:
        return open(system_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[load_system_prompts]解析系统提示词出错，{str(e)}")

def load_rag_prompts():
    try:
        rag_prompt_path = get_abs_path(prompts_conf["rag_summarize_prompt_path"])
    except KeyError as e:
        logger.error(f"[load_rag_prompts]在yaml配置项中没有rag_summarize_prompt_path配置项")
        raise e

    try:
        return open(rag_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[load_rag_prompts]解析系统提示词出错，{str(e)}")

def load_report_prompts():
    try:
        report_prompt_path = get_abs_path(prompts_conf["report_prompt_path"])
    except KeyError as e:
        logger.error(f"[load_report_prompts]在yaml配置项中没有report_prompt_path配置项")
        raise e

    try:
        return open(report_prompt_path, "r", encoding="utf-8").read()
    except Exception as e:
        logger.error(f"[load_report_prompts]解析系统提示词出错，{str(e)}")

if __name__ == '__main__':
    # print(load_system_prompts())
    # print(load_rag_prompts())
    print(load_report_prompts())

6.9 [Agent项目]向量存储服务开发

import sys
import os

# 获取当前文件的目录 (rag目录)
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (P6_AI大模型RAG与智能体开发_Agent项目目录)
project_root = os.path.dirname(current_dir)
# 将项目根目录添加到 Python 路径
sys.path.append(project_root)

import os.path
from langchain_chroma import Chroma
from langchain_core.documents import Document
from model.factory import embed_model
from langchain_text_splitters import RecursiveCharacterTextSplitter
from utils.config_handler import chroma_conf
from utils.path_tool import get_abs_path
from utils.file_handler import pdf_loader, txt_loader, listdir_with_allowed_type, get_file_md5_hex
from utils.logger_handler import logger

class VectorStoreService:
    def __init__(self):
        self.vector_store = Chroma(
            collection_name = chroma_conf["collection_name"],
            embedding_function = embed_model,
            persist_directory = chroma_conf["persist_directory"],
        )
        self.spliter = RecursiveCharacterTextSplitter(
            chunk_size=chroma_conf["chunk_size"],
            chunk_overlap=chroma_conf["chunk_overlap"],
            separators=chroma_conf["separators"],
            length_function=len,
        )

    def get_retriever(self):
        return self.vector_store.as_retriever(search_kwargs={"k":chroma_conf["k"]})

    def load_document(self):
        """
        从数据文件内读取数据文件（data文件夹中），转为向量存入向量库
        要计算文件的md5做去重
        :return: None
        """
        def check_md5_hex(md5_for_check: str):
            if not os.path.exists(get_abs_path(chroma_conf["md5_hex_store"])):
                # 创建文件
                open(get_abs_path(chroma_conf["md5_hex_store"]), "w", encoding='utf-8').close()
                return False # md5没处理过

            with open(get_abs_path(chroma_conf["md5_hex_store"]),"r",encoding="utf-8") as f:
                for line in f.readlines():
                    line = line.strip()
                    if line == md5_for_check:
                        return True # md5处理过
                return False

        def save_md5_hex(md5_for_check: str):
            with open(get_abs_path(chroma_conf["md5_hex_store"]), "a", encoding='utf-8') as f:
                f.write(md5_for_check + "\n")

        def get_file_documents(read_path:str):
            if read_path.endswith("txt"):
                return txt_loader(read_path)

            if read_path.endswith("pdf"):
                return pdf_loader(read_path,password=None)

            return []

        allowed_files_path: list[str] = listdir_with_allowed_type(
            get_abs_path(chroma_conf["data_path"]),
            tuple(chroma_conf["allow_knowledge_file_type"])
        )
        for path in allowed_files_path:
            # 获取文件的md5
            md5_hex = get_file_md5_hex(path)
            if check_md5_hex(md5_hex):
                logger.info(f"[加载文件]{path}内容已经存在知识库内，跳过")
                continue
            try:
                documents:list[Document] = get_file_documents(path)
                if not documents:
                    logger.warning(f"[加载知识库]{path}内没有有效文件内容，跳过")
                    continue

                split_document:list[Document] = self.spliter.split_documents(documents)
                if not split_document:
                    logger.warning(f"[加载知识库]{path}分片后没有有效文件内容，跳过")
                    continue

                # 将内容存入向量库
                self.vector_store.add_documents(split_document)

                # 记录这个已经处理好的文件的md5,避免下次重复加载
                save_md5_hex(md5_hex)
                logger.info(f"[加载知识库]{path}内容加载成功")

            except Exception as e:
                # exc_info 为True会记录详细的报错堆栈，如果为False仅记录报错信息本身
                logger.error(f"[加载知识库]{path}加载失败：{str(e)}", exc_info=True)
                continue

if __name__ == '__main__':
    vs = VectorStoreService()
    vs.load_document()
    retriever = vs.get_retriever()
    res = retriever.invoke("迷路")
    for f in res:
        print(f.page_content)
        print("-"*20)

9.10 [Agent项目]RAG总结服务开发

"""
总结服务类：用户提问，搜索参考资料，将提问和参考资料提交给模型，让模型总结回复
"""

import sys
import os

from langchain_core.documents import Document

# 获取当前文件的目录 (rag目录)
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (P6_AI大模型RAG与智能体开发_Agent项目目录)
project_root = os.path.dirname(current_dir)
# 将项目根目录添加到 Python 路径
sys.path.append(project_root)


from rag.vector_store import VectorStoreService
from utils.prompt_loader import load_rag_prompts
from model.factory import chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

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

class RagSummarizeService(object):
    def __init__(self):
        self.vector_store = VectorStoreService()
        self.retriever = self.vector_store.get_retriever()
        self.prompt_text = load_rag_prompts()
        self.prompt_template = PromptTemplate.from_template(self.prompt_text)
        self.model = chat_model
        self.chain = self._init_chain()

    def _init_chain(self):
        chain = self.prompt_template | print_prompt | self.model | StrOutputParser()
        return chain

    def retriever_docs(self, query:str) -> list[Document]:
        return self.retriever.invoke(query)

    def rag_summarize(self, query:str) -> str:
        context_docs = self.retriever_docs(query)
        context = ""
        counter = 0
        for doc in context_docs:
            counter += 1
            context += f"【参考资料{counter}】：参考资料：{doc.page_content} | 参考元数据：{doc.metadata}"

        return self.chain.invoke(
            {
                "input":query,
                "context":context
            }
        )

if __name__ == '__main__':
    rag = RagSummarizeService()
    print(rag.rag_summarize("小户型适合哪些扫地机器人"))

9.11 [Agent项目]tools工具开发

import sys
import os

from langchain_core.documents import Document

# 获取当前文件的目录 (rag目录)
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (P6_AI大模型RAG与智能体开发_Agent项目目录)
project_root = os.path.dirname(current_dir)
# 将项目根目录添加到 Python 路径
sys.path.append(project_root)

from langchain_core.tools import tool
from rag.rag_service import RagSummarizeService
import random
from utils.config_handler import agent_conf
from utils.path_tool import get_abs_path
from utils.logger_handler import logger

rag = RagSummarizeService()

user_ids = ["1001","1002","1003","1004","1005","1006","1007","1008","1009","1010",]
month_arr = ["2025-01","2025-02","2025-03","2025-04","2025-05","2025-06","2025-07","2025-08","2025-09","2025-10","2025-11","2025-12"]
external_data = {}

@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str) -> str:
    return rag.rag_summarize(query)

@tool(description="获取指定城市的天气，以消息字符串的形式返回")
def get_weather(city: str) -> str:
    return f"城市{city}天气为晴天，气温26摄氏度，空气湿度50%，南风1级，AQI21，最近6小时降雨概率低"

@tool(description="获取用户所在城市的名称，以纯字符串形式返回")
def get_user_location() -> str:
    return random.choice(["深圳","合肥","杭州"])

@tool(description="获取用户的ID、以纯字符串形式返回")
def get_user_id() -> str:
    return random.choice(user_ids)

@tool(description="获取当前月份，以纯字符串形式返回")
def get_current_month() -> str:
    return random.choice(month_arr)

def generate_external_data():

    """
    {
        "user_id":{
            "month":{"特征",xxx,"效率":xxx,....},
            "month":{"特征",xxx,"效率":xxx,....},
            "month":{"特征",xxx,"效率":xxx,....},
            ...
        },
        "user_id":{
            "month":{"特征",xxx,"效率":xxx,....},
            "month":{"特征",xxx,"效率":xxx,....},
            "month":{"特征",xxx,"效率":xxx,....},
            ...
        },
        "user_id":{
            "month":{"特征",xxx,"效率":xxx,....},
            "month":{"特征",xxx,"效率":xxx,....},
            "month":{"特征",xxx,"效率":xxx,....},
            ...
        },
        ...
    }
    """
    if not external_data:
        external_data_path = get_abs_path(agent_conf["external_data_path"])

        if not os.path.exists(external_data_path):
            raise FileNotFoundError(f"外部数据文件{external_data_path}不存在")

        with open(external_data_path, "r", encoding="utf-8") as f:
            for line in f.readlines()[1:]:
                arr: list[str] = line.strip().split(",")
                user_id: str = arr[0].replace('"','')
                feature:str = arr[1].replace('"','')
                efficiency:str = arr[2].replace('"','')
                consumables:str = arr[3].replace('"','')
                comparison:str = arr[4].replace('"','')
                time:str = arr[5].replace('"','')
                if user_id not in external_data:
                    external_data[user_id] = {}

                external_data[user_id][time] = {
                    "特征":feature,
                    "效率":efficiency,
                    "耗材":consumables,
                    "对比":comparison,
                }

@tool(description="从外部系统中获取用户的使用记录，以纯字符串形式返回，如果未检索到返回空字符串")
def fetch_external_data(user_id:str, month:str) -> str:
    generate_external_data()
    try:
        return external_data[user_id][month]
    except KeyError:
        logger.warning(f"[fetch_external_data]未能检索到用户：{user_id}在{month}的使用记录数据")
        return ""

if __name__ == '__main__':
    print(fetch_external_data("1001", "2025-01"))

9.12 [Agent项目]中间件和Agent创建

import sys
import os

from langchain_core.documents import Document

# 获取当前文件的目录 (rag目录)
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (P6_AI大模型RAG与智能体开发_Agent项目目录)
project_root = os.path.dirname(current_dir)
# 将项目根目录添加到 Python 路径
sys.path.append(project_root)


from langchain.agents import AgentState
from langgraph.runtime import Runtime
from langgraph.types import Command
from langchain.agents.middleware import wrap_tool_call, before_model, dynamic_prompt, ModelRequest
from langchain.tools.tool_node import ToolCallRequest
from typing import Callable
from langchain_core.messages import ToolMessage
from utils.logger_handler import logger
from utils.prompt_loader import load_report_prompts, load_system_prompts


@wrap_tool_call
def monitor_tool(
        # 请求的数据封装
        request:ToolCallRequest,
        # 执行的函数本身
        handler: Callable[[ToolCallRequest],ToolMessage | Command],
):  # 工具执行的监控
    logger.info(f"[monitor_tool]执行工具：{request.tool_call['name']}")
    logger.info(f"[monitor_tool]传入参数：{request.tool_call['args']}")
    try:
        result = handler(request)
        logger.info(f"[monitor_tool]工具{request.tool_call['name']}调用成功")

        if request.tool_call['name'] == 'fill_context_for_report':
            request.runtime.context["report"] = True
        return result
    except Exception as e:
        logger.error(f"工具{request.tool_call['name']}调用失败，原因：{str(e)}")
        raise e

@before_model
def log_before_model(
        state: AgentState, # 整个Agent智能体中的状态记录
        runtime: Runtime # 记录了整个执行过程中的上下文信息
):  # 在模型执行前输出日志
    logger.info(f"[log_before_model]即将调用模型，带有{len(state['messages'])}条消息。")
    logger.debug(f"[log_before_model]{type(state['messages'][-1].__name__)} | {state['messages'][-1].content.strip()}")
    return None

@dynamic_prompt  # 每一次在生成提示词之前，调用此函数
def report_prompt_switch(request: ModelRequest):  # 动态切换提示词
    is_report = request.runtime.context.get("report", False) # 拿到字典 拿不到就是False
    if is_report: # 是报告生成场景，返回报告生成提示词内容
        return load_report_prompts()
    return load_system_prompts()

import sys
import os

# 获取当前文件的目录 (rag目录)
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (P6_AI大模型RAG与智能体开发_Agent项目目录)
project_root = os.path.dirname(current_dir)
# 将项目根目录添加到 Python 路径
sys.path.append(project_root)


from langchain.agents import create_agent
from model.factory import chat_model
from utils.prompt_loader import load_system_prompts
from agent.tools.agent_tools import rag_summarize, get_weather, get_user_location, get_user_id, get_current_month, fetch_external_data, fill_context_for_report
from agent.tools.middleware import monitor_tool, log_before_model, report_prompt_switch

class ReactAgent:
    def __init__(self):
        self.agent = create_agent(
            model=chat_model,
            system_prompt=load_system_prompts(),
            tools=[rag_summarize,get_weather,get_user_location,get_user_id,get_current_month,fill_context_for_report,fetch_external_data],
            middleware=[monitor_tool,log_before_model,report_prompt_switch],
        )

    def execute_stream(self,query: str):
        input_dict = {
            "messages":[
                {"role":"user","content":query},
            ]
        }
        # 第三个参数context就是上下文runtime中的信息，就是我们做提示词切换的标记
        for chunk in self.agent.stream(input_dict, stream_mode="values", context={"report": False}):
            latest_message = chunk["messages"][-1]
            if latest_message.content:
                yield latest_message.content.strip() + "\n"

if __name__ == '__main__':
    agent = ReactAgent()
    for chunk in agent.execute_stream("给我生成我的使用报告"):
        print(chunk, end="", flush=True)

9.13 [Agent项目]用户界面开发

import sys
import os
import time

# 获取当前文件的目录 (rag目录)
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取项目根目录 (P6_AI大模型RAG与智能体开发_Agent项目目录)
project_root = os.path.dirname(current_dir)
# 将项目根目录添加到 Python 路径
sys.path.append(project_root)


import streamlit as st
from agent.react_agent import ReactAgent

# 标题
st.title("智扫通机器人智能客服")
st.divider()

if "agent" not in st.session_state:
    st.session_state["agent"] = ReactAgent()

if "message" not in st.session_state:
    st.session_state["message"] = []

for message in st.session_state["message"]:
    st.chat_message(message["role"]).write(message["content"])

# 用户输入提示词
prompt = st.chat_input()

if prompt:
    st.chat_message("user").write(prompt)
    st.session_state["message"].append({"role":"user","content":prompt})

    response_messages = []
    with st.spinner("智能客服思考中..."):
        res_stream = st.session_state["agent"].execute_stream(prompt)

        def capture(generator, cache_list):
            for chunk in generator:
                cache_list.append(chunk)
                for char in chunk:
                    time.sleep(0.01)
                    yield char # 一个字符一个字符的输出 不是一段一段的半流式输出

        st.chat_message("assistant").write_stream(capture(res_stream,response_messages))
        st.session_state["message"].append({"role":"assistant","content":response_messages[-1]})
        st.rerun() # 页面刷新