用户真实输入

背后核心需求

传统搜索痛点

产品化影响

“年货”

坚果、糖果、礼盒等具体品类，需多样化选择

关键词“年货”太泛，命中率极低

用户找不到心仪商品，直接退出

“除了肉之外的礼盒”

排除肉类，需要其他品类礼盒（坚果、茶叶等）

无法处理否定语义，仍会推荐肉类礼盒

用户觉得“不智能”，体验打折

“送女朋友的礼物”

结合用户偏好（如之前买过美妆），推荐适配女生的商品

无用户记忆，推荐千篇一律

推荐精准度低，转化率上不去

“上次买的那种外套”

检索历史购买记录，匹配相似款式

无历史记忆能力，无法关联过往行为

回头客体验差，复购率受影响

plain text
┌─────────────────────────────────────────────────────────────────────┐
│                    购物Agent搜索反馈回路（产品化版）                 │
│                                                                     │
│   START → 记忆检索 → 查询理解 → 执行搜索 → 质量评估 → 条件路由       │
│                                  │                                  │
│                   （改进查询）← ｜        ｜→ 精选商品 → 记忆记录 → END
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

python
from langgraph.graph import StateGraph, START, END

# 1. 定义状态（Agent的“工作记忆”，产品化核心：所有信息可追溯、可扩展）
class SearchFeedbackState(TypedDict):
    # 输入层（用户相关）
    user_query: str          # 用户原始需求
    user_id: str             # 关联用户，用于记忆绑定
    language: str            # 语言标识，适配多语言场景
   
    # 理解层（意图结构化）
    query_understanding: Optional[QueryUnderstanding]  # 结构化意图结果
    current_keyword: str     # 当前搜索关键词
   
    # 搜索层（迭代核心）
    search_results: list[dict]       # 当前轮搜索结果
    best_results: list[dict]        # 历史最佳结果（不降级）
    best_quality_score: float       # 历史最高质量分
    search_history: Annotated[list, add]  # 搜索历史（自动追加）
    iteration_count: int            # 迭代轮次（≤3）
    search_cache: dict[str, list]   # 关键词缓存，避免重复查询
   
    # 记忆层（个性化核心）
    long_term_memory: str           # 用户偏好摘要
    memory_preferences: list[str]   # 具体偏好列表
   
    # 输出层（用户可见）
    selected_products: list[dict]   # 精选商品列表
    natural_message: str            # 自然语言推荐话术

# 2. 构建工作流（产品化设计：流程可配置、节点可替换）
def build_search_workflow() -> StateGraph:
    workflow = StateGraph(SearchFeedbackState)

    # 注册7个核心节点（每个节点对应一个产品功能）
    workflow.add_node("memory_retrieval", memory_retrieval_node)  # 记忆检索
    workflow.add_node("query_understanding", query_understanding_node)  # 意图理解
    workflow.add_node("search", search_node)  # 执行搜索
    workflow.add_node("evaluate", evaluate_node)  # 质量评估
    workflow.add_node("improve", query_improve_node)  # 查询改进
    workflow.add_node("select", selection_node)  # 商品精选
    workflow.add_node("memory_record", memory_record_node)  # 记忆记录

    # 线性流程（确保用户体验流畅）
    workflow.add_edge(START, "memory_retrieval")
    workflow.add_edge("memory_retrieval", "query_understanding")
    workflow.add_edge("query_understanding", "search")
    workflow.add_edge("search", "evaluate")

    # 核心：条件路由（Agent自主决策的关键，产品化核心逻辑）
    workflow.add_conditional_edges(
        "evaluate",
        should_continue_search,  # 路由判断函数
        {"improve": "improve", "select": "select", END: END}
    )

    # 反馈闭环（改进后重新搜索，形成自主进化）
    workflow.add_edge("improve", "search")
    workflow.add_edge("select", "memory_record")
    workflow.add_edge("memory_record", END)

    return workflow

python
async def memory_retrieval_node(state):
    user_id = state.get("user_id", "")

    # 产品化优化1：300ms硬超时（记忆是加分项，不能拖慢搜索体验）
    try:
        memory_result = await asyncio.wait_for(
            search_memory_safe(
                query=state["user_query"],
                user_id=user_id,
                limit=3,  # 只取3条核心偏好，避免信息过载
                scope="shopping"  # 作用域隔离，不混入无关记忆
            ),
            timeout=0.3
        )
    except asyncio.TimeoutError:
        # 产品化优化2：超时降级，不影响主流程
        return {"long_term_memory": "", "memory_preferences": []}

    return {
        "long_term_memory": memory_result["summary"],
        "memory_preferences": memory_result.get("memories", [])[:3]
    }

python
async def query_understanding_node(state):
    # 产品化优化：传入真实品类，避免LLM幻觉
    available_categories = await get_available_categories(max_categories=30)

    # 一次LLM调用，输出完整结构化意图
    understanding = unified_query_understanding(
        user_query=state["user_query"],
        available_categories=available_categories,  # 锚定真实数据
        long_term_memory=state.get("long_term_memory", ""),  # 结合用户记忆
        conversation_history=state.get("conversation_history", "")
    )

    # 合并关键词，用于多关键词批量搜索（提升命中率）
    all_keywords = [understanding["primary_keyword"]] + understanding["alternative_keywords"]

    return {
        "query_understanding": understanding,
        "current_keyword": " ".join(all_keywords),
        "is_multi_keyword": True
    }

用户输入

核心关键词

备选关键词

排除词

用户感知

“年货”

坚果

零食、礼盒、糖果、茶叶

无

 Agent自动推荐多种年货品类，不用用户再细化

“除了肉的礼盒”

坚果礼盒

茶叶礼盒、零食礼盒、水果礼盒

肉、生鲜、海鲜

自动排除不想要的品类，精准匹配需求

分词策略

作用

产品价值

示例

ik_smart + AND

精确匹配所有关键词，不遗漏核心信息

避免“关键词部分匹配”导致的不相关结果

“红色连衣裙” → 必须同时包含“红色”和“连衣裙”

phrase短语匹配

匹配连续词序，保证语义连贯

避免“词序颠倒”导致的误解