RAG02-项目工具

2026-07-22 20:48 python RAG项目 17754 字 · 61 分钟

项目工具

1 Ollama大模型高效管理工具

1.1 学习目标 ¶

了解项目开发背景
掌握融合本地知识的RAG系统实现过程

1.2 Ollama概述 ¶

1 Ollama简介 ¶

Ollama 是一个开源工具，让用户只需一条命令即可在本地运行大语言模型（如 Llama、Qwen 等），无需复杂的部署配置，极大降低了本地运行大模型的门槛。

2 Ollama的安装 ¶

Ollama主页： https://ollama.com/ ，支持macOS, Linux 和 Windows 系统，如果是macOS、Linux系统，直接安装使用即可；如果是Windows，需要按照以下教程进行安装使用：
点击下载按钮，获取 OllamaSetup.exe 安装程序。

双击安装文件，点击「Install」开始安装。目前的Ollama会默认安装到C盘，路径如下： C:\Users%username%\AppData\Local\Programs\Ollama ，并不会让用户自定义选择安装路径。

也可以强制更改安装路径

E:\downloads\OllamaSetup.exe /DIR="E:\Ollama"
E:\downloads\OllamaSetup.exe /D=E:\Ollama

安装完成后，会在电脑右下角出现Ollama图标，可以不用先启动Ollama，先退出做一些必要路径的配置！右键图标，会出现退出的按钮「Quit Ollama」，注意：一定要退出Ollama，否则下边的环境配置无法生效！

关闭开机自启动（可选）：

Ollama 默认自动启动，可以删除启动文件夹里的 Ollama 快捷方式，阻止它自动启动。

windows系统删除启动文件夹里的 Ollama 快捷方式

按下 Win + R 组合键，输入以下命令：

shell:startup

回车，打开当前用户的启动文件夹；
找到 Ollama.lnk 快捷方式，右键删除；
重启电脑，即可关闭 Ollama 开机自启。

修改模型存储路径（必须）：

Ollama 的默认模型存储路径如下：C:\Users%username%.ollama\models，建议换一个路径。
以下操作二选一即可，Settings设置或配置环境变量

打开ollama -> Settings -> Model location

配置环境变量

打开「系统环境变量」，新建一个系统变量OLLAMA_MODELS ，然后设置ollama模型的存储路径。
变量名：OLLAMA_MODELS
变量值（路径）：D:\Work\ollama\models

运行Ollama ¶

安装完成后，可通过以下任一方式启动 Ollama：

在 Windows「开始」菜单中搜索并打开 Ollama。
或直接运行安装目录中的 ollama.exe。

启动成功后，Ollama 会在后台运行，并在系统托盘（右下角）显示 Ollama 图标。

启动命令行

按 Win + R，输入 cmd，打开命令提示符（Command Prompt）。
3. 运行模型

执行命令 ollama run <模型名称> ，首次执行会从模型库中下载模型。模型库地址： https://ollama.com/search

推荐显存 ≈ 模型参数量 × 1.2，如果没有GPU默认加载CPU；如果有默认加载GPU

等待下载模型完成后，就可以使用了：

1.3 Ollama应用 ¶

1 在 Python 中使用 Ollama API ¶

安装python的ollama工具:

pip install ollama

# 使用ollama.chat()
import ollama

# 调用聊天接口
response = ollama.chat(
    model='qwen3:4b',
    messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}]
)
print(response)
print(response['message']['content'])

# 使用Client对象，可以远程调用 ollama
from ollama import Client

# client = Client(host='http://192.168.1.100:11434')
client = Client(host='http://127.0.0.1:11434')

response = client.chat(
    model='qwen3:4b',
    messages=[
        {
            'role': 'user',
            'content': '为什么天空是蓝色的？',
        },
    ],
)
print(response['message']['content'])

# 流式输出
import ollama

stream = ollama.chat(
    model='qwen3:4b',
    messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}],
    stream=True,
)
for chunk in stream:
    print(chunk['message']['content'], end='', flush=True)

2 LangChain调用 ¶

pip install langchain langchain_ollama

# from langchain_community.llms import Ollama
from langchain_ollama import OllamaLLM
host = "127.0.0.1"
port = "11434"  # 默认的端口号为11434
# 如果自己本地系统有ollama服务，可以省略base_url
llm = OllamaLLM(base_url=f"http://{host}:{port}", model="qwen3:4b", temperature=0)
res = llm.invoke("你是谁")
print(res)

3 requests调用 ¶

pip install requests

import requests

host = "127.0.0.1"
port = "11434"

url = f"http://{host}:{port}/api/chat"
model = "qwen3:4b"
headers = {"Content-Type": "application/json"}

data = {
    "model": model,  # 模型选择
    "options": {
        "temperature": 0,  # 0 表示结果相对稳定
    },
    "stream": False,  # 是否流式输出
    "messages": [
        {
            "role": "user",
            "content": "你是谁？",
        }
    ],  # 对话列表
}

response = requests.post(url, json=data, headers=headers, timeout=60)
res = response.json()
print(res)

1.4 本节小结 ¶

调用Ollama本地大模型的常用方法。

1. ollama.chat
1. Client
1. langchain_ollama
1. requests

2 LangChain基础知识

2.1 学习目标 ¶

熟悉什么是LangChain及其作用
熟悉LangChain中基本组件及使用

2.2 简述 ¶

1 什么是LangChain ¶

LangChain 是一个用于构建代理和大语言模型LLMs应用的框架。为各种LLMs实现通用的接口，帮助串联互相操作的组件和第三方集成，简化LLMs应用开发。

2022.10, Harrison Chase 在GITHUB上开源了 LangChain，只有800行代码。

2022.11，ChatGPT发布，LangChain作为构建LLM应用的首选框架，关注度迅速上升。

2025.07，新一轮融资后，LangChain估值达到10亿美元。

LangChain官网：https://python.langchain.com/docs/introduction/

LangChain官网中文版：https://www.langchain.com.cn/
LangChain 简化了 LLM 应用程序生命周期的每个阶段：

开发：使用 LangChain 的开源组件和第三方集成构建应用程序。使用 LangGraph 构建具有一流流式和人机交互支持的状态智能体。
生产化：使用 LangSmith 检查、监控和评估应用程序，方便持续优化和部署。
部署：使用 LangGraph平台将应用程序转变为可用于生产的 API 和助手。

2 主要组件 ¶

Models：模型，各种类型的模型和模型集成，比如GPT-4
Prompts：提示，包括提示管理、提示优化和提示序列化
Chains：链，一系列对各种组件的调用
Memory：记忆，用来保存和模型交互时的上下文状态
Indexes：索引，用来结构化文档，以便和模型交互
Agents：代理，决定模型采取哪些行动，执行并且观察流程，直到完成为止

3 LangChain核心包 ¶

langchain-core ：聊天模型和其他组件的基础抽象。
集成包（例如 langchain-openai、langchain-anthropic 等）：重要的集成被拆分为轻量级的独立包，由 LangChain 团队和集成方共同维护。
langchain ：包含链（chains）、智能体（agents）和检索策略，这些构成了应用的认知架构。
langchain-community ：由社区维护的第三方集成。
langgraph ：一个编排框架，用于将 LangChain 组件组合成可用于生产的应用，支持持久化、流式处理及其他关键特性。

4 环境准备 ¶

本课程以LangChain+Qwen进行学习，需要提前安装

pip install openai
pip install langchain
pip install modelscope

借助阿里云-百炼平台（需要申请API Key）：
https://bailian.console.aliyun.com/#/home

2.3 Models ¶

现在市面上的模型多如牛毛，各种各样的模型不断出现，LangChain模型组件提供了与各种模型的集成，并为所有模型提供一个精简的统一接口。
LangChain目前支持三种类型的模型：LLMs、Chat Models(聊天模型)、Embeddings Models(嵌入模型）。

LLMs：大语言模型接收文本字符作为输入，返回的也是文本字符。
聊天模型：基于LLMs, 不同的是它接收聊天消(一种特定格式的数据)作为输入，返回的也是聊天消息。
文本嵌入模型：文本嵌入模型接收文本作为输入, 返回的是浮点数列表。

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

1 LLMs ¶

LLMs（大语言模型）使用场景最多，常用大模型的下载库：
https://huggingface.co/models
https://modelscope.cn/models
下面Qwen为例进行讲解。
模型调用有2种方式：

通过ChatOpenAI进行调用

from langchain_openai import ChatOpenAI
import os
from dotenv import load_dotenv  # 加载环境变量的库, 需要安装dotenv: pip install dotenv
# 1.加载环境变量，获取API密钥
load_dotenv()
# 实例化模型
model = ChatOpenAI(
    base_url=os.environ.get('base_url', 'https://dashscope.aliyuncs.com/compatible-mode/v1'),
    model='qwen3.6-flash',
    openai_api_key=os.environ.get('DASHSCOPE_API_KEY', 'sk-XXXXXX'),
    max_tokens=1000,
    temperature=0,
)

# 获取问答结果
result = model.invoke("帮我讲个笑话吧")
print(result.content)

通过Ollama进行调用 Ollama支持模型

# from langchain_community.llms import Ollama  # Langchain 0.x版本使用
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 获取问答结果
result = model.invoke("请给我讲个笑话吧")
print(f"result->{result}")

2 Chat Models ¶

聊天模型，聊天消息包含下面几种类型，使用时需要按照约定传入合适的值：

AIMessage : 就是 AI 输出的消息，可以是针对问题的回答.
HumanMessage : 人类消息就是用户信息，由人给出的信息发送给LLMs的提示信息，比如“实现一个快速排序方法”.
SystemMessage : 可以用于指定模型具体所处的环境和背景，如角色扮演等。你可以在这里给出具体的指示，比如“作为一个代码专家”，或者“返回json格式”.
ChatMessage : Chat 消息可以接受任意角色的参数，但是在大多数时间，我们应该使用上面的三种类型.

LangChain支持大量的chat 模型,可以通过官网查询：
https://python.langchain.com/docs/integrations/chat/
也可以通过langchain源码查看

SystemMessage+HumanMessage+AIMessage

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
# from langchain_community.chat_models import ChatOllama  # Langchain 0.x版本使用
from langchain_ollama import ChatOllama

# 实例化模型
model = ChatOllama(model="qwen3:4b")

# 定义提示词
messages = [
    SystemMessage(content="现在你是一个著名的诗人"),
    HumanMessage(content="给我写一首唐诗"),
]

# 获取问答结果
result = model.invoke(messages)
# print(f"result->{result}")
print(result.content)

3 Embeddings Models ¶

Embeddings Models(嵌入模型）特点：将字符串作为输入，返回一个浮动数的列表。在NLP中，Embedding的作用就是将数据进行文本向量化。

Embeddings Models可以为文本创建向量映射，这样就能在向量空间里去考虑文本，执行诸如语义搜索之类的操作，比如说寻找相似的文本片段。
https://python.langchain.com/docs/integrations/text_embedding/
不同的Embedding模型对多语言支持和文本类型有不同的特点：

多语言支持： text-embedding-ada-002 ：支持多种语言，但对中文等亚洲语言的支持相对较弱 bge-large-zh ：对中文有很好的支持 multilingual-e5-large ：对多语言都有较好的支持 mxbai-embed-large：
text-embedding-ada-002 ：支持多种语言，但对中文等亚洲语言的支持相对较弱
bge-large-zh ：对中文有很好的支持
multilingual-e5-large ：对多语言都有较好的支持
mxbai-embed-large：
文本类型适用性：代码文本：建议使用专门的代码Embedding模型，如 CodeBERT 通用文本：可以使用 text-embedding-ada-002 或 bge-large-zh 专业领域文本：建议使用该领域的专门模型
代码文本：建议使用专门的代码Embedding模型，如 CodeBERT
通用文本：可以使用 text-embedding-ada-002 或 bge-large-zh
专业领域文本：建议使用该领域的专门模型

可以参考MTEB（大规模文本嵌入基准）排行榜以获取最新模型效果： https://huggingface.co/spaces/mteb/leaderboard
接下来以一个文本嵌入模型的例子进行说明：

# from langchain_community.embeddings import OllamaEmbeddings  # Langchain 0.x版本使用
from langchain_ollama import OllamaEmbeddings

# 初始化Ollama嵌入模型，使用mxbai-embed-large模型，温度设置为0
model = OllamaEmbeddings(model="mxbai-embed-large", temperature=0)

# 对单个查询文本进行嵌入编码
res1 = model.embed_query('这是第一个测试文档')
print(f'result1->{res1}')
print(f'result1的长度->{len(res1)}')

# 对多个文档进行批量嵌入编码
res2 = model.embed_documents(['这是第一个测试文档', '这是第二个测试文档'])
print(res2)

2.4 Prompts ¶

1 通用prompt ¶

Prompt是指当用户输入信息给模型时加入的提示，这个提示的形式可以是zero-shot或者few-shot等方式，目的是让模型理解更为复杂的业务场景以便更好的解决问题。
提示模板：如果你有了一个起作用的提示，你可能想把它作为一个模板用于解决其他问题，LangChain就提供了PromptTemplates组件，它可以帮助你更方便的构建提示。

zero-shot提示方式

from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 定义模板
template = "我的邻居姓{lastname},他生了个儿子，给他儿子起个名字"
prompt = PromptTemplate(
    input_variables=["lastname"],
    template=template,
)
prompt_text = prompt.format(lastname="王")
print(prompt_text)

# 调用模型
result = model.invoke(prompt_text)
print(f"result->{result}")

few-shot提示方式

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 定义模板
examples = [
    {"word": "开心", "antonym": "难过"},
    {"word": "高", "antonym": "矮"},
    {"word": "英俊", "antonym": "丑陋"},
    {"word": "胖", "antonym": "瘦"},
]
example_template = """
词: {word}
反义词: {antonym}
"""
example_prompt = PromptTemplate(
    input_variables=["word", "antonym"],
    template=example_template,
)
few_shot_prompt = FewShotPromptTemplate(
    prefix="给出每个词的反义词", # 前缀
    examples=examples,
    example_prompt=example_prompt,
    suffix="词: {input} 反义词:",   # 后缀
    input_variables=["input"],
    example_separator="\n",
)

prompt_text = few_shot_prompt.format(input="冷")
print(f"prompt_text->{prompt_text}")

# 调用模型
result = model.invoke(prompt_text)
print(f"result->{result}")

2 ChatPrompts ¶

适合交互式对话应用，如聊天机器人、智能客服等，这些应用需要处理用户和LLM之间的多轮对话。
ChatPromptTemplate
SystemMessagePromptTemplate
HumanMessagePromptTemplate
history=[(“system”,”……”),(‘human’,”……”),(“ai”,”……”)]

直接提问

提示模板就是把一些常见的提示整理成模板，用户只需要修改模板中特定的词语，就能快速准确地告诉模型自己的需求。

from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 定义模板
template_str = "帮我讲个关于{name}的笑话"
prompt_template = ChatPromptTemplate.from_template(template_str)
prompt = prompt_template.format_messages(name="气球")
print(f"prompt->{prompt}")

# 调用模型
result = model.invoke(prompt)
print(f"result->{result}")

zero-shot提示方式

from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate
from langchain_core.messages import SystemMessage
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 系统信息
system_prompt = SystemMessage("你是取名专家。")

# 用户信息模版
human_str = "我的邻居姓{lastname}，他生了个儿子，给他儿子起个名字。"
human_template = HumanMessagePromptTemplate.from_template(human_str)

# 组装
chat_template = ChatPromptTemplate.from_messages([system_prompt, human_template])

# 生成最终的提示词
prompt = chat_template.format_messages(lastname="王")
print(f'prompt-->{prompt}')

# 调用模型
result = model.invoke(prompt)  # 返回结果出了content外，还有元数据信息
print(f'result-->{result}')

few-shot提示方式

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 创建 prompt 模版
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "给出每个单词的反义词"),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])

# 创建 few-shot prompt
# history = [
#     HumanMessage(content="开心"),
#     AIMessage(content="难过"),
#     HumanMessage(content="高"),
#     AIMessage(content="矮")
# ]
history = [
    ("human", "开心"),
    ("ai", "难过"),
    ("human", "高"),
    ("ai", "矮"),
]

prompt = prompt_template.format_messages(history=history, question="富有")
print(f"prompt-->{prompt}")

# 调用模型
result = model.invoke(prompt)
print(f'result-->{result}')

2.5 Chains ¶

在LangChain中，Chains链将LLM与其他组件串联成一个应用程序。
针对上一小节的提示模版例子，zero-shot里面，我们可以用链来连接提示模版组件和模型，进而可以更改代码，主要使用LCEL方法。
LCEL（Lang Chain Expression Language）是一种声明式的方法，用于轻松组合链条。
LCEL的基本语法规则是使用 | 符号将不同的组件连接起来，形成一个链式结构。 | 符号类似于Unix的管道操作符，它将一个组件的输出作为下一个组件的输入，从而实现数据的传递和处理。
上一个组件的输出作为下一个组件的输入，输出和输入的类型必须保持一致，否则不能连接。

from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaLLM

# 定义模板
template = "我的邻居姓{lastname}，他生了个儿子，给他儿子起个名字"
prompt = PromptTemplate(input_variables=["lastname"], template=template)

# 实例化模型
llm = OllamaLLM(model="qwen3:4b")

chain = prompt | llm

# 执行链
print(chain.invoke("王"))

如果你想将第一个模型输出的结果，直接作为第二个模型的输入，使用管道符, 代码如下：

from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaLLM

# 创建第一条链
template = "我的邻居姓{lastname}，他生了个儿子，给他儿子起个名字"
first_prompt = PromptTemplate(input_variables=["lastname"], template=template)

# 实例化模型
llm = OllamaLLM(model="qwen3:4b")

first_chain = first_prompt | llm

# 创建第二条链
second_prompt = PromptTemplate(
    input_variables=["child_name"],
    template="邻居的儿子名字叫{child_name}，给他起一个小名",
)
second_chain = second_prompt | llm

# 链接两条链
overall_chain = first_chain | second_chain
print(overall_chain)
print('*' * 80)

# 执行链，只需要传入第一个参数
catchphrase = overall_chain.invoke("王")
print(catchphrase)

2.6 output_parsers ¶

LLM 的输出是自然语言文本，但在应用开发中，经常需要将文本转换为结构化的数据格式，如列表、字典或对象。
LangChain 输出解析器负责获取 LLM 的输出并将其转换为更合适的格式。
部分解析器如下：

解析器名称	核心功能	输出的 Python 类型	工业级应用场景
`StrOutputParser`	默认解析器，将 LLM 的输出直接解析为字符串	`str`	只需要原始回答时，如问答任务、对话场景
`CommaSeparatedListOutputParser`	将 LLM 输出的、用逗号分隔的文本解析为列表	`list[str]`	列表、枚举型输出
`JsonOutputParser`	极其常用，将 LLM 输出的 JSON 字符串解析为 Python 字典	`dict`	结构化 JSON 输出
`PydanticOutputParser`	极其常用，将 LLM 输出解析为预先定义的 Pydantic 对象，提供类型安全和数据验证	`自定义的 pydantic.BaseModel 对象`	输出需要严格结构化（JSON-like）数据时
`DatetimeOutputParser`	从文本中智能地解析出日期和时间信息	`datetime.datetime`	需要时间格式时

1 字符串解析器 ¶

StrOutputParser，最简单的解析器，用于提取模型返回的原始文本：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 创建简单链
prompt = ChatPromptTemplate.from_template("解释{topic}是什么？回答控制20字以内")
chain1 = prompt | model
result1 = chain1.invoke({"topic": "ai"})
print(f"result1-->{result1}")

# 添加字符串解析器
parser = StrOutputParser()

chain2 = prompt | model | parser
result2 = chain2.invoke({"topic": "ai"})
print(f"result2-->{result2}")

result1-->AI 是人工智能，让机器模拟人类智能学习。
result2-->人工智能，是机器模拟人类智能的技术。

2 列表解析器 ¶

CommaSeparatedListOutputParser，将逗号分隔的文本转换为Python列表：

from langchain_core.output_parsers import CommaSeparatedListOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 创建列表解析器
parser = CommaSeparatedListOutputParser()
print(parser.get_format_instructions())

# 创建带格式说明的提示模板
prompt = ChatPromptTemplate.from_template(
    "用中文列出{topic}的五个最重要特点。\n{format_instructions}"
)

# 组合组件
chain = prompt | model | parser

# 调用链
result = chain.invoke({"topic": "大模型", "format_instructions": parser.get_format_instructions()})
print(f"result->{result}")

3 JSON解析器 ¶

JsonOutputParser，将JSON格式文本转换为Python字典或列表：

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")

# 创建JSON解析器
json_parser = JsonOutputParser()
print(json_parser.get_format_instructions())
# 创建带格式说明的提示模板
json_prompt = ChatPromptTemplate.from_template(
    "生成一个包含{person}基本信息的JSON。应包括姓名、职业、年龄和技能列表, 不要包含任何注释或额外说明。\n{format_instructions}"
)

# 组合组件
json_chain = json_prompt | model | json_parser

# 调用链
result = json_chain.invoke({"person": "雷军", "format_instructions": json_parser.get_format_instructions()})
print(f"result->{result}")

4 Pydantic解析器 ¶

PydanticOutputParser，使用Pydantic模型定义输出结构：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")


# 定义Pydantic模型
class Movie(BaseModel):
    title: str = Field(description="电影标题")
    director: str = Field(description="导演姓名")
    year: int = Field(description="上映年份")
    genre: List[str] = Field(description="电影类型")
    rating: float = Field(description="评分（1-10）")


# 创建Pydantic解析器
pydantic_parser = PydanticOutputParser(pydantic_object=Movie)
print(pydantic_parser.get_format_instructions())

# 创建带格式说明的提示模板
pydantic_prompt = ChatPromptTemplate.from_template(
    "生成一部{genre}电影的信息。\n{format_instructions}"
)

# 组合组件
pydantic_chain = pydantic_prompt | model | pydantic_parser

# 调用链
movie_data = pydantic_chain.invoke({"genre": "科幻", "format_instructions": pydantic_parser.get_format_instructions()})
print(movie_data)

5 自定义解析器 ¶

创建自定义输出解析器

from langchain_core.output_parsers import BaseOutputParser
from typing import Dict, Any
import re
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import OllamaLLM

# 实例化模型
model = OllamaLLM(model="qwen3:4b")


class CustomKeyValueParser(BaseOutputParser[Dict[str, Any]]):
    """解析形如'key: value'或'key：value'的文本"""

    def parse(self, text: str) -> Dict[str, Any]:
        """从文本中解析键值对（兼容中文冒号、列表序号和 Markdown 强调）"""
        result = {}
        lines = text.strip().split("\n")
        for raw_line in lines:
            line = raw_line.strip()
            if not line:
                continue

            # 去掉常见列表前缀：-、*、1.、1)
            line = re.sub(r"^[-*•\s]+", "", line)
            line = re.sub(r"^\d+[\.)]\s*", "", line)

            # 去掉 Markdown 加粗标记
            line = line.replace("**", "").strip()

            # 统一中文/英文冒号
            normalized = line.replace("：", ":")
            if ":" in normalized:
                key, value = normalized.split(":", 1)
                key = key.strip()
                value = value.strip()
                if key and value:
                    result[key] = value
        return result

    def get_format_instructions(self) -> str:
        """提供格式指导给模型"""
        return """请以'键: 值'的格式返回信息，每行一个键值对。
例如：
名称: 爱因斯坦
职业: 物理学家
贡献: 相对论"""


# 使用自定义解析器
custom_parser = CustomKeyValueParser()
print(custom_parser.get_format_instructions())
custom_prompt = ChatPromptTemplate.from_template(
    "提供关于{person}的基本信息。\n{format_instructions}"
)

# 组合组件
custom_chain = custom_prompt | model | custom_parser

# 调用模型
result = custom_chain.invoke({
    "person": "牛顿",
    "format_instructions": custom_parser.get_format_instructions(),
})
print(f"result->{result}")

2.7 Memory ¶

大模型本身没有记忆能力，它并不保存上次交互的内容，ChatGPT能够连续对话，因为它将历史消息记录回传给了模型。
LangChain 也提供了Memory组件, 用于保存和管理对话历史。
Memory分为两种类型：短期记忆和长期记忆。

类型	作用	特点	常用 API / 组件
短期记忆（Short-Term Memory）	保存当前对话的历史消息	使用相同的 `thread_id` 可以继续之前的对话	`ChatMessageHistory`、`InMemorySaver`
长期记忆（Long-Term Memory）	将重要信息保存到数据库或文件中	程序重启后仍然可以读取	数据库、向量数据库、文件存储

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.messages import messages_to_dict, messages_from_dict
import json

# 1. 创建一个 ChatMessageHistory 对象，用来存储对话信息
chat_history = ChatMessageHistory()

# 添加用户消息
chat_history.add_user_message("在吗？")

# 添加大模型消息
chat_history.add_ai_message("在")

# 打印所有消息
print(f"chat_history.messages --> {chat_history.messages}")

# 2. 将 chat_history.messages 转成字典，便于保存到数据库或文件
dicts = messages_to_dict(chat_history.messages)
print(f"dicts --> {dicts}")

# 2.2 保存到文件
with open("history.json", "w", encoding="utf-8") as f:
    f.write(json.dumps(dicts, indent=2, ensure_ascii=False))

# 3. 从文件读取并还原成消息对象
with open("history.json", "r", encoding="utf-8") as f:
    messages = json.load(f)

chat_messages = messages_from_dict(messages)
print(f"chat_messages --> {chat_messages}")

chat_history.messages --> [HumanMessage(content='在吗？', additional_kwargs={}, response_metadata={}), AIMessage(content='在', additional_kwargs={}, response_metadata={}, tool_calls=[], invalid_tool_calls=[])]
dicts --> [{'type': 'human', 'data': {'content': '在吗？', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': None}}, {'type': 'ai', 'data': {'content': '在', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': None, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}}]
chat_messages --> [HumanMessage(content='在吗？', additional_kwargs={}, response_metadata={}), AIMessage(content='在', additional_kwargs={}, response_metadata={}, tool_calls=[], invalid_tool_calls=[])]

1 短期记忆（Short-term Memory） ¶

短期记忆的本质是线程级状态管理。在 LangGraph 中：
短期记忆 = Agent 状态（State） + 检查点持久化（Checkpointer） + 线程标识（thread_id）。

（1）状态（State）
通常是一个包含 messages 字段的字典或 Pydantic 模型（如 MessagesState 或自定义 CustomState），用于存储当前对话的所有消息、中间变量、工具调用结果等。

（2）检查点器（Checkpointer）
负责将状态序列化并持久化到内存、SQLite、PostgreSQL 等数据库。每次状态变更（如新增一条消息）都会触发一次检查点保存。

（3）对话ID（thread_id）
作为对话的唯一标识符，确保不同用户或不同对话之间信息的完全隔离。即使多个用户并发交互，也不会发生记忆混淆。

import time
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_ollama import ChatOllama
from langchain.messages import HumanMessage

# 1.初始化大模型
model = ChatOllama(model="qwen3:4b")

# 2.创建内存保存器，用于保存对话历史
checkpointer = InMemorySaver()

# 3.主对话配置
config = {
    "configurable": {
        "thread_id": str(time.time())    # 对话ID，用于追踪对话历史
    }
}

# 4.创建 Agent 实例
agent = create_agent(
    model=model,    # 底层模型, 这里使用 Ollama 聊天模型
    tools=[],       # 工具列表, 表示 Agent 可以使用的工具
    checkpointer=checkpointer,    # 内存保存器, 给 Agent 添加记忆能力
    system_prompt="你是一个通用任务助手。"  # 系统提示词，设定 Agent 的角色和任务
)

# 5.对话测试
# 第一轮对话
input_1 = {
    "messages": [HumanMessage(content="你好，我叫小易。")]
}
response_a = agent.invoke(input_1, config=config)

# 第二轮对话
input_2 = {
    "messages": [HumanMessage(content="你好，我叫什么？")]
}

# 新线程（无历史记忆）
config_2 = {
    "configurable": {
        "thread_id": str(time.time_ns())
    }
}
response_b = agent.invoke(input_2, config=config_2)

# 回到原线程（有历史记忆）
response_c = agent.invoke(input_2, config=config)

def last_content(resp):
    msgs = resp.get("messages", [])
    return msgs[-1].content if msgs else resp

print("Response A (第一次对话):")
print(last_content(response_a))

print("\nResponse B (新线程，无记忆):")
print(last_content(response_b))

print("\nResponse C (回到原线程，有记忆):")
print(last_content(response_c))

代码运行逻辑：

checkpointer (记忆核心) ：这是 LangGraph 区别于传统简单 API 调用的关键。它允许 Agent 在多次 invoke 调用之间保存状态。
thread_id (记忆索引) ：你可以把 thread_id 想象成数据库中的主键。
Response A：向 ID 为 100 的记录里写入了名字。
Response B：向 ID 为 200 的新记录提问，因为 200 是空的，所以 Agent 不知道名字。
Response C：又回到 ID 为 100 的记录提问，Agent 读取了历史记录，所以知道名字。
System Prompt ：虽然输入是在对话，但因为设定了 “你是一个翻译官”，Agent 可能会试图在回答的同时进行翻译，或者在回复格式上符合翻译官的身份。

2 长期记忆 ¶

对于需要长期运行和可靠记忆的应用，推荐使用数据库进行持久化。详见后面阶段LangGraph部分。

2.8 Indexes ¶

Indexes组件作用: 让LangChain具备文档处理能力。这里的文档不局限于txt、pdf等，还涵盖email、区块链、视频等格式。
Indexes组件主要包含：

文档加载器
文本分割器
VectorStores
检索器

总结:

组件	作用	常用 API
文档加载器（Document Loaders）	读取文档内容并转换为 LangChain 文档对象，支持 `txt`、`pdf`、`docx`、`json`、`html` 等格式。	`TextLoader`、`UnstructuredLoader`
文本分割器（Text Splitters）	将长文本切分为较小文本块（chunk），便于向量化与检索。	`RecursiveCharacterTextSplitter`、`CharacterTextSplitter`
向量库（VectorStores）	存储文本向量，并支持相似度检索。	`Chroma`、`FAISS`、`Milvus`、`similarity_search()`
检索器（Retrievers）	从向量库中检索与问题相关的文档。	`as_retriever()`、`BM25Retriever`

1 文档加载器 ¶

https://python.langchain.com/v0.2/docs/introduction/
文档加载器主要基于 Unstructured 包， Unstructured 是一个python包，可以把各种类型的文件转换成文本。文档加载器使用起来很简单，只需要引入相应的loader工具。
LangChain支持的文档加载器 (部分)：

如果出现nltk相关的错误，比如
LookupError Traceback (most recent call last) Cell In[5], line 5 3 # 创建 UnstructuredLoader 对象 4 loader = UnstructuredLoader('./data/衣服属性.txt', encoding='utf8') ----> 5 docs = loader.load()
可以尝试下载对应的资源包：

# import nltk
# nltk.download('averaged_perceptron_tagger_eng')

[nltk_data] Downloading package averaged_perceptron_tagger_eng to
[nltk_data]     C:\Users\angmi\AppData\Roaming\nltk_data...
[nltk_data]   Unzipping taggers\averaged_perceptron_tagger_eng.zip.

from langchain_unstructured import UnstructuredLoader

# 创建 UnstructuredLoader 对象
loader = UnstructuredLoader('./data/衣服属性.txt', encoding='utf8')
docs = loader.load()
print(f'docs-->{docs}')
print(f'len-->{len(docs)}')
print(f'第一行数据-->{docs[0].page_content}')
print('*' * 100)

from langchain_community.document_loaders import TextLoader

# 创建 TextLoader 对象
loader = TextLoader('./data/衣服属性.txt', encoding='utf8')
docs = loader.load()
print(f'docs-->{docs}')
print(f'len-->{len(docs)}')
print('第一行数据-->{}'.format(docs[0].page_content.split('\n')[0]))

2 文本分割器 ¶

由于模型对输入长度有限制，在碰到很长的文本时，需要把文本分割成多个小片段。
文本分割最简单的方式是按照字符长度进行分割，但是这会带来很多问题，比如如果文本是一段代码，一个函数被分割到两段之后就成了没有意义的字符，所以整体原则是把语义相关的文本片段放在一起。
LangChain支持多种文本分割器：

分割器	作用	适用场景
`RecursiveCharacterTextSplitter`	递归分割文本，优先按段落、换行等自然边界切分	最常用，适用于 RAG、PDF、网页、知识库等通用文本处理
`CharacterTextSplitter`	按指定字符直接分割文本	简单文本处理，如 txt、日志文件
`TokenTextSplitter`	按 Token 数量分割文本	控制 LLM 输入长度，避免超过 Token 限制
`MarkdownHeaderTextSplitter`	按 Markdown 标题结构分割	Markdown 文档、技术文档、知识库
CharacterTextSplitter（字符文本分割器） - 最基本的文本分割器	按照指定的分隔符（默认“\n\n”）进行分割，并且考虑文本片段的最大长度。

from langchain_core.documents import Document
from langchain_text_splitters import CharacterTextSplitter

# 创建分词器
# separator: 分隔符
# chunk_size: 每个块的最大长度
# chunk_overlap: 相邻块的重叠长度
text_splitter = CharacterTextSplitter(
    separator=" ",
    chunk_size=5,
    chunk_overlap=1,
)

# 一句话分割
result1 = text_splitter.split_text("a b c d e f")
print(f"result1 ---> {result1}")

# 多句话分割（生成 Document 列表）
result2 = text_splitter.create_documents(["a b c d e f", "e f g h"])
print(f"result2 ---> {result2}")

# Document 列表分割
result3 = text_splitter.split_documents(
    [Document(page_content="a b c d e f", metadata={"id": "1"})]
)
print(f"result3 ---> {result3}")

RecursiveCharacterTextSplitter（递归字符文本分割器）-最常用的文本分割器
递归字符文本分割器是一种更智能的分割方法，它尝试在特定分隔符处分割文本，以保持更好的语义完整性。
特点：

尝试在自然断点处分割文本
比简单的字符分割更能保持语义完整性
适用于结构化程度较高的文本，如 Markdown、HTML 等

运行流程：

首先尝试使用第一个分隔符（如 “\n\n”）分割文本
如果分割后的块仍然过大，则使用下一个分隔符继续分割
重复此过程，直到达到指定的 chunk_size 或用完所有分隔符

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=20,  # 每个块最多 20 个字符
    chunk_overlap=6,  # 相邻分块会共享 6 个字符
    length_function=len,  # 用字符数来衡量长度
    separators=["\n\n", "\n", " ", ""]  # 优先按 \n\n 分段，依次按 \n、空格、逐字符切分
)

text = """
人工智能正在快速发展，尤其是大语言模型的应用，正在改变人类的工作方式。
它们可以帮助人们进行写作、代码生成、甚至是科研探索。
相比之下，新能源的发展同样重要。
电动车和太阳能正在逐渐替代传统能源，减少碳排放，对全球环境保护至关重要。
"""

docs = text_splitter.split_text(text)
print(docs)

['人工智能正在快速发展，尤其是大语言模型', '是大语言模型的应用，正在改变人类的工作方', '人类的工作方式。', '它们可以帮助人们进行写作、代码生成、甚', '代码生成、甚至是科研探索。', '相比之下，新能源的发展同样重要。', '电动车和太阳能正在逐渐替代传统能源，减', '传统能源，减少碳排放，对全球环境保护至关', '环境保护至关重要。']

SemanticChunker（语义文本分割器） - 高级文本分割器
语义文本分割器使用语义理解来分割文本，是一种更高级的分割方法。特点：

基于语义相似性分割文本
能够更好地保持语义完整性
计算成本较高，处理大量文本时可能效率较低
适用于需要高度语义理解的场景

from langchain_experimental.text_splitter import SemanticChunker
from langchain_ollama import OllamaEmbeddings

# 初始化向量模型
embed = OllamaEmbeddings(model="mxbai-embed-large")

# 创建语义分割器
semantic_splitter = SemanticChunker(
    embeddings=embed,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=50.0,           # 数值越小，切分越积极
    sentence_split_regex=r"(?<=[。！？.!?])\s*",  # 中英文句末切分
    min_chunk_size=10,                           # 最小块长度
)

# 使用已有的 text 变量进行切分
text = """
人工智能正在快速发展，尤其是大语言模型的应用，正在改变人类的工作方式。
它们可以帮助人们进行写作、代码生成、甚至是科研探索。
相比之下，新能源的发展同样重要。
电动车和太阳能正在逐渐替代传统能源，减少碳排放，对全球环境保护至关重要。
"""
semantic_docs = semantic_splitter.split_text(text)

# 打印结果
for i, chunk in enumerate(semantic_docs, start=1):
    print(f"------ Chunk {i} ------")
    print(chunk.strip())
    print()

INFO: HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"

------ Chunk 1 ------
人工智能正在快速发展，尤其是大语言模型的应用，正在改变人类的工作方式。 它们可以帮助人们进行写作、代码生成、甚至是科研探索。
------ Chunk 2 ------
相比之下，新能源的发展同样重要。 电动车和太阳能正在逐渐替代传统能源，减少碳排放，对全球环境保护至关重要。
------ Chunk 3 ------

MarkdownHeaderTextSplitter（Markdown文本分割器）
适用于Markdown文档，按照标题进行拆分

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)

markdown_text = """# Header 1

Some text

## Header 2

More text

### Header 3

Even more text
"""

markdown_docs = markdown_splitter.split_text(markdown_text)
print(markdown_docs)

[Document(metadata={'Header 1': 'Header 1'}, page_content='Some text'), Document(metadata={'Header 1': 'Header 1', 'Header 2': 'Header 2'}, page_content='More text'), Document(metadata={'Header 1': 'Header 1', 'Header 2': 'Header 2', 'Header 3': 'Header 3'}, page_content='Even more text')]

3 VectorStores ¶

VectorStores是向量数据库，它的作用是存储嵌入向量，提供相似查询等功能。
LangChain支持的VectorStore有 https://python.langchain.com/docs/integrations/vectorstores/ ，常见的如下：

我们使用 Chroma 组件作为例子：

pip install chromadb
pip install langchain-chroma

from langchain_text_splitters import CharacterTextSplitter
from langchain_chroma import Chroma
from langchain_community.document_loaders import TextLoader
from langchain_ollama import OllamaEmbeddings

# 1.加载文档
loader = TextLoader('./data/pku.txt', encoding='utf-8')
docs = loader.load()
# print(f'docs-->{docs}')

# 2.将文档进行分块
text_splitter = CharacterTextSplitter(separator="\n\n", chunk_size=200, chunk_overlap=30)
split_docs = text_splitter.split_documents(docs)
print(f'split_docs-->{split_docs}')

# 3.将分割后的文档存储到向量数据库中
embedding = OllamaEmbeddings(model="mxbai-embed-large")

# 创建向量数据库，需要指定 存储的文档和向量模型名称以及持久化目录
chromadaDB = Chroma.from_documents(
    documents=split_docs,
    embedding=embedding,
    persist_directory='./chroma_db',
)
# 假如你的向量数据库已经存在，那么可以直接加载
# chromadaDB = Chroma(persist_directory='./chroma_db', embedding_function=embedding)

# 4.使用向量数据库进行查询
query = "1937年北京大学发生了什么？"
result = chromadaDB.similarity_search(query, k=2)
print(f'result-->{result}')

4 检索器 ¶

LangChain的检索器定义 ¶

检索器是 LangChain 中负责信息检索的模块，通常与索引（Indexes）模块（如向量存储、嵌入模型）结合使用。
核心功能是：

输入：接收用户查询（通常是文本）。
处理：根据查询从数据源中检索相关内容。
输出：返回一组相关文档或文本片段（通常是 Document 对象列表）。

检索器在以下场景中扮演关键角色：

问答系统：从文档或知识库中检索答案的上下文。
语义搜索：根据查询的语义返回相关结果。
上下文增强：为语言模型提供外部知识，解决其知识局限。

检索器的工作原理 ¶

检索器通常与向量存储（Vector Stores）配合，通过嵌入模型（Embedding Models）将查询和文档转为向量，基于相似性进行检索。
工作流程：

查询嵌入：将用户查询通过嵌入模型（如 OpenAIEmbeddings）转为向量表示
相似性搜索：在向量存储中查找与查询向量最相似的文档向量。
文档返回：返回匹配的文档（包含内容、元数据等）。
后处理（可选）：对检索结果进行排序、过滤或重新排名。

检索器的核心依赖：

嵌入模型：将文本转为向量（如 OpenAIEmbeddings, HuggingFaceEmbeddings）。
向量存储：存储文档向量（如 Chroma、FAISS、Pinecone）。
相似性度量：如余弦相似度、欧式距离

检索器类型 ¶

langchain支持很多检索器 https://python.langchain.com/docs/integrations/retrievers/ ，部分如下：

此处我们讲解VectorStoreRetriever。
在 LangChain 中，as_retriever() 方法的 search_type 参数决定了向量检索的具体算法和行为。

retriever = vector_store.as_retriever(
    search_type="similarity",  # 可选 "similarity"|"mmr"|"similarity_score_threshold"
    search_kwargs={
        "k": 5,                           # 返回结果数量
        "score_threshold": 0.7,           # 仅当search_type="similarity_score_threshold"时有效,低于阈值的都丢弃
        "filter": {"source": "重要文档.pdf"},  # 元数据过滤，只会检索满足条件的文档
        "lambda_mult": 0.25,              # 仅MMR搜索有效(控制多样性)：接近 0 则更强调和查询的相关性；接近 1 则更强调结果之间的差异性
    },
)

以下是三种搜索类型的对比：

from langchain_chroma import Chroma
from langchain_ollama import OllamaEmbeddings

embedding = OllamaEmbeddings(model="mxbai-embed-large")

# 向量数据库已经存在，那么可以直接加载
chromadaDB = Chroma(persist_directory='./chroma_db', embedding_function=embedding)

# 使用向量数据库进行查询
query = "1937年北京大学发生了什么？"

# 使用 as_retriever 方法返回 Retriever 对象，然后调用 invoke 方法进行查询
retriever = chromadaDB.as_retriever(search_kwargs={"k": 2})
result = retriever.invoke(query)
print(f'result-->{result}')
result = chromadaDB.similarity_search(query, k=2)
print(f'result-->{result}')

拓展：
vectordb.as_retriever() 和 vectordb.similarity_search() 都是用于从向量数据库中检索相关文档的方法，它们有什么异同：

相同
- 核心功能：两者都基于向量相似度（如余弦相似度）从向量数据库中检索与查询最相关的文档。
- 底层技术：通常使用相同的嵌入模型和相似度计算方式（如 FAISS、Chroma、Pinecone 等）。
不同

2.9 Agents ¶

1 Agent 概念 ¶

Agent（智能体）是基于大模型构建的能够感知环境、进行决策和执行动作的智能实体。
它不仅会回答问题，还会在需要时调用工具完成任务。

为什么需要 Agent？

大模型很强，但有明显限制，例如：

不能直接获取最新实时信息
复杂计算、查询、执行操作能力有限
不能直接访问外部系统（数据库、搜索、第三方API）

所以需要 Agent 连接工具，补齐这些能力。

Agent 的核心组成

Agent = 大模型（LLM） + 任务规划（Planning） + 工具调用（Tools/Action） + 记忆（Memory）

LLM：理解问题、推理决策
Planning：把大任务拆成可执行小步骤
Tools：调用搜索、计算器、数据库、API 等
Memory：保存上下文，支持多轮连续对话

把 Agent 当成“会用工具的助手”：
大模型负责“想”，工具负责“做”，记忆负责“记住过程”。

2 Agent 的工作流程

flowchart TD
    A[用户问题query]
    --> B[构建输入<br/>Query+Prompt+Memory]

    --> D[Agent决策<br>LLM理解任务并选择执行策略]

    D --> E1[Tool Calling]
    D --> E2[ReAct]
    D --> E3[Plan and Execute]
    D --> E4[Multi-Agent]

    E1 --> F[生成最终答案]
    E2 --> F
    E3 --> F
    E4 --> F

用户提出问题
Agent 构建输入（用户问题 + 系统提示词 + 历史上下文）
Agent 决策：
- LLM理解任务并选择执行策略（直接回答、Tool Calling、ReAct、Plan and Execute、Multi-Agent）
输出最终答案

经典 ReAct 工作流
ReAct框架，调用LLM 循环执行 Thought-Action-Observe 的流程，直到完成任务。参考 https://docs.langchain.com/oss/javascript/langchain/agents

flowchart TD
    A[用户问题] --> B[Thought 思考]
    B --> C[Action 调用工具]
    C --> D[Observation 观察结果]
    D --> E{任务完成?}

    E -- 否 --> B
    E -- 是 --> F[Final Answer]

用户问题：输入真实需求（如“帮我查广州今天天气并给穿衣建议”）。
Thought：思考，模型先判断要做什么。
Action：调用工具（搜索、数据库、代码执行、API）。
Observation：读取工具返回数据。
循环：循环执行 Thought→Action→Observation，直到任务完成。
最终回答：给出可执行、可解释的结果。

3 langchain实现Agent ¶

LangChain 提供了不同类型的代理:

Zero-shot ReAct Description
- 基于 ReAct 框架（推理 + 行动），仅依赖工具的描述来决定调用哪个工具。
- 特点：无需额外示例（zero-shot），但 不具备记忆能力，每次推理都独立进行。
- 使用场景：简单任务，工具选择完全靠工具描述即可。
Structured Chat Zero-shot ReAct Description
- 同样基于 ReAct 框架，但可以处理 结构化输入，即支持带多个参数的工具（类似函数调用）。
- 特点：不仅能像第一种代理那样根据描述选择工具，还能正确组织并传递复杂参数。
- 使用场景：调用接口类工具、需要多参数输入的任务。
Conversational ReAct Description
- 在 ReAct 框架基础上，增强了 对话记忆能力。
- 特点：能根据上下文对话历史来做出工具选择和回应，更适合持续性对话场景。
- 使用场景：多轮对话，用户可能引用之前的内容或需要长期上下文跟踪。

from langchain_community.agent_toolkits.load_tools import get_all_tool_names
results = get_all_tool_names()
print(results)

接下来，通过一个示例来学习Agent的基本使用。

需要提前安装 wikipedia
pip install wikipedia

from langchain_core.tools import tool  # 工具装饰器
from langchain_core.messages import HumanMessage  # 消息类型
from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver  # 记忆存储
from langchain_openai import ChatOpenAI  # 大模型
import os
from dotenv import load_dotenv

load_dotenv()

# 1.定义工具
@tool
def calculator(expression: str) -> str:
    """执行数学计算

    参数：
        expression: 数学表达式，如 "300 * 0.25"
    返回：
        计算结果的字符串
    """
    try:
        result = eval(expression)
        return f"结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

@tool
def wikipedia_search(query: str) -> str:
    """搜索维基百科

    参数：
        query: 搜索关键词
    返回：
        搜索结果
    """
    try:
        from langchain_community.tools import WikipediaQueryRun
        from langchain_community.utilities import WikipediaAPIWrapper

        wikipedia = WikipediaQueryRun(
            api_wrapper=WikipediaAPIWrapper()
        )
        result = wikipedia.run(query)
        return result if result else "未找到相关信息"
    except Exception as e:
        return f"搜索错误: {str(e)}"

# 2.创建OpenAI模型
model = ChatOpenAI(
    model="qwen3.6-plus",
    temperature=0,
    max_tokens=1000,
    base_url=os.environ.get('base_url', 'https://dashscope.aliyuncs.com/compatible-mode/v1'),
    openai_api_key=os.environ.get('DASHSCOPE_API_KEY'),
)

# 3.创建Agent
# 3.1 工具列表
tools = [calculator, wikipedia_search]
# 3.2 创建记忆存储器
memory = MemorySaver()
# 3.3 创建Agent
agent = create_agent(
    model=model,  # 使用的底层大模型
    tools=tools,  # 工具列表，这里包含calculator,wikipedia_search, Agent可以调用这些工具来获取信息和进行计算
    checkpointer=memory,  # 传入上面的内存保存器，赋予Agent记忆能力
)

# 4.使用Agent进行对话
config = {
    "configurable": {"thread_id": "user_123456"}
}
# 数学计算
print("数学计算：")
for chunk in agent.stream(
        {"messages": [HumanMessage(content="计算一下300的25%是多少？")]},
        config
):
    print(chunk)
    print("----")
print("维基百科搜索：")
for chunk in agent.stream(
        {"messages": [HumanMessage(content="法国首都是哪里？")]},
        config
):
    print(chunk)
    print("----")

参考官网 https://python.langchain.com/docs/how_to/custom_tools/

2.10 LangChain使用场景 ¶

个人助手
基于文档的问答系统
聊天机器人
Tabular数据查询
API交互
信息提取
文档总结

2.11 本节小结 ¶

本章主要介绍了LangChain的基本组件及使用方式。

2.12 本节作业 ¶

1 作业1 ¶

题目1：什么是LangChain？请解释其核心组件和设计思想。
题目2：解释LangChain中的链（Chain）概念，并介绍主要的链类型。
题目3：解释LangChain中的Prompts概念，并介绍主要类型。
题目4：实现LangChain中的output_parsers的具体使用方式

2 作业2 ¶

题目1：什么是Memory组件？请解释LangChain中不同类型的记忆机制。
题目2：解释LangChain中的提示工程及其最佳实践。
题目3：LangChain的Document Loaders？请介绍其功能和主要类型。
题目4 ：什么是LangChain的代理（Agent）？请说明其工作原理和应用。

3 Milvus向量数据库

3.1 学习目标 ¶

理解什么是向量数据库
理解Milvus和Mysql的区别
掌握Milvus数据库的增删改查

3.2 Milvus 是什么？

Milvus 是 2019 年开源的专用向量数据库，主要用于存储、索引和检索向量数据，支持万亿级向量相似度搜索。

1 为什么需要 Milvus？

现实中的大量数据都是非结构化数据（如文本、图片、音频、视频），无法直接进行语义检索，因此需要先转换成向量，再存入向量数据库。

flowchart LR;
    A[非结构化数据] --> B[Embedding模型]
    B --> C[向量Vector]
    C --> D[Milvus]
    D --> E[相似度搜索]
    E --> F[TopK检索结果Context]

使用 Embedding 模型将文本、图片等转换为向量；
将向量存入 Milvus；
检索向量库Milvus，返回最相似的 TopK 数据-检索上下文Context。

2 Milvus 的优势

支持亿级～万亿级向量数据
基于 ANN（近似最近邻）算法，检索速度快
支持稠密向量、稀疏向量和混合检索

3.3 Milvus 与 MySQL 的区别

对比项	MySQL	Milvus
存储数据	结构化数据	向量数据
数据示例	用户、订单、商品	文本向量、图片向量
查询方式	精确查询	相似度搜索
底层索引	B+Tree	ANN（HNSW、IVF 等）
常见操作	`WHERE id=1`	TopK 相似向量搜索
应用场景	电商、ERP、金融系统	RAG、推荐系统、语义搜索、AI Agent
举个栗子	电商场景，查订单，MySQL: `SELECT * FROM orders WHERE order_id=100;`	电商推荐，推荐相似商品，用户输入“运动鞋”, 返回 “跑步鞋”、“篮球鞋”、“足球鞋”等商品列表
数据存储结构	Database-Table-Field/Column-Row	Database-Collection-Field-Entity-Vector

3.4 关键概念

1 Milvus中数据

在 Milvus 中，一条数据（Entity）通常由多个字段（Field）组成。比如：

id	text	vector	source
1	Transformer 是一种深度学习模型……	[0.12,-0.35,…]	Transformer介绍.pdf

其中：

id：唯一编号（主键）
text：原始文本
vector：Embedding 向量（用于相似度搜索）
source：文档来源

2 Milvus 数据结构

Milvus 的数据组织方式与 MySQL 十分类似。

Milvus	MySQL	说明
Database	Database	数据库
Collection	Table	一张表，存储一类数据
Field	Column	字段（列）
Entity	Row	一条数据（行）
Primary Key	Primary Key	主键，唯一标识一条数据

例如：

Database: itcast
│
├── Collection：demo01
│      │
│      ├── id
│      ├── text
│      ├── vector
│      └── source
└── Collection：demo02
       │
       ├── id
       ├── text
       ├── vector
       └── source

3 Schema（数据结构定义）

Schema = 数据结构定义（Definition）

创建 Collection 前，需要先定义 Collection 的结构。

创建 Collection
↓
定义 Schema
↓
插入数据

4 Collection Schema（集合定义）

Collection Schema 用于描述整个 Collection 的结构。

它规定：

Collection 有哪些字段（Field）
每个字段的数据类型
哪个字段是主键

例如：

Collection Schema
├── Field Schema(id)
├── Field Schema(name)
├── Field Schema(vector)

5 Field（字段定义）

Collection 中的每一个字段，都需要定义其属性。

属性	描述	备注
`name`	要创建的集合中的字段名称	String，必填
`dtype`	字段的数据类型,如 `INT`、`VARCHAR`、`FLOAT_VECTOR`	必填
`description`	字段描述	String，选填
`is_primary`	是否设置该字段为主键字段	Boolean（true or false），必填
`auto_id`	切换以启用或禁用自动 ID（主键）分配	`True` 或 `False`，主键字段必填
`dim`	向量的维数	∈ [1, 32768]，向量字段必填

例如：

字段name	类型dtype	说明
id	INT64	主键
text	VARCHAR	文本内容
vector	FLOAT_VECTOR	文本向量
source	VARCHAR	数据来源

6 常见字段类型

类型	用途
INT64	主键、ID
VARCHAR	文本
FLOAT_VECTOR	稠密向量
SPARSE_FLOAT_VECTOR	稀疏向量（Hybrid Search）
JSON	元数据（metadata）
BOOL	布尔值
FLOAT	浮点数
ARRAY	数组

7 Milvus 的几个限制

限制	说明
Database 数量	≤64
Collection 主键	≤1
向量字段	≤4

8 本节小结

Database
    │
    ├── Collection
    │       │
    │       ├── Field:id
    │       ├── Field:text
    │       └── Field:vector
    │
    └── Collection

创建 Collection 的流程：

① 创建 Collection Schema
        ↓
② 添加多个 Field
        ↓
③ 创建 Collection
        ↓
④ 插入 Entity（数据）
        ↓
⑤ 创建索引
        ↓
⑥ 相似度搜索

3.5 索引类型和相似度度量 ¶

在Milvus中，索引是向量数据的组织方式。搜索或查询向量前，需要先指定 **索引类型(Index) 和相似度度量(Metric)**。如果未指定索引类型，则 Milvus 将默认使用FLAT(暴力搜索)。

1 索引类型 ¶

Milvus 常见向量索引包括 FLAT、IVF_FLAT、IVF_SQ8、IVF_PQ、HNSW。
其中 FLAT 属于精确检索，其余多数属于近似最近邻检索（ANNS）。

索引类型	一句话原理	优点	局限	适用场景	市场使用情况
FLAT	全量向量逐一比对	精度最高、实现最简单	数据大时最慢、算力开销高	小数据集或精度基线	较少作为生产主索引，常用于基线对比与离线评测
IVF_FLAT	先聚类分桶，再在部分桶内精确检索	速度/精度平衡好	依赖参数调优	中大规模检索（`nprobe` 越大召回越高但更慢）	主流方案，工业场景长期广泛使用
IVF_SQ8	IVF_FLAT + 标量量化压缩（如 int8）	更省内存、更快	精度有损	内存受限的大规模在线检索	较少使用，多见于强内存约束场景
IVF_PQ	IVF_FLAT + 乘积量化压缩	压缩率高，适合超大规模	调参复杂，精度通常低于 IVF_FLAT	超大规模高维检索	中等偏少，在超大规模场景有一定采用
HNSW	分层图近邻搜索	高召回下仍有较低时延	内存和建索引成本较高	高性能检索（`M`、`efConstruction`、`efSearch`）	热门方案，近年增长快

IVF_FLAT：是一种基于聚类+倒排的索引方法，先把数据分簇，再在少量相关簇中做精确距离计算，常用于在速度和召回之间做平衡的大规模检索场景。

工作流为:

先聚类（查询前）

用 k-means 把向量分成多个簇，每个簇的表示向量为簇中心。

建立倒排表/归类向量（查询前）

把每个向量按所属簇归档：簇ID -> 该簇中的向量列表。

查询处理（查询时）

先找组，再组内搜索。找出与查询向量最近的若干簇中心，在这些簇内做 FLAT 精确距离计算并返回 TopK。
搜索簇数为 nprobe, 增大 nprobe 可以搜索更多簇，提高精确度，查询时间更长。
HNSW：用“分层图结构”来快速找到最相似的向量，适合对搜索效率有高要求的场景。可参考：https://cloud.tencent.com/developer/article/2548878

2 相似度度量 ¶

在 Milvus 中，相似度度量用于衡量向量之间的相似性。根据输入数据的形式，选择特定的相似度度量方法可以获得最优的性能。
对于嵌入向量，通常使用以下指标：

欧式距离（L2）

$$ d(x,y) = \sqrt{\sum_{i=1}^n (x_i - y_i)^2} $$

内积（IP）

$$ ip(x,y) = x^T y $$

余弦相似度 (COSINE)

$$ cosine(x,y) = \frac{x^T y}{|x||y|} $$

3.6 Milvus数据库操作 ¶

先启动Docker的Milvus容器

pip install pymilvus

1 database操作 ¶

要创建本地的 Milvus 向量数据库，只需实例化一个 MilvusClient

import random
from pymilvus import MilvusClient, DataType, AnnSearchRequest, WeightedRanker, RRFRanker
import json

# 1.database操作
def operate_db():
    # 1.创建MilvusClient对象
    # 如果 uri 为本地数据库路径，代表 本地操作数据库
    # 如果 uri 为链接地址，代表 连接远程服务
    client = MilvusClient(uri="http://127.0.0.1:19530")
    print(f"client -> {client}")

    # 2.查看库中有多少database
    databases = client.list_databases()
    print(f"databases -> {databases}")

    # 3.创建milvus_demo数据库
    if "milvus_demo" not in databases:
        client.create_database("milvus_demo")
    print(f"databases -> {client.list_databases()}")

    # 4.切换数据库
    client.use_database("milvus_demo")
    print(f"databases -> {client.list_databases()}")

    # 5.删除数据库
    # client.drop_database("milvus_demo")
    # print(f"databases -> {client.list_databases()}")
    return client

client = operate_db()

2 Collections操作 ¶

在 Milvus 中，需要一个 Collections 来存储向量及其相关元数据。类比传统 SQL 数据库中的表格。创建 Collections 时，可以定义 Schema 和索引参数来配置向量规格，如维度、索引类型和相似度度量。


# 2.collection操作
def operate_collection():
    # 1.创建MilvusClient对象
    client = operate_db()
    print(f"client -> {client}")

    # 2.创建schema
    schema = client.create_schema(
        auto_id=False,   # 是否开启自动增长主键id
        enable_dynamic_field=True,  # 开启动态字段,可以插入未定义字段，未定义字段存储为JSON格式
    )

    # 3.添加字段field
    # field_name: 字段名称
    # datatype: 字段数据类型, 有 INT64,FLOAT_VECTOR,VARCHAR
    # is_primary: 是否是主键字段，默认为False，主键必须唯一且不能为空
    schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
    schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=5)
    schema.add_field(field_name="scalar1", datatype=DataType.VARCHAR, max_length=1000)

    # 4.创建collection
    print(f"collections -> {client.list_collections()}")
    if "demo_v1" in client.list_collections():
        client.drop_collection("demo_v1")
    if "demo_v1" not in client.list_collections():
        client.create_collection(collection_name="demo_v1", schema=schema)
    print(f"collections -> {client.list_collections()}")

    # 5.创建向量的索引
    # 准备索引参数
    index_params = client.prepare_index_params()
    # 为向量字段添加索引
    index_params.add_index(
        field_name="vector", # 字段名称
        index_type='AUTOINDEX', # 索引类型，空则使用默认索引，可选 FLAT, IVF_FLAT, HNSW, AUTOINDEX
        metric_type="COSINE", # 距离计算方式, COSINE, L2, IP
        index_name="vector_index", # 索引名称
    )
    # 创建索引
    client.create_index(collection_name="demo_v1", index_params=index_params)
    # 查看索引信息
    print(f"索引信息 -> {client.list_indexes(collection_name='demo_v1')}")
    print(f"指定索引详细信息 -> {client.describe_index(collection_name='demo_v1', index_name='vector_index')}")

    # 6.加载 和 释放 collection
    # 加载 collection，加载后才能查询
    print(f"demo_v1的加载状态: {client.get_load_state(collection_name='demo_v1')}")
    client.load_collection(collection_name="demo_v1")
    print(f"demo_v1的加载状态: {client.get_load_state(collection_name='demo_v1')}")
    # 释放 collection
    client.release_collection(collection_name="demo_v1")
    print(f"demo_v1的加载状态: {client.get_load_state(collection_name='demo_v1')}")
    # 删除索引
    client.drop_index(collection_name="demo_v1", index_name="vector_index")
    print(f"索引信息 -> {client.list_indexes(collection_name='demo_v1')}")
    print(f"指定索引详细信息 -> {client.describe_index(collection_name='demo_v1', index_name='vector_index')}")

    # 7.创建标量的索引
    index_params1 = client.prepare_index_params()
    index_params1.add_index(
        field_name="scalar1", # 字段名称
        index_type='AUTOINDEX', # 索引类型，空则使用默认索引，可选 FLAT, IVF_FLAT, HNSW, AUTOINDEX
        index_name="scalar1_index",
    )
    client.create_index(collection_name="demo_v1", index_params=index_params1)
    print(f"索引信息 -> {client.list_indexes(collection_name='demo_v1')}")
    print(f"指定索引详细信息 -> {client.describe_index(collection_name='demo_v1', index_name='scalar1_index')}")

operate_collection()

3 Entity操作-增删改 ¶

在 Milvus 中，实体指的是 Collections 中共享相同 Schema 的数据记录，一行数据就是一个实体。

# 3.entity实体操作，增删改
def operate_entity():
    # 1.创建MilvusClient对象
    client = operate_db()
    print(f"client -> {client}")

    # 2.创建schema
    schema = client.create_schema(
        auto_id=False,  # 是否开启自动增长主键id
        enable_dynamic_field=True,  # 开启动态字段,可以插入未定义字段，未定义字段存储为JSON格式
    )

    # 3.添加字段field
    # field_name: 字段名称
    # datatype: 字段数据类型, 有 INT64,FLOAT_VECTOR,VARCHAR
    # is_primary: 是否是主键字段，默认为False，主键必须唯一且不能为空
    schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
    schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=5)

    # 4.创建collection
    if "demo_v2" in client.list_collections():
        client.drop_collection(collection_name="demo_v2")
    if "demo_v2" not in client.list_collections():
        client.create_collection(collection_name="demo_v2", schema=schema)
    print(f"collections -> {client.list_collections()}")

    # 5.创建向量的索引
    # 准备索引参数
    index_params = client.prepare_index_params()
    # 为向量字段添加索引
    index_params.add_index(
        field_name="vector",  # 字段名称
        index_type='AUTOINDEX',  # 索引类型，空则使用默认索引，可选 FLAT, IVF_FLAT, HNSW, AUTOINDEX
        metric_type="COSINE",  # 距离计算方式, COSINE, L2, IP
        index_name="vector_index",  # 索引名称
    )
    # 创建索引
    client.create_index(collection_name="demo_v2", index_params=index_params)
    # 查看索引信息
    print(f"索引信息 -> {client.list_indexes(collection_name='demo_v2')}")
    print(f"指定索引详细信息 -> {client.describe_index(collection_name='demo_v2', index_name='vector_index')}")

    # 6.加载 collection
    # 加载 collection，加载后才能查询
    print(f"demo_v2的加载状态: {client.get_load_state(collection_name='demo_v2')}")
    client.load_collection(collection_name="demo_v2")
    print(f"demo_v2的加载状态: {client.get_load_state(collection_name='demo_v2')}")

    # 7.insert 插入数据
    data = [
        {"id": 0, "vector": [0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354,
                             0.9029438446296592], "color": "pink_8682"},
        {"id": 1, "vector": [0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501,
                             0.838729485096104], "color": "red_7025"},
        {"id": 2, "vector": [0.43742130801983836, -0.5597502546264526, 0.6457887650909682, 0.7894058910881185,
                             0.20785793220625592], "color": "orange_6781"},
        {"id": 3, "vector": [0.3172005263489739, 0.9719044792798428, -0.36981146090600725, -0.4860894583077995,
                             0.95791889146345], "color": "pink_9298"},
        {"id": 4, "vector": [0.4452349528804562, -0.8757026943054742, 0.8220779437047674, 0.46406290649483184,
                             0.30337481143159106], "color": "red_4794"},
        {"id": 5, "vector": [0.985825131989184, -0.8144651566660419, 0.6299267002202009, 0.1206906911183383,
                             -0.1446277761879955], "color": "yellow_4222"},
        {"id": 6, "vector": [0.8371977790571115, -0.015764369584852833, -0.31062937026679327, -0.562666951622192,
                             -0.8984947637863987], "color": "red_9392"},
        {"id": 7, "vector": [-0.33445148015177995, -0.2567135004164067, 0.8987539745369246, 0.9402995886420709,
                             0.5378064918413052], "color": "grey_8510"},
        {"id": 8, "vector": [0.39524717779832685, 0.4000257286739164, -0.5890507376891594, -0.8650502298996872,
                             -0.6140360785406336], "color": "white_9381"},
        {"id": 9, "vector": [0.5718280481994695, 0.24070317428066512, -0.3737913482606834, -0.06726932177492717,
                             -0.6980531615588608], "color": "purple_4976"}
    ]
    res = client.insert(collection_name="demo_v2", data=data)
    print(f"res -> {res}")
    # 插入数据到特定分区
    data = [
        {"id": 10, "vector": [-0.5570353903748935, -0.8997887893201304, -0.7123782431855732, -0.6298990746450119,
                              0.6699215060604258], "color": "red_1202"},
        {"id": 11, "vector": [0.6319019033373907, 0.6821488267878275, 0.8552303045704168, 0.36929791364943054,
                              -0.14152860714878068], "color": "blue_4150"},
        {"id": 12, "vector": [0.9483947484855766, -0.32294203351925344, 0.9759290319978025, 0.8262982148666174,
                              -0.8351194181285713], "color": "orange_4590"},
        {"id": 13, "vector": [-0.5449109892498731, 0.043511240563786524, -0.25105249484790804, -0.012030655265886425,
                              -0.0010987671273892108], "color": "pink_9619"},
        {"id": 14, "vector": [0.6603339372951424, -0.10866551787442225, -0.9435597754324891, 0.8230244263466688,
                              -0.7986720938400362], "color": "orange_4863"},
        {"id": 15, "vector": [-0.8825129181091456, -0.9204557711667729, -0.935350065513425, 0.5484069690287079,
                              0.24448151140671204], "color": "orange_7984"},
        {"id": 16, "vector": [0.6285586391568163, 0.5389064528263487, -0.3163366239905099, 0.22036279378888013,
                              0.15077052220816167], "color": "blue_9010"},
        {"id": 17, "vector": [-0.20151825016059233, -0.905239387635804, 0.6749305353372479, -0.7324272081377843,
                              -0.33007998971889263], "color": "blue_4521"},
        {"id": 18, "vector": [0.2432286610792349, 0.01785636564206139, -0.651356982731391, -0.35848148851027895,
                              -0.7387383128324057], "color": "orange_2529"},
        {"id": 19, "vector": [0.055512329053363674, 0.7100266349039421, 0.4956956543575197, 0.24541352586717702,
                              0.4209030729923515], "color": "red_9437"}
    ]
    # 创建分区
    client.create_partition(collection_name="demo_v2", partition_name="partition_1")
    # 插入数据到指定分区
    res = client.insert(collection_name="demo_v2", data=data, partition_name="partition_1")
    print(f"res -> {res}")

    # 8.upsert 更新插入数据
    # upsert = update + insert
    # 更新插入数据，根据主键id进行判断：如果存在，更新数据；如果不存在，插入
    data = [
        {"id": 0, "vector": [-0.819954382375778, 0.4479436794798608, -0.17493894838751745, -0.4248030059917294,
                             -0.8648452746018911], "color": "black_9898"},
        {"id": 1, "vector": [0.4762662251462588, -0.6942502138717026, -0.4490002642657902, -0.628696575798281,
                             0.9660395877041965], "color": "red_7319"},
        {"id": 2, "vector": [-0.8864122635045097, 0.9260170474445351, 0.801326976181461, 0.6383943392381306,
                             0.7563037341572827], "color": "white_6465"},
        {"id": 3, "vector": [0.14594326235891586, -0.3775407299900644, -0.3765479013078812, 0.20612075380355122,
                             0.4902678929632145], "color": "orange_7580"},
        {"id": 4, "vector": [0.4548498669607359, -0.887610217681605, 0.5655081329910452, 0.19220509387904117,
                             0.016513983433433577], "color": "red_3314"},
        {"id": 5, "vector": [0.11755001847051827, -0.7295149788999611, 0.2608115847524266, -0.1719167007897875,
                             0.7417611743754855], "color": "black_9955"},
        {"id": 6, "vector": [0.9363032158314308, 0.030699901477745373, 0.8365910312319647, 0.7823840208444011,
                             0.2625222076909237], "color": "yellow_2461"},
        {"id": 7, "vector": [0.0754823906014721, -0.6390658668265143, 0.5610517334334937, -0.8986261118798251,
                             0.9372056764266794], "color": "white_5015"},
        {"id": 8, "vector": [-0.3038434006935904, 0.1279149203380523, 0.503958664270957, -0.2622661156746988,
                             0.7407627307791929], "color": "purple_6414"},
        {"id": 9, "vector": [-0.7125086947677588, -0.8050968321012257, -0.32608864121785786, 0.3255654958645424,
                             0.26227968923834233], "color": "brown_7231"}
    ]
    res = client.upsert(collection_name="demo_v2", data=data)
    print(f"res -> {res}")
    # 主键是全局唯一的，和分区无关。同一个数据可以在多个分区中
    res = client.upsert(collection_name="demo_v2", data=data, partition_name="partition_1")
    print(f"res -> {res}")

    # 9.delete 删除数据
    res = client.delete(collection_name="demo_v2", filter="id in [0, 1, 2, 3]")
    print(f"res -> {res}")
    res = client.delete(collection_name="demo_v2", ids=[0, 1, 2, 3, 4, 5, 6, 7])
    print(f"res -> {res}")
    ...
operate_entity()

4 Entity简单查询 ¶

import json
# 4.entity简单查询
def query_operation():
    # 1.创建MilvusClient对象
    client = operate_db()

    # 2.单向量搜索
    res = client.search(
        collection_name="demo_v2", # 集合名称
        data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501,
       0.838729485096104]], # 搜索向量
        limit=3, # 最大搜索数量，topK
        search_params={"metric_type": "COSINE"}, # 搜索时的相似度度量方式, 如果定义索引时已经定义了metric_type，这里可以不写
        output_fields=["id", "vector"], # 搜索结果返回的字段
        )
    print(f"res -> {json.dumps(res, indent=4, ensure_ascii=False)}")
    print("-"*70)

    # 3.批量向量搜索
    res = client.search(
        collection_name="demo_v2",  # 集合名称
        data=[[0.19886812562848388, 0.06023560599112088, 0.6976963061752597, 0.2614474506242501,
               0.838729485096104],
              [0.3172005263489739, 0.9719044792798428, -0.36981146090600725, -0.4860894583077995,
               0.95791889146345]],  # 搜索向量
        limit=3,  # 最大搜索数量
        search_params={"metric_type": "COSINE"},  # 搜索时的相似度度量方式, 如果定义索引时已经定义了metric_type，这里可以不写
        output_fields=["id", "vector"],  # 搜索结果返回的字段
    )
    print(f"res -> {json.dumps(res, indent=4, ensure_ascii=False)}")
    print("-" * 70)

    # 4.分区搜索
    res = client.search(
        collection_name="demo_v2",  # 集合名称
        data=[[0.02174828545444263, 0.058611125483182924, 0.6168633415965343, -0.7944160935612321, 0.5554828317581426]],  # 搜索向量
        limit=3,  # 最大搜索数量
        search_params={"metric_type": "COSINE"},  # 搜索时的相似度度量方式, 如果定义索引时已经定义了metric_type，这里可以不写
        output_fields=["id", "vector"],  # 搜索结果返回的字段
        partition_names=["partition_1"], # 指定分区
    )
    print(f"res -> {json.dumps(res, indent=4, ensure_ascii=False)}")
    print("-" * 70)

    # 5.设置搜索返回的字段
    res = client.search(
        collection_name="demo_v2",  # 集合名称
        data=[[0.02174828545444263, 0.058611125483182924, 0.6168633415965343, -0.7944160935612321, 0.5554828317581426]],
        # 搜索向量
        limit=5,  # 最大搜索数量 topK
        search_params={"metric_type": "COSINE"},  # 搜索时的相似度度量方式, 如果定义索引时已经定义了metric_type，这里可以不写
        output_fields=["id", "vector", "color"],  # 搜索结果返回的字段
    )
    print(f"res -> {json.dumps(res, indent=4, ensure_ascii=False)}")
    print("-" * 70)

    # 6.过滤搜索
    # 过滤器filter: 指定标量列的过滤条件
    # 比如，筛选color 以red为前缀的 数据
    res = client.search(
        collection_name="demo_v2",  # 集合名称
        data=[[0.02174828545444263, 0.058611125483182924, 0.6168633415965343, -0.7944160935612321,
               0.5554828317581426]], # 搜索向量
        limit=5,  # 最大搜索数量 topK
        search_params={"metric_type": "COSINE"},  # 搜索时的相似度度量方式, 如果定义索引时已经定义了metric_type，这里可以不写
        output_fields=["id", "vector", "color"],  # 搜索结果返回的字段
        filter="color like 'red%'", # 过滤器filter: 筛选color 以red为前缀的 数据
    )
    print(f"res -> {json.dumps(res, indent=4, ensure_ascii=False)}")
    print("-" * 70)

    # 7.范围搜索
    # 如果向量库中的数据 和 query向量 完全没关系，只是用相似度+topK，只能获取 不相关的数据。
    # 需要 添加 阈值限制，只有 相似度超过阈值，才认为是相似的数据。
    # 范围搜索: radius: 定义相似度的下限阈值。搜索时只考虑超过这个相似度下限阈值的数据;
    # range_filter: 定义相似度的上限阈值。搜索时只考虑小于这个相似度上限阈值的数据;
    # COSINE: (radius, range_filter)
    search_params = {
        "metric_type": "COSINE",
        "params": {
            "radius": 0.5, # 相似度下限阈值
            "range_filter": 1, # 相似度上限阈值
        }
    }
    res = client.search(
        collection_name="demo_v2",  # 集合名称
        data=[[0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354,
               0.9029438446296592]], # 搜索向量
        limit=5,  # 最大搜索数量
        search_params=search_params,  # 搜索时的相似度度量方式, 如果定义索引时已经定义了metric_type，这里可以不写
        output_fields=["id", "vector", "color"],  # 搜索结果返回的字段
    )
    print(f"res -> {json.dumps(res, indent=4, ensure_ascii=False)}")
    print("-" * 70)
    ...
query_operation()

5 Entity混合检索 ¶

是什么：针对同时检索多个向量字段的场景, 要把多组 ANN检索结果合并后重新排序为最终的检索结果。

常用的两种混合检索策略：
1. WeightedRanker：知道哪个字段更重要。
2. RRFRanker：不确定权重时优先使用，更通用。

对比项	WeightedRanker（加权排序）	RRFRanker（倒数排序融合,更通用）
核心思想	给不同向量字段设置权重，突出重要字段，要求相似度计算方式metric_type一致	根据排名位置融合结果，平等对待每个字段，允许相似度计算方式不一致
核心公式	$Score(d)=\sum_{i=1}^{N} w_i \cdot s_i(d)$	$Score(d)=\sum_{i=1}^{N}\dfrac{1}{k+\text{rank}_i(d)}$
关键参数	$s_i(d)$ 为第 $i$ 个字段的相似度分数,权重 $w_i \in [0,1]$，$\sum w_i$ 无需为1，越大该字段影响越强	$\text{rank}_i(d)$ 为文档 $d$ 在第 $i$ 个字段检索结果中的排名，$k$ 默认 60，控制排名差异的平滑程度，越大越平滑
代码示例	`WeightedRanker(0.9, 0.1)`	`RRFRanker()` 或 `RRFRanker(100)`
适合场景	明确知道某字段更重要（如图文搜索）	不确定权重时，通用混合检索（如RAG）
优点	可突出重点字段	简单稳定，无需调参
缺点	需要人工调权重	无法突出某字段
数据示例	候选文档 A、B：两个检索结果：向量字段1分数 A=0.9，B=0.5；向量字段2分数 A=0.2，B=0.8。设权重为 0.8 和 0.3： A=0.8×0.9+0.3×0.2=0.78，B=0.8×0.5+0.3×0.8=0.64。结论：A 排名更高。	两个检索结果：向量字段1检索结果：A 第1、B 第2；向量字段2检索结果：A 第2、B 第1。取 k=60： A=1/61+1/62，B=1/62+1/61。结论：A 与 B 得分相同，更公平。

from pymilvus import WeightedRanker, RRFRanker

ranker = WeightedRanker(0.8, 0.3)
ranker = RRFRanker(100)

import random
from pymilvus import AnnSearchRequest, RRFRanker

# 5.entity混合检索
def complex_query():
    # todo 1.前期操作,构建collection
    # 1.定义schema
    client = operate_db()
    schema = client.create_schema(enable_dynamic_field=False)
    # 主键字段
    schema.add_field(
        field_name="film_id", datatype=DataType.INT64, is_primary=True, auto_id=False
    )
    # 电影文本向量
    schema.add_field(
        field_name="film_vector", datatype=DataType.FLOAT_VECTOR, dim=5
    )
    # 海报图像向量
    schema.add_field(
        field_name="poster_vector", datatype=DataType.FLOAT_VECTOR, dim=5
    )

    # 2.定义索引
    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="film_vector", # 字段名称
        index_type="IVF_FLAT", # 索引类型
        metric_type="COSINE", # 相似度度量方式
        params={"nlist": 32} # 索引参数，nlist: 索引列表大小
    )
    index_params.add_index(
        field_name="poster_vector", # 字段名称
        index_type="IVF_FLAT", # 索引类型
        metric_type="COSINE", # 相似度度量方式
        params={"nlist": 32} # 索引参数，nlist: 索引列表大小
    )

    # 3.创建集合
    # 如果集合demo_v3已存在，则删除后再创建，否则直接创建
    if client.has_collection("demo_v3"):
        client.drop_collection("demo_v3")
        client.create_collection(
            collection_name="demo_v3",
            schema=schema,
            index_params=index_params
        )
    else:
        client.create_collection(
            collection_name="demo_v3",
            schema=schema,
            index_params=index_params
        )

    # 4.插入测试数据到collection
    entities = []
    # 构造测试数据
    # 固定随机性
    random.seed(7)
    for i in range(1000):
        # 创建唯一主键
        film_id = i
        # 创建film_vector
        film_vector = [random.random() for _ in range(5)]
        # 创建poster_vector
        poster_vector = [random.random() for _ in range(5)]
        # 创建 entity
        entity = {
            "film_id": film_id,
            "film_vector": film_vector,
            "poster_vector": poster_vector,
        }
        entities.append(entity)

    # 插入测试数据
    client.insert(
        collection_name="demo_v3",
        data=entities,
    )
    # flush：将数据持久化到磁盘
    client.flush(collection_name="demo_v3")
    print(f"插入数据成功，数量为：{len(entities)}")
    print("-" * 70)

    # todo 2.混合检索
    # 1.创建2个AnnSearchRequest,对应两个向量字段，进行混合检索
    # film_vector的查询请求
    query_film_vector = [
        [0.8896863042430693, 0.370613100114602, 0.23779315077113428, 0.38227915951132996, 0.5997064603128835]
    ]
    query_search_params = {
        "data": query_film_vector, # 搜索向量
        "anns_field": "film_vector", # 搜索向量的字段名称
        "param": {"metric_type": "COSINE", "nprobe": 16}, # 搜索参数
        "limit": 2, # 最大搜索数量，topK
    }
    request_1 = AnnSearchRequest(**query_search_params)

    # poster_vector的查询请求
    query_poster_vector = [
        [0.02550758562349764, 0.006085637357292062, 0.5325251250159071, 0.7676432650114147, 0.5521074424751443]
    ]
    query_search_params = {
        "data": query_poster_vector,
        "anns_field": "poster_vector",  # 搜索向量的字段名称
        "param": {"metric_type": "COSINE", "nprobe": 16},  # 搜索参数
        "limit": 2,
    }
    request_2 = AnnSearchRequest(**query_search_params)

    # 2.构建请求列表
    requests = [request_1, request_2]

    # 3.创建混合检索策略 ranker
    """
    最终融合分数：
    Score = 0.7 * filmVector_score + 0.3 * posterVector_score
    """
    # ranker = WeightedRanker(0.7,0.3) # 加权排序
    ranker = RRFRanker() # 倒数排序

    # 4.调用hybrid_search
    hits = client.hybrid_search(
        collection_name="demo_v3", # 集合名称
        reqs=requests, # 查询请求列表
        ranker=ranker, # 混合检索策略
        output_fields=["film_id"], # 搜索结果返回的字段
        # output_fields=["film_id", "film_vector", "poster_vector"], # 搜索结果返回的字段
        limit=2, # 最大搜索数量 topK
    )
    print(f"hits -> {json.dumps(hits, indent=4, ensure_ascii=False)}")
    ...

complex_query()

client--><pymilvus.milvus_client.milvus_client.MilvusClient object at 0x0000013698C56DD0>
databases-->['milvus_demo', 'default']
databases-->['default', 'milvus_demo']
client--><pymilvus.milvus_client.milvus_client.MilvusClient object at 0x0000013698C22DD0>
databases-->['default', 'milvus_demo']
databases-->['milvus_demo', 'default']
data: [[{'id': 9510, 'distance': 0.7945637106895447, 'entity': {'film_id': 9510}}, {'id': 1121, 'distance': 0.7924863696098328, 'entity': {'film_id': 1121}}]]

6 加载现有数据 ¶

from pymilvus import MilvusClient
# client = MilvusClient("milvus_demo.db")
client = MilvusClient("http://localhost:19530")
# 查看database列表
databases = client.list_databases()
print(f"databases -> {databases}")
# 查看collection列表
collections = client.list_collections()
print(f"collections -> {collections}")

3.7 本章小结 ¶

本章节主要介绍了什么是Milvus向量数据库，并对该数据库的增删改查等操作进行了详细的讲解

4 扩展：Dense 向量与 Sparse 向量相似度示例

假设两个文档如下：

文档	内容
A	从前有一段真挚的爱情放在我的面前
B	从前有一段纯洁的友谊放在我的面前

4.1 Dense 向量相似度

Dense 向量表示句子整体语义。示例向量如下：

文档	向量
A	`[0.8, 0.6, 0.2]`
B	`[0.7, 0.5, 0.3]`

Milvus 中 Dense 向量通常使用 COSINE：

$$
\cos(A,B)=\frac{A\cdot B}{\lVert A\rVert \lVert B\rVert}
$$

计算过程：

$$
A\cdot B = 0.8\times0.7 + 0.6\times0.5 + 0.2\times0.3 = 0.92
$$

$$
\lVert A\rVert = \sqrt{0.8^2+0.6^2+0.2^2} \approx 1.02
$$

$$
\lVert B\rVert = \sqrt{0.7^2+0.5^2+0.3^2} \approx 0.91
$$

$$
\cos(A,B)=\frac{0.92}{1.02\times0.91}\approx0.99
$$

指标	结果	含义
Dense Cosine	0.99	整体语义非常接近

4.2 Sparse 向量相似度

Sparse 向量表示关键词及其权重。示例如下：

文档	稀疏表示
A	`从前:0.5，真挚:1.2，爱情:3.0，面前:0.3`
B	`从前:0.5，纯洁:1.1，友谊:2.8，面前:0.3`

展开后：

词项	从前	真挚	爱情	纯洁	友谊	面前
A	0.5	1.2	3.0	0	0	0.3
B	0.5	0	0	1.1	2.8	0.3

Milvus 中 Sparse 向量通常使用 IP（内积）：

$$
IP(A,B)=A \cdot B
$$

计算过程：

$$
IP(A,B)=0.5\times0.5 + 1.2\times0 + 3.0\times0 + 0\times1.1 + 0\times2.8 + 0.3\times0.3
$$

$$
=0.25+0.09=0.34
$$

指标	结果	含义
Sparse IP	0.34	关键词重合较少

4.3 为什么 Dense 高而 Sparse 低

类型	关注点	本例表现
Dense	句子整体语义	“真挚的爱情” 与 “纯洁的友谊” 语义相近，因此分数高
Sparse	关键词是否重合	“爱情 ≠ 友谊”，“真挚 ≠ 纯洁”，因此分数低

4.4 总结

对比项	Dense Vector（稠密向量）	Sparse Vector（稀疏向量）
特点	大部分维度都有值	大部分维度为0，仅少量维度有值
常见来源	神经网络 Embedding（BERT、Transformer、CNN）	TF-IDF、BM25、One-Hot
检索特点	擅长语义相似搜索	擅长关键词匹配
Milvus 数据类型	`DataType.FLOAT_VECTOR`	`DataType.SPARSE_FLOAT_VECTOR`
Milvus 索引类型	`HNSW`、`IVF_FLAT`等	`SPARSE_INVERTED_INDEX`
Milvus 度量方式	`COSINE`	`IP`（内积）
数据示例	完整向量`[0.12, 0.53, 0.88, 0.31]`	非0项的索引权重`{3: 0.88, 100: 0.42}`
是否节省内存	一般	非常节省
常见应用	AI语义搜索、图像搜索	文本关键词搜索、BM25检索

本文为程序员青阳原创文章，遵循 CC BY-NC-SA 4.0 版权协议，转载请附上原文链接及本声明。

原文链接：https://heliufang.github.io/posts/3cddd11/index.html