导学

课程内容

前置准备

01.通义千问大模型的接入

02.代码调用云端的大模型

pip install opneai

openai官方提供的python访问云端的工具库

03.使用环境变量保护APIKEY

点控制面板 -> 环境变量 -> 用户变量 -> 新建 ->变量名OPENAI_API_KEY,变量值就是api

->变量名DASHSCOPE_API_KEY,变量值就是api

配置好之后重启电脑，然后再打开vscode去测试

04.Ollama简介

免费额度没了之后还想继续写代码就用ollama

后续ollama运行起来后通过Restful API接口的形式调用，让我们的代码可以调用ollama管理的模型还是非常方便的

05.win&mac部署ollama并运行蒸馏模型

06.代码调用ollama的本地模型

这种方法运行的前提是ollama已经启动！

再提醒一次，ollama本地模型使用是防止阿里云免费额度用光！

OpenAI库基础使用

01.OpenAI库的基础使用

from openai import OpenAI
#1.获取client对象，OpenAI类对象
client = OpenAI(
    #API_KEY:已经在系统变量中所以跳过
    base_url = "https://ws-g52ciwdl25ej8cmi.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)
#2.调用模型
response = client.chat.completions.create(
    model = "qwen3.7-max",
    messages = [
        {"role": "system", "content": "你是一个python编程专家，并且不说废话总是简单回答"}, #规则设定
        {"role": "assistant", "content": "好的，我是编程专家，并且话不多，你要问什么？"},  #手动写进去的历史回答
        {"role": "user", "content": "输出1-10的数字，使用python代码"},  #最新的用户问题
    ]
)
#3.处理结果
print(response.choices[0].message.content) 

02.OpenAI库的流式输出模式

from openai import OpenAI
#1.获取client对象，OpenAI类对象
client = OpenAI(
    #API_KEY:已经在系统变量中所以跳过
    base_url = "https://ws-g52ciwdl25ej8cmi.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)
#2.调用模型
response = client.chat.completions.create(
    model = "qwen3.7-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 #立刻刷新缓冲区
          )

实际使用中注意不要让他话特别多

03.OpenAi库附带历史消息调用模型

from openai import OpenAI
#1.获取client对象，OpenAI类对象
client = OpenAI(
    #API_KEY:已经在系统变量中所以跳过
    base_url = "https://ws-g52ciwdl25ej8cmi.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)
#2.调用模型
response = client.chat.completions.create(
    model = "qwen3.7-max",
    messages = [
        {"role": "system", "content": "你是一个AI助理，你的回答很简洁"}, #规则设定
        {"role": "user", "content": "小明有两条宠物狗"},  #手动写进去的历史回答
        {"role": "assistant", "content": "ok"},  #最新的用户问题
        {"role": "user", "content": "小红有两只宠物猫"},  #手动写进去的历史回答
        {"role": "assistant", "content": "ok"},  #最新的用户问题
        {"role": "user", "content": "总共有几个宠物呢？"},  #手动写进去的历史回答
    ],
    stream = True #开启了流式输出的功能
)
#3.处理结果user
# print(response.choices[0].message.content) 
# for chunk in response:
#     print(chunk.choices[0].delta.content,
#           end=" ",#每一段之间以空格分隔
#           flush = True #立刻刷新缓冲区
#     )
for chunk in response:
    if not chunk.choices:
        continue

    content = chunk.choices[0].delta.content

    if content is None:
        continue

    print(content, end="", flush=True)

提示词工程

01.大模型prompt工程指南

为了降低模型的幻觉可以使用参考文本做答

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

Zero-shot思想

Few-shot思想

03.提示词优化案例：金融文本分类任务

from openai import OpenAI

# 1. 获取client对象，OpenAI类对象
client = OpenAI(
    base_url = "https://ws-g52ciwdl25ej8cmi.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
    #base_url="http://localhost:11434/v1"
)



examples_data = {       # 示例数据
    '新闻报道': '今日，股市经历了一轮震荡，受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整，以适应市场的不确定性。',
    '财务报告': '本公司年度财务报告显示，去年公司实现了稳步增长的盈利，同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。',
    '公司公告': '本公司高兴地宣布成功完成最新一轮并购交易，收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域，提高市场竞争力',
    '分析师报告': '最新的行业分析报告指出，科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素，投资者应关注这些趋势'
}
# 分类列表
examples_types = ['新闻报道', '财务报道', '公司公告', '分析师报告']

# 提问数据
questions = [
    "今日，央行发布公告宣布降低利率，以刺激经济增长。这一降息举措将影响贷款利率，并在未来几个季度内对金融市场产生影响。",
    "ABC公司今日发布公告称，已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉，此次收购将进一步巩固ABC公司在行业中的地位，并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏",
    "公司资产负债表显示，公司偿债能力强劲，现金流充足，为未来投资和扩张提供了坚实的财务基础。",
    "最新的分析报告指出，可再生能源行业预计将在未来几年经历持续增长，投资者应该关注这一领域的投资机会",
    "小明喜欢小新哟"
]

"""
[
    {"role": "system",      "content": "你是金融专家，将文本分类为['新闻报道', '财务报道', '公司公告', '分析师报告']，不清楚的分类为'不清楚类别' 下面有示例："},
     
    {"role": "user",        "content": "今日，央行发布公告宣布降............."},
    {"role": "assistant",   "content": "新闻报道"},
    {"role": "user",        "content": "ABC公司今日发布公告称，已成功完成对XYZ公司股................."},
    {"role": "assistant",   "content": "财务报告},
    {"role": "user",        "content": "公司资产负债表显示，公司偿债能力强劲，现金流充足..................."},
    {"role": "assistant",   "content": "公司公告"},
    {"role": "user",        "content": "最新的分析报告指出，可再生能源............."},
    {"role": "assistant",   "content": "分析师报告"},
    
    {"role": "user",        "content": "要提问的问题"}
]
"""

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",
        model="qwen3.7-max",
        messages=messages + [{"role": "user", "content": f"按照示例，回答这段文本的分类类别：{q}"}]
    )

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

04.Json数据格式

json数组就是列表套字典

import json

d = {
    "name": "周杰轮",
    "age": 11,
    "gender": "男"
}

s = json.dumps(d, ensure_ascii=False) #json对象
print(s)
#这种方法出来是双引号，如果直接print(str(d))的话是单引号，标准json只认双引号，这样做可以确保跨语言的兼容性

l = [
    {
        "name": "周杰轮",
        "age": 11,
        "gender": "男"
    },
    {
        "name": "蔡依临",
        "age": 12,
        "gender": "女"
    },
    {
        "name": "小明",
        "age": 16,
        "gender": "男"
    }
]

print(json.dumps(l, ensure_ascii=False)) #json数组

json_str = '{"name": "周杰轮", "age": 11, "gender": "男"}'
json_array_str = '[{"name": "周杰轮", "age": 11, "gender": "男"}, {"name": "蔡依临", "age": 12, "gender": "女"}, {"name": "小明", "age": 16, "gender": "男"}]'


res_dict = json.loads(json_str)
print(res_dict, type(res_dict))

res_list = json.loads(json_array_str)
print(res_list, type(res_list))

05.提示词优化案例：金融文本信息抽取

from openai import OpenAI
import json

client = OpenAI(
    base_url = "https://ws-g52ciwdl25ej8cmi.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
    #base_url="http://localhost:11434/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字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："},
    
    {"role": "user",        "content": "2023-01-10，股市震荡。股票强大科技A股今日开盘价100人民币，一度飙升至105人民币，随后回落至98人民币，最终以102人民币收盘，成交量达到520000。"},
    {"role": "assistant",   "content": '{"日期":"2023-01-10","股票名称":"强大科技A股","开盘价":"100人民币","收盘价":"102人民币","成交量":"520000"}'},
    {"role": "user",        "content": "2024-05-16，股市利好。股票英伟达美股今日开盘价105美元，一度飙升至109美元，随后回落至100美元，最终以116美元收盘，成交量达到3560000。"},
    {"role": "assistant",   "content": '{"日期":"2024-05-16","股票名称":"英伟达美股","开盘价":"105美元","收盘价":"116美元","成交量":"3560000"}'},
    
    {"role": "user",        "content": f"按照上述示例，现在抽取这个句子的信息：{要抽取的句子文本}"}]}
]
"""


messages = [
    {"role": "system",      "content": f"你帮我完成信息抽取，我给你句子，你抽取{schema}信息，按JSON字符串输出，如果某些信息不存在，用'原文未提及'表示，请参考如下示例："}
]  #list

for example in examples_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.7-max",
        messages=messages + [{"role": "user", "content": f"按照上述的示例，现在抽取这个句子的信息：{q}"}]
    )

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

06.提示词优化案例：金融文本匹配

RAG开发

01.LangChain的简介

02.LangChain的环境部署

03.RAG介绍

可用rag解决这四个问题

04.【扩展】向量的基础概念

维度越多匹配越精准，1536维比较好

05.【扩展】余弦相似度算法

 谁更靠近1，就代表谁和谁是更加相似的

余弦相似度完全撇除了长度的影响只关注方向

import numpy as np
"""
计算两个向量的余弦相似度（衡量方向相似性，剔除长度影响）

参数：
    vec_a (np.array): 向量A
    vec_b (np.array): 向量B
返回：
    float: 余弦相似度结果（范围[-1,1]，越接近1方向越一致）
公式：
    cos_sim = (vec_a · vec_b) / (||vec_a|| × ||vec_b||)
    拆解：
    1. 点积：vec_a · vec_b = vec_a[0]×vec_b[0] + vec_a[1]×vec_b[1] + ... + vec_a[n]×vec_b[n]
    2. 模长：||vec_a|| = √(vec_a[0]² + vec_a[1]² + ... + vec_a[n]²)
    3. 模长：||vec_b|| = √(vec_b[0]² + vec_b[1]² + ... + vec_b[n]²)

A: [0.5, 0.5]
B: [0.7, 0.7]
C: [0.7, 0.5]
D: [-0.6, -0.5]
"""


def get_dot(vec_a, vec_b): #计算点积
    """计算2个向量的点积，2个向量同维度数字乘积之和"""
    if len(vec_a) != len(vec_b):
        raise ValueError("2个向量必须维度数量相同")

    dot_sum = 0
    for a, b in zip(vec_a, vec_b):
        dot_sum += a * b

    return dot_sum


def get_norm(vec):
    """计算单个向量的模长：对向量的每个数字求平方在求和在开根号"""
    sum_square = 0
    for v in vec:
        sum_square += v * v

    # numpy sqrt函数完成开根号
    return np.sqrt(sum_square)


def cosine_similarity(vec_a, vec_b):
    """余弦相似度：2个向量的点积 除以 2个向量模长的乘积"""

    result = get_dot(vec_a, vec_b) / (get_norm(vec_a) * get_norm(vec_b))
    return result


if __name__ == '__main__':
    vec_a = [0.5, 0.5]
    vec_b = [0.7, 0.7]
    vec_c = [0.7, 0.5]
    vec_d = [-0.6, -0.5]

    print("ab:", cosine_similarity(vec_a, vec_b))
    print("ac:", cosine_similarity(vec_a, vec_c))
    print("ad:", cosine_similarity(vec_a, vec_d))

06.LangChain调用大语言模型

# langchain_community
from langchain_community.llms.tongyi import Tongyi

# 不用qwen3-max，因为qwen3-max是聊天模型，qwen-max是大语言模型
model = Tongyi(model="qwen-max")

# 调用invoke向模型提问
res = model.invoke(input="你是谁呀能做什么？")

print(res)

# langchain_ollama
from langchain_ollama import OllamaLLM

model = OllamaLLM(model="qwen3.6")

res = model.invoke(input="你是谁呀能做什么？")

print(res)

07.LangChain模型的流式输出

# from langchain_community.llms.tongyi import Tongyi
#
# model = Tongyi(model="qwen-max")
#
# # 通过stream方法获得流式输出
# res = model.stream(input="你是谁呀能做什么？")
#
# for chunk in res:
#     print(chunk, end="", flush=True)

from langchain_ollama import OllamaLLM

model = OllamaLLM(model="qwen3.6")

res = model.stream(input="你是谁呀能做什么？")

for chunk in res:
    print(chunk, end="", flush=True) #flush=True表示立即刷新缓冲区，而不是等待缓冲区满后再刷新

08.LangChain调用聊天模型

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 得到模型对象, qwen3-max就是聊天模型
model = ChatTongyi(model="qwen3.7-max")

# 准备消息列表
messages = [
    SystemMessage(content="你是一个边塞诗人。"),
    HumanMessage(content="写一首唐诗"),
    AIMessage(content="锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),
    HumanMessage(content="按照你上一个回复的格式，在写一首唐诗。")
]

# 调用stream流式执行
res = model.stream(input=messages)

# for循环迭代打印输出，通过.content来获取到内容
for chunk in res:
    print(chunk.content, end="", flush=True)

09.LangChain消息的简写形式

from langchain_community.chat_models.tongyi import ChatTongyi

# 得到模型对象, qwen3-max就是聊天模型
model = ChatTongyi(model="qwen3-max")

# 准备消息列表
messages = [
    # (角色，内容)  角色：system/human/ai
    ("system", "你是一个边塞诗人。"),
    ("human", "写一首唐诗。"),
    ("ai", "锄禾日当午，汗滴禾下土，谁知盘中餐，粒粒皆辛苦。"),
    ("human", "按照你上一个回复的格式，在写一首唐诗。")
]

# 调用stream流式执行
res = model.stream(input=messages)

# for循环迭代打印输出，通过.content来获取到内容
for chunk in res:
    print(chunk.content, end="", flush=True)

10.LangChain调用嵌入模型

from langchain_community.embeddings import DashScopeEmbeddings

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

# 不用invoke stream
# embed_query、embed_documents
print(model.embed_query("我喜欢你")) #单次转换，返回一个向量
print(model.embed_documents(["我喜欢你", "我稀饭你", "晚上吃啥"])) #多次转换，返回一个向量列表

from langchain_ollama import OllamaEmbeddings


model = OllamaEmbeddings(model="qwen3-embedding:4b")


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

11.LangChain调用提示词模板

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi
# zero-shot
prompt_template = PromptTemplate.from_template(
    "我的邻居姓{lastname}, 刚生了{gender}, 你帮我起个名字，简单回答。"
)
model = Tongyi(model="qwen-max")
# 调用.format方法注入信息即可
# prompt_text = prompt_template.format(lastname="张", gender="女儿")
#
# model = Tongyi(model="qwen-max")
# res = model.invoke(input=prompt_text)
# print(res)

chain = prompt_template | model  #构建执行链条
#"|"在LangChain里面它表示：把前一个组件的输出，交给后一个组件,也就是：prompt_template | model 意思是：先用 prompt_template 生成提示词,再把提示词交给 model

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

12.FewChain通用提示词模板

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_community.llms.tongyi import Tongyi

# 示例的模板
example_template = PromptTemplate.from_template("单词：{word}, 反义词：{antonym}")

# 示例的动态数据注入 要求是list内部套字典
examples_data = [
    {"word": "大", "antonym": "小"},
    {"word": "上", "antonym": "下"},
]

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() #to_string() 主要作用是将已经格式化好的提示词对象，转换为纯文本字符串（String）。
print(prompt_text)

model = Tongyi(model="qwen-max") #大语言模型

print(model.invoke(input=prompt_text))  #invoke 的意思是 “调用” 或 “执行”

13.模板类的format和invoke方法

invoke需要to_string才能得到字符串

from langchain_core.prompts import PromptTemplate
from langchain_core.prompts import FewShotPromptTemplate
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.llms.tongyi import Tongyi #导入通义千问的纯文本大语言模型（LLM）接口。
from langchain_community.chat_models.tongyi import ChatTongyi #导入通义千问的聊天模型（Chat Model）接口。

"""
PromptTemplate -> StringPromptTemplate -> BasePromptTemplate -> RunnableSerializable -> Runnable
FewShotPromptTemplate -> StringPromptTemplate -> BasePromptTemplate  -> RunnableSerializable -> Runnable
ChatPromptTemplate -> BaseChatPromptTemplate -> BasePromptTemplate  -> RunnableSerializable -> Runnable
Tongyi -> BaseLLM -> BaseLanguageModel -> RunnableSerializable -> Runnable
ChatTongyi -> BaseChatModel -> BaseLanguageModel -> RunnableSerializable -> Runnable
"""


template = PromptTemplate.from_template("我的邻居是：{lastname}，最喜欢：{hobby}")

res = template.format(lastname="张大明", hobby="钓鱼")
print(res, type(res))


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

构建链上Chain之后肯定首选invoke

14.ChatPromptTemplate的使用

15.Chain的使用

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")

# 组成链，要求每一个组件都是Runnable接口的子类
chain = chat_prompt_template | model

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

# 通过stream流式输出
for chunk in chain.stream({"history": history_data}):
    print(chunk.content, end="", flush=True)

16.[扩展]或运算符的重写

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')
    e = Test('e')
    f = Test('f')
    g = Test('g')

    d = a | b | c | e | f | g  # a.__or__(b)
    d.run()
    print(type(d))

17.简单理解Runnable接口

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi


prompt = PromptTemplate.from_template("你是一个AI助手")
model = Tongyi(model="qwen3-max")

chain = prompt | model | prompt | model
chain.invoke()
chain.stream()
print(type(chain))

18.StrOutputParser（字符串输出解析器）和多模型执行链

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: str = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)
print(type(res))

19.JsonOutputParser和多模型执行链

提示词，模板接受的输入是字典类型

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
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}，请帮我解析含义。"
)

# 构建链   （AIMessage("{name: 张若曦}")
chain = first_prompt | model | json_parser | second_prompt | model | str_parser

for chunk in chain.stream({"lastname": "张", "gender": "女儿"}):
    print(chunk, end="", flush=True)

20.自定义函数加入链

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

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

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅生成一个名字，并告知我名字，不要额外信息。"
)

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

# 函数的入参：AIMessage -> dict  ({"name": "xxx"})
# my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

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)

21.Memory临时会话记忆

一个提供功能，一个完成存储

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)

22.Memory长期会话记忆

23.CSVLoader

24.JSONLoader

25.TextLoader和文档分割器

26.PyPDFLoader

27.VectorStores向量存储

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

29.RunnablePassthrough的使用

RAG项目

01.案例介绍

离线（索引）

app_file_upload.py:用来上传文件

knowledge_base.py:知识库更新服务

data中是私有知识

chroma_db：向量数据库

chat_history：长期的会话记忆

02.文本上传WEB服务

补充

streamLit入门：

Streamlit 是一个专为数据科学和机器学习领域打造的开源 Python 框架。它的核心作用是让开发者能够仅使用纯 Python 代码，快速构建并分享精美、交互式的 Web 应用程序，而完全无需具备 HTML、CSS 或 JavaScript 等前端开发知识。

简单来说，Streamlit 能将你原本只能生成静态图表或控制台输出的 Python 数据脚本，在几分钟内转变为一个带有滑块、按钮、下拉框等交互控件的网站，方便你与他人分享和探索数据。

可以参照streamlit官方文档来操作

标题

默认占用端口：localhost:8501

streamlit基础用法

import streamlit as st

#大标题
st.title("StramLit 入门演示")
st.header("StreamLit 一级标题")
st.subheader("StreamLit 二级标题")

#输出段落文字
st.write("我是真无语了1")
st.write("我是真无语了2")
st.write("我是真无语了3")
st.write("我是真无语了4")

#图片
st.image(r"E:\yanguangguang\work\aimodel_study\heima_rag_agent\streamlit\【哲风壁纸】中野一花-中野三玖.png")

# #音频
# st.audio(r"连接")

# #视频
# st.vedio(r"连接")

# #logo 
# st.logo(r" ") #出现在左上角

#表格
student_data = {
    "姓名":["A","B","C","D","E"], #表头：数据
    "学号":[18,19,20,21,22], #表头：数据
    "语文":["1","2","3","4","5"], #表头：数据
    "数学":["1","2","3","4","5"], #表头：数据
    "英语":["1","2","3","4","5"], #表头：数据
    "总分":["1","2","3","4","5"], #表头：数据
} #字典

st.table(student_data)

#普通输入框
name = st.text_input("请输入你的名字：")
st.write(f"你输入的名字是：{name}")

#密码输入框
password = st.text_input("请输入你的密码：",type = "password")
st.write(f"你输入的密码是：{password}")

#单选按钮
gender = st.radio("请输入您的性别：",["男","女","未知"],index = 2)
st.write(f"你选择的性别是：{gender}")

streamlit页面设置

import streamlit as st

#设置页面的配置项
st.set_page_config( 
    page_title="Streamlit 入门演示", #网页标题
    page_icon="😊",
    #控制
    layout="wide",
    #控制侧边栏的状态
    initial_sidebar_state="expanded",
    #菜单信息
    menu_items={
        'Get Help': 'https://www.baidu.com/',
        "Report a bug": "https://www.baidu.com/",
        "About": "Streamlit is an open-source app framework built for building data apps." #页面的描述信息
    }
) #设置页面的配置项


#大标题
st.title("StramLit 入门演示")
st.header("StreamLit 一级标题")
st.subheader("StreamLit 二级标题")

#输出段落文字
st.write("我是真无语了1")
st.write("我是真无语了2")
st.write("我是真无语了3")
st.write("我是真无语了4")

#图片
st.image(r"E:\yanguangguang\work\aimodel_study\heima_rag_agent\streamlit\【哲风壁纸】中野一花-中野三玖.png")

# #音频
# st.audio(r"连接")

# #视频
# st.vedio(r"连接")

# #logo 
# st.logo(r" ") #出现在左上角

#表格
student_data = {
    "姓名":["A","B","C","D","E"], #表头：数据
    "学号":[18,19,20,21,22], #表头：数据
    "语文":["1","2","3","4","5"], #表头：数据
    "数学":["1","2","3","4","5"], #表头：数据
    "英语":["1","2","3","4","5"], #表头：数据
    "总分":["1","2","3","4","5"], #表头：数据
} #字典

st.table(student_data)

#普通输入框
name = st.text_input("请输入你的名字：")
st.write(f"你输入的名字是：{name}")

#密码输入框
password = st.text_input("请输入你的密码：",type = "password")
st.write(f"你输入的密码是：{password}")

#单选按钮
gender = st.radio("请输入您的性别：",["男","女","未知"],index = 2)
st.write(f"你选择的性别是：{gender}")

补充结束

本节实现离线流程，主要就是更新本地的知识库

app_file_upload.py:用来上传文件

knowledge_base.py:知识库更新服务

md5去重

"""
基于Streamlit完成WEB网页上传服务

pip install streamlit

Streamlit：当WEB页面元素发生变化，则代码重新执行一遍
"""
import time

import streamlit as st
from knowledge_base import KnowledgeBaseService

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

# file_uploader
uploader_file = st.file_uploader(
    "请上传TXT文件",
    type=['txt'],
    accept_multiple_files=False,    # False表示仅接受一个文件的上传
)

# session_state就是一个字典
if "service" not in st.session_state:
    st.session_state["service"] = KnowledgeBaseService()


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")

    with st.spinner("载入知识库中。。。"):       # 在spinner内的代码执行过程中，会有一个转圈动画
        time.sleep(1)
        result = st.session_state["service"].upload_by_str(text, file_name)
        st.write(result)

03.md5工具函数开发

知识库相关内容：

md5算法牛逼在不管内容是什么计算出来的都是32位

04.知识库更新服务

05.完成离线流程开发

重跑一次会导致状态的丢失

"""
基于Streamlit完成WEB网页上传服务

pip install streamlit

Streamlit：当WEB页面元素发生变化，则代码重新执行一遍
"""
import time

import streamlit as st
from knowledge_base import KnowledgeBaseService  #KnowledgeBaseService 通常是一个知识库服务类。在实际开发中，它往往被用来封装与知识库相关的业务逻辑

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

# file_uploader
uploader_file = st.file_uploader(
    "请上传TXT文件",
    type=['txt'],
    accept_multiple_files=False,    # False表示仅接受一个文件的上传
)

# session_state就是一个字典
if "service" not in st.session_state:
    st.session_state["service"] = KnowledgeBaseService()


if uploader_file is not None:
    # 提取文件的信息
    file_name = uploader_file.name
    file_type = uploader_file.type
    file_size = uploader_file.size / 1024    #uploader_file.size得到的是文件大小，单位是字节 Byte,除以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")  #uploader_file.getvalue()会获取文件的原始内容。


#st.spinner() 的作用是：告诉用户程序正在处理，不是卡住了。
    with st.spinner("载入知识库中。。。"):       # 在spinner内的代码执行过程中，会有一个转圈动画
        time.sleep(1)
        result = st.session_state["service"].upload_by_str(text, file_name)
        st.write(result)

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

“检索器”（Retriever）就是一个“高效找资料的工具”。它的任务不是回答问题，而是根据你给出的线索，从海量数据中把最相关的信息“捞”出来。

07.rag服务核心代码开发

08.历史会话记录功能的实现

09.聊天界面开发