掌柜智库项目 Day 09｜意图识别与多路检索 | 𝐺ℎ𝑎𝑜 の 𝐵𝑙𝑜𝑔

承接 item_name 识别，先回顾 Milvus 单列检索与混合检索并改造公共类，再实现 node_search_embedding、node_search_embedding_hyde 与 node_web_search_mcp 三个检索节点。

本篇要点

回顾 Milvus 单列检索与混合检索（AnnSearchRequest、WeightReranker / RrfReranker）
改造 Milvus 公共类，封装 create_requests / hybrid_search
实现 node_search_embedding 节点：重写问题的向量做混合 + 条件检索
实现 node_search_embedding_hyde 节点：用假设答案增强召回
实现 node_web_search_mcp 节点：MCP 概念、协议与百炼联网搜索工具接入

内容回顾

1、node_item_name_confirm 节点

作用：从提问与历史上下文中识别 item_names，并重写 original_query → rewritten_query；也有可能识别不到（分数低 / 完全没识别）。
参数：使用 session_id、original_query、is_stream；生成 item_names、rewritten_query、answer（可能）。
核心步骤：校验参数 → 取该会话最近 10 条聊天记录拼成 history_text → 调用大模型返回字典（item_names / rewritten_query）→ item_names 向量检索 → 置信度判断 → 按情形改写 state → 保存本次提问。
面试题：怎么确保大语言模型返回的数据是 JSON 格式？① 提示词强调返回数据格式并提供返回数据示例；② 模型的 json 模式设置；③ 做好返回值参数校验 JsonOutputParser；④ 返回字典 key 的校验（item_names 无 → []，rewritten_query 无 → original_query）。

day09 笔记的第 1 节是 Day08 中 node_item_name_confirm 的重复拷贝，不再展开，今天从第 2 节 Milvus 回顾开始。

今天内容

1、Milvus 回顾

1.1 单列检索

单列检索的核心在于“选哪一列”来做相似度比较：

注重语义 → 用稠密向量。
注重关键字命中 → 用稀疏向量。

res = client.search(
    collection_name="quick_setup",   # 搜索的集合
    ann_filed="列名",                 # 根据哪一列搜索
    data=[query_vector],             # 搜索的向量，[] -> 批量搜索
    limit=3,                         # 返回数据的条数，分数最高的 3 条
    search_params={"metric_type": "IP"},  # 相似度比较方式
    output_fields=["color"],         # 输出的字段
    filter='color like "red%" and likes > 50',  # 过滤条件
)

返回结构如下，外层长度等于 data 的长度（list - len），内层长度等于 limit：

res = [                     # len = data -> list - len
    [                       # limit
        {
            id: 主键值,      # 使用自己的 milvus 可能这个不叫 id，叫 pk 或者 chunk_id（主键的名字）
            distance: 分数,  # 分数已经排好序了！！
            entity: {
                检索字段: 值..
            }
        }
    ]
]

注意： search_params={"metric_type": "IP"} 里的相似度方式必须与当初创建集合时该列对应的方式一致！

注意： filter 表示可以添加过滤条件，且是先过滤、再相似度搜索。

1.2 混合检索

混合检索其实是单列检索的升级版本：根据一个实体的多种状态进行搜索，最终确定目标。比如搜索一种动物，通过它的外观、声音、肤色来共同确定。

混合检索常用的两种模式：

多模态搜索：对一行数据的多个字段进行搜索（外观、声音、肤色）。
稠密和稀疏向量：对一行数据的一个字段生成稠密和稀疏向量进行匹配。

混合检索的语法分三步：

第一步：创建 AnnSearchRequest

可以把每个 AnnSearchRequest 理解为一路检索。

1	AnnSearchRequest(data=[向量], ann_filed="搜索列", limit=本路搜索的数量)

第二步：配置 Rerankers 策略

Reranker 是把多路结果合并到一起的规则和策略，milvus 提供的策略有两种：

WeightReranker → 分数 + 权重。

例如：第一路 0.9、第二路 0.8、第三路 0.7，权重分别为 0.5、0.3、0.2：
1
0.9 * 0.5 + 0.8 * 0.3 + 0.7 * 0.2 = 0.83
基本使用 milvus 时权重会更多，可以控制每一路的重要性。
RrfReranker → 排名。

例如排名为 1、2、3：
1
1 / k + rank = 1/61 + 1/62 + 1/63 = xx
公平，但是不可控，不推荐！

后话：自己合并两路数据的时候，用的是 RRF + weight 的思路。

第三步：进行混合检索

res = client.hybrid_search(          # hybrid_search 根本就不会检索向量库！只是将 reqs 每路的结果进行合并！！
    collection_name="my_collection",  # 对应的集合
    reqs=reqs,                        # AnnSearchRequest 多路检索集合 [1, 2, 3 -> 2]
    ranker=ranker,                    # 6 条 -> 怎么筛选出的 2 条数据
    limit=2,                          # 最终只要两个数据
)

注意： hybrid_search 并不会去检索向量库，它做的是把 reqs 每一路的结果进行合并。

1.3 修改 Milvus 公共类

为支持多路检索与权重合并，给公共类新增两个方法：

def create_requests(
    self,
    dense_vector: list[float],
    sparse_vector: dict[int, float],
    expr: str = None,
    limit: int = 5,
):
    return create_hybrid_search_requests(
        dense_vector=dense_vector,
        sparse_vector=sparse_vector,
        expr=expr,
        limit=limit,
    )

def hybrid_search(
    self,
    *,
    collection_name: str,
    reqs: list[Any],
    ranker_weights: tuple[float, float] = (0.5, 0.5),
    norm_score: bool = False,
    limit: int = 5,
    output_fields: list[str] | None = None,
    search_params: dict | None = None,
):
    return hybrid_search(
        client=self.milvus_client,
        collection_name=collection_name,
        reqs=reqs,
        ranker_weights=ranker_weights,
        norm_score=norm_score,
        limit=limit,
        output_fields=output_fields,
        search_params=search_params,
    )

2、节点 node_search_embedding

作用： 把问题（rewritten_query）的向量与数据库 chunks 的 content（item_name + title + content）向量做相似度比较，获取匹配的切块。

参数：

使用：item_names 作为前置条件，通过 annSearchRequest(expr="item_name in [item_names]") 限定范围；并结合 rewritten_query。
生成：embedding_chunks。

步骤：

获取并校验参数（state）→ item_names、rewritten_query。
进行向量的混合 + 条件检索（item_names、rewritten_query）：
1
[{id/chunk_id: x, distance: 0.9, entity: {输出的 field}}]

对结果做统一的格式化处理：

1
2
3

[{id/chunk_id: x, distance: 0.9, entity: {输出的 field}}]
    ↓
[{id: x, 输出的 field, score:, type: milvus, url: ""}, {}, {}]

返回结果即可。

3、节点 node_search_embedding_hyde

作用： 让问题（rewritten_query）先经过大模型获取答案，再用“问题 + 答案”的向量与数据库 chunks 的 content（item_name + title + content）向量做相似度比较，获取匹配的切块。

HyDE 的核心思路：

依托大模型，基于用户原始问题生成贴合业务场景、语义饱满的假设性标准回答；
拼接标准化改写问句与生成的假设文档，构建高语义密度的检索输入；
复用 Milvus 混合向量检索链路，生成第二路增强召回结果。

参数：

使用：item_names 作为前置条件，通过 annSearchRequest(expr="item_name in [item_names]") 限定范围；并结合 rewritten_query。
生成：hyde_embedding_chunks。

步骤：

获取并校验参数（state）→ item_names、rewritten_query。
根据问题使用模型获取答案。
进行向量的混合 + 条件检索（item_names、rewritten_query、答案）：
1
[{id/chunk_id: x, distance: 0.9, entity: {输出的 field}}]

进行结果的统一格式化处理：

1
2
3

[{id/chunk_id: x, distance: 0.9, entity: {输出的 field}}]
    ↓
[{id: x, 输出的 field, score:, type: milvus, url: ""}, {}, {}]

返回结果即可。

4、节点 node_web_search_mcp

作用： 从网络搜索数据，补充数据内容。网络内容比模型内容更全、更新！这里体现的是一种“舍近求远”的思路：

1 2	舍近：tavily -> 函数（关键字）-> list 求远：bailian -> 平台 -> 网络搜索 tool -> mcp

参数：

使用：rewritten_query。
生成：web_search_docs。

4.1 MCP 是什么

mcp = Model Context Protocol，模型上下文协议，是一套标准和规范。
MCP 的作用：基于 mcp 标准的服务端和客户端可以共享 tools / resources / prompt 提示词。

对比 skill：

对比项	MCP	Skill
相同点	都是提升 agent 的能力	都在提升 agent 的能力
侧重维度	代码维度	提示词和业务点
不同点	通过网络调用获得服务端提供的代码	文件的加载

4.2 MCP 常用的具体协议

MCP 服务端 = 调用 = MCP 客户端（我们这边，如何调用服务端）。

MCP 是一个概念，真正调用要靠具体的技术：

stdio：本地协议调用（同一个电脑上的 server 与 client），本质是进程通信，非常少见。
sse：网络协议调用（服务端 URL 网络地址）。
streamablehttp：是 sse 的升级版本，网络连接更轻、性能更好。

4.3 MCP 的使用方式

方案一：硬编码

from mcp.client.stdio import stdio_client

# 只适合 stdio 协议，sse、streamablehttp 使用不同的加载方式分别加载！！
server_params = StdioServerParameters(  # 加载 mcp 服务端
    # 启动服务端使用的 Python 解释器路径（必须是服务端所在虚拟环境）
    command=r"C:\Users\admin\PycharmProjects\langchain_demo\.venv\Scripts\python.exe",
    # 要运行的 MCP 服务端文件（告诉 Python 要执行哪个文件）
    # args=[r"./mcp_server_stdio.py"],
    args=[r"./01_mcp_stdio_server.py"],
)

session.list_tools()
session.call_tool()

调用关系是：我们 -> 客户端 -> 加载 -> 1 2 3 -> 不同的 mcp server。这种方式不推荐。

方案二：框架实现（Adapter）

思路是把“多个 server 分别加载”收敛成“一种方式统一加载”：

我们 -> 客户端 -> 加载 -> 1 种方式 -> 1 2 3 -> 不同的 mcp server

我们   1
      2  ===adapter===
      3

适配器 1：langchain 的适配器。

from langchain_mcp_adapters.client import MultiServerMCPClient

MultiServerMCPClient  # 加载不同的 mcp 服务，各种协议、多少都行
MultiServerMCPClient.get_tools()  # -> tools -> create_agent(tools=tools)

缺点：只能给 langchain.create_agent 提供。

适配器 2：OpenAI 提供的 mcp 处理适配器。

方式 1：给 OpenAI 方式创建的智能体配置工具。

async def main() -> None:
    token = os.environ["MCP_SERVER_TOKEN"]
    async with MCPServerStreamableHttp(
        name="Streamable HTTP Python Server",
        params={
            "url": "http://localhost:8000/mcp",
            "headers": {"Authorization": f"Bearer {token}"},
            "timeout": 10,
        },
        cache_tools_list=True,
        max_retry_attempts=3,
    ) as server:
        agent = Agent(
            name="Assistant",
            instructions="Use the MCP tools to answer the questions.",
            mcp_servers=[server],
            model_settings=ModelSettings(tool_choice="required"),
        )

        result = await Runner.run(agent, "Add 7 and 22.")
        print(result.final_output)

asyncio.run(main())

方式 2：设计了更普遍的工具调用方式（不是智能体 / langgraph 图，而是自己代码调用 tool）。

mcp_server = MCPServerStreamableHttp(
    name="Streamable HTTP Python Server",
    params={
        "url": "http://localhost:8000/mcp",
        "headers": {"Authorization": f"Bearer {token}"},
        "timeout": 10,
    },
    cache_tools_list=True,
    max_retry_attempts=3,
)

await mcp_server.connect()
结果 = mcp_server.call_tools("工具名字", {参数})
mcp_server.cleanup()

小结一下本节要点：

MCP 是什么？tool / resources / prompt。
MCP 和 skill 的异同点？能力扩展，一个偏代码、一个偏提示词与业务。
MCP 具体的使用协议：stdio、sse、streamable_http。
MCP 具体的使用：① 硬编码（不同协议不同的加载方式）；② 适配器（langchain | openai 提供）。

4.4 接入百炼联网搜索工具

第一步：百炼开启网络搜索工具

控制台链接：https://bailian.console.aliyun.com/cn-beijing/?tab=mcp#/mcp-market/detail/WebSearch
MCP 地址：https://dashscope.aliyuncs.com/api/v1/mcps/WebSearch/mcp

bailian_web_search tool 的参数：query => rewritten、count = 5。

返回结果样例（文本内容为 JSON 字符串）：

{
  "isError": false,
  "content": [
    {
      "text": "{\"pages\":[{\"snippet\":\"...\",\"hostname\":\"知乎\",\"hostlogo\":\"...\",\"title\":\"...\",\"url\":\"https://zhuanlan.zhihu.com/p/2052758783554856438\"}, ...]}",
      "type": "text"
    }
  ]
}

第二步：检查配置文件 .env。

第三步：获取并校验参数（state）→ rewritten_query。

第四步：async 使用 openai 提供的 mcp 方式进行调用（rewritten_query）→ 查询结果。

第五步：结果解析。

第六步：返回列表即可。

4.5 今日内容小结

item_name_confirm 节点：混合检索的概念与使用——① 创建 AnnSearchRequest；② 创建 Reranker（权重 | 排名）；③ 进行混合检索，HyDE 结果合并和排序；检索 → 确认 | 可选 → 修改 state。
问题检索和建设性答案检索：混合检索 item_name in item_names。
web_mcp_search：MCP 概念、作用、使用（langchain | openai）、开通百炼的 mcp 联网搜索服务、实现代码。