我要投稿

深度实战：我用 Dify 复刻了 1688 的 AI 搜索，“多路召回”才是灵魂

发布日期：2025-08-20 13:13:30 浏览次数： 1525

作者：dify实验室

微信搜一搜，关注“dify实验室”

我们每天都在用各种搜索，但你有没有觉得，传统的“关键词搜索”有时候真的很“笨”？你搜“夏天穿的透气T恤”，它可能因为你不懂行话，没输入“速干”、“涤纶”这些词，就给你推荐一堆厚重的纯棉款。

最近，我接到了一个极具挑战的任务：构建一个能听懂“人话”的工业级AI搜索系统，就像 1688 那样，能精准理解复杂的采购需求。这篇文章，就是我从0到1，用 Dify 把这套系统复刻出来的完整记录。准备好了吗？我们出发！

核心1：我踩过的坑——“一把梭哈”的 LLM 陷阱

一开始，我的想法很天真。现在的大语言模型（LLM）这么强大，我能不能把用户需求和所有商品资料，一股脑儿地喂给它，让它直接告诉我答案？我把这个不成熟的想法称为“一把梭哈法”。

结果呢？惨败！

又慢又贵：
每次查询都要处理海量的商品文本，Token 消耗巨大，响应时间长到无法忍受。
事实幻觉：
LLM 开始“一本正经地胡说八道”，给我推荐一些根本不存在的商品组合。
无法扩展：
商品库稍微大一点，上下文窗口就直接被撑爆了。

这次失败让我彻底明白：LLM 是一个顶级的“战略家”，但不应该让它去做“地毯式搜索”这种脏活累活。

核心2：顿悟时刻——“专家团队”与“多路召回”架构

痛定思痛后，我悟出了一个绝佳的比喻：一个强大的AI搜索系统，应该像一个高效的“专家团队”，而不是一个孤军奋战的天才。

在这个团队里，LLM 扮演 “项目经理” 的角色，它负责理解用户的复杂需求，并将其拆解成具体的任务指令。然后，它把这些指令分发给三个“专家”成员并行处理：

1. 向量检索专家 (Vector Search): 它最懂“感觉”。负责理解语义，处理“夏天户外运动穿的”这种模糊、感性的需求。
2. 文本检索专家 (Text Search): 它是个“老古董”，最认“死理”。负责处理“纯棉”、“T恤”这种必须包含的关键词。
3. 属性过滤专家 (Attribute Filter): 它是“数据控”。负责处理“价格低于50元”、“供应商在广州”这种精确的结构化数据筛选。

这三位专家从各自的渠道把可能符合要求的商品ID（候选人）找出来，这个过程就叫做“多路召回”。最后，再由“项目经理”LLM 对所有候选人进行最终面试（重排序），选出最优答案。

而 Dify 的工作流画布，就是搭建这个“专家团队”协同工作的完美“办公室”！

核心3：实战步骤——用 Dify 组建你的“AI搜索专家团队”

接下来，就是最激动人心的“保姆级”实操环节。我将带领你一步步构建后端服务，并在Dify中完成工作流编排。

Phase 1: 搭建三位“专家”的微服务 (后端环境)

我们使用 Docker 来快速部署三位专家的“办公室”和“资料库”。

Qdrant:
向量数据库，供“向量检索专家”使用。
Meilisearch:
全文搜索引擎，供“文本检索专家”使用。
SQLite:
关系型数据库，供“属性过滤专家”使用。

（为节省篇幅，此处聚焦于 Dify 的工作流搭建。）

Phase 2: 在 Dify 中编排工作流 (核心环节)

现在，进入Dify，让我们把整个流程串起来。

第一步：意图识别 (LLM担任项目经理)

我们创建一个 LLM 节点，让它把用户的自然语言查询分解成三路专家能看懂的结构化指令。

用户输入:
"帮我找一款适合夏天户外运动的速干T恤，要求是纯棉材质，价格在50元以内，供应商在广州地区。"

LLM拆解输出:
semantic_query
"夏天户外运动速干 T恤" (给向量专家)
keyword_query
"速干T恤纯棉" (给文本专家)
max_price
50 (给属性专家)
location
"广州" (给属性专家)

第二步：并行召回 (三位专家同时开工)

使用 Dify 的并行分支功能，同时创建三个 HTTP 请求节点，分别去调用我们搭建好的三个微服务 API，并将上一步的输出作为参数传入。

第三步：结果融合与详情获取 (汇总候选人)

三位专家都交回了他们找到的商品ID列表。我们需要用一个代码节点，将这些ID合并、去重，并调用服务获取所有候选商品的详细信息。

# Python 代码节点 
def main(v_res: list, t_res: list, a_res: list) -> dict:
    # 1. 合并并去重所有召回的 ID
    seen_ids = set()
    all_ids = []
    for res_list in [v_res, t_res, a_res]:
        for item in res_list:
            product_id = item.get('id')
            if product_id and product_id not in seen_ids:
                seen_ids.add(product_id)
                all_ids.append(product_id)

    # 2. 调用服务批量获取商品详情
    full_details = get_product_details(all_ids)

    return {"fused_list": full_details}