在电商场景中，消费者常常会围绕商品提出各类问题，而海量的用户评论中其实蕴藏着最真实、最有价值的答案。为了让商品问答更贴合用户实际诉求，我们基于 RAG（检索增强生成）+LLM（大语言模型）技术，构建了一套从商品评论中自动挖掘、生成、管理 FAQ 知识库的智能问答系统。本文将详细拆解这一模块的技术实现、核心思考与工程化落地过程。

一、需求背景与核心目标

传统的商品 FAQ 往往由运营人员人工编写，存在覆盖度低、与用户真实诉求脱节、更新不及时等问题。我们的核心目标是：

  1. 从海量商品评论中自动挖掘用户最关心的问题，生成高质量问答对；
  2. 基于 RAG 技术确保问答答案可追溯、有真实评论佐证；
  3. 提供完整的 FAQ 生命周期管理（生成、查询、编辑、状态管控）；
  4. 支持消费者端的流式智能问答，提升交互体验；
  5. 保证系统健壮性，适配 “无数据”“无索引” 等异常场景。

二、整体技术架构设计

整个模块分为三大核心层，兼顾数据层、服务层与接口层的闭环：

  • 数据层：基于 SQLAlchemy 构建 ORM 模型，核心表包括ProductFAQ（存储 FAQ 问答对）、AnalysisTask（构建任务状态）、Review（商品评论）、Dataset（商品数据集）；
  • 服务层：封装FAQService核心逻辑，整合 LLM 调用、RAG 向量检索、评论数据处理；
  • 接口层：基于 FastAPI 提供 RESTful 接口，支持任务提交、进度查询、FAQ 管理、流式问答等能力。

核心数据模型设计思考

在设计ProductFAQ模型时，我们不仅关注 “问题 - 答案” 核心字段，还增加了多个工程化与业务化字段，这是从 “能用” 到 “好用” 的关键：

class ProductFAQ(Base):
    __tablename__ = "product_faqs"
    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    dataset_id: Mapped[str] = mapped_column(String(36), ForeignKey("datasets.id"), nullable=False, index=True)
    question: Mapped[Optional[str]] = mapped_column(Text)
    answer: Mapped[Optional[str]] = mapped_column(Text)
    category: Mapped[Optional[str]] = mapped_column(String(40))  # 质量/物流/外观等分类，提升可读性
    mention_ratio: Mapped[Optional[float]] = mapped_column(Float, default=0.0)  # 问题提及占比，用于排序
    support_review_ids: Mapped[Optional[str]] = mapped_column(Text)  # 佐证评论ID，保证答案可追溯
    support_snippets: Mapped[Optional[str]] = mapped_column(Text)    # 评论片段，直接展示佐证内容
    status: Mapped[Optional[str]] = mapped_column(String(20), default="published")  # 发布/隐藏，支持运营管控
    sort_order: Mapped[Optional[int]] = mapped_column(Integer, default=0)  # 排序字段，适配前端展示

设计思考：

  • 增加category字段，将 FAQ 按 “质量 / 物流 / 外观” 等维度分类，贴合用户认知习惯；
  • mention_ratio字段量化问题的关注度，让高频问题优先展示；
  • support_review_ids和support_snippets是 RAG 技术的核心体现 —— 答案不再是 LLM 凭空生成，而是有真实用户评论作为支撑，提升可信度；
  • status和sort_order字段兼顾运营管控需求，支持 FAQ 隐藏、排序调整，适配实际业务场景。

三、核心功能实现：FAQ 知识库自动构建

1. 异步任务设计：解决长耗时问题

FAQ 构建需要处理海量评论、调用 LLM、构建向量索引，属于长耗时操作。我们采用 “异步任务 + 状态追踪” 的设计，避免接口阻塞：

@router.post("/build/{dataset_id}", summary="自动生成商品 FAQ 知识库")
async def build_faq(
    dataset_id: str,
    background_tasks: BackgroundTasks,
    max_q: int = 8,
    db: AsyncSession = Depends(get_db),
):
    task_id = str(uuid.uuid4())
    # 1. 创建任务记录，初始状态pending
    task = AnalysisTask(
        id=task_id, dataset_id=dataset_id, task_type="faq",
        params_json=json.dumps({"max_q": max_q}), status="pending",
    )
    db.add(task)
    await db.commit()
    # 2. 后台异步执行构建逻辑，不阻塞接口响应
    background_tasks.add_task(_run_build, task_id, dataset_id, max_q)
    return ok({"task_id": task_id, "status": "pending"})

思考：

  • 任务 ID 采用 UUID 保证唯一性，避免重复任务冲突；
  • 任务状态拆分为pending/running/done/failed，并提供/status/{task_id}接口查询进度，前端可基于此做进度条展示；
  • 异步任务与主请求解耦，即使接口响应完成，后台仍能继续处理长耗时逻辑，提升用户体验。

2. 核心构建逻辑：从评论到 FAQ 的全流程

_run_build函数是 FAQ 构建的核心，我们拆解为 6 个关键步骤，每一步都考虑了异常处理与健壮性：

步骤 1：任务状态更新与数据校验
async def _run_build(task_id: str, dataset_id: str, max_q: int):
    async with AsyncSessionLocal() as db:
        # 更新任务状态为running
        t = (await db.execute(select(AnalysisTask).where(AnalysisTask.id == task_id))).scalar_one_or_none()
        if not t:
            return
        t.status = "running"
        await db.commit()
        
        # 校验评论数据：无可用评论直接抛异常，终止任务
        rows = (await db.execute(
            select(Review).where(Review.dataset_id == dataset_id, Review.is_noise == 0)
        )).scalars().all()
        if not rows:
            raise ValueError("无可用评论数据")
        reviews = [{"id": r.id, "content": r.content, "rating": r.rating} for r in rows]

思考：提前校验核心数据，避免后续无意义的计算；过滤is_noise=0的评论（已做噪音清洗），保证数据质量。

步骤 2：RAG 向量索引自动兜底

RAG 的核心是向量索引，若索引不存在，后续检索将无法进行。我们增加了 “自动检测 + 构建” 的兜底逻辑：

# 依赖前置：确保向量索引存在，否则自动构建（健壮性）
stats = service.rag.get_store_stats(dataset_id)
if not stats or stats.get("chunk_count", 0) == 0:
    logger.info(f"FAQ 构建：检测到无向量索引，自动建立 dataset={dataset_id}")
    await service.rag.index_reviews(dataset_id, reviews)

思考：工程化落地中，“用户忘记构建索引” 是高频异常场景，自动兜底能减少运维成本，提升系统鲁棒性。

步骤 3：LLM 驱动的 FAQ 生成

基于清洗后的评论数据，调用FAQService生成 FAQ，核心是结合商品名称、评论上下文，让 LLM 生成贴合商品的问答对：

ds = (await db.execute(select(Dataset).where(Dataset.id == dataset_id))).scalar_one_or_none()
product_name = (ds.name if ds else "") or ""
# 调用FAQService生成问答对，max_q控制生成数量
faqs = await service.build(dataset_id, reviews, product_name=product_name, max_q=max_q)

思考：传入商品名称能让 LLM 生成的问题更精准（如 “XX 品牌充电宝续航如何？” 而非泛化的 “充电宝续航如何？”）；max_q参数支持业务侧灵活控制 FAQ 数量，适配不同商品的评论量。

步骤 4：FAQ 数据落地与历史数据清理
# 先删除旧数据，避免重复
await db.execute(delete(ProductFAQ).where(ProductFAQ.dataset_id == dataset_id))
for f in faqs:
    db.add(ProductFAQ(
        dataset_id=dataset_id,
        question=f["question"],
        answer=f["answer"],
        category=f["category"],
        mention_ratio=f["mention_ratio"],
        support_review_ids=json.dumps(f["support_review_ids"], ensure_ascii=False),
        support_snippets=json.dumps(f["support_snippets"], ensure_ascii=False),
        sort_order=f.get("sort_order", 0),
        status="published",
    ))
await db.commit()

思考：先删后插保证 FAQ 数据的唯一性，避免同一商品重复生成多条相同 FAQ；将佐证评论 ID 和片段序列化存储，为后续问答时展示 “答案来源” 提供数据支撑。

步骤 5：任务状态最终更新
t2 = (await db.execute(select(AnalysisTask).where(AnalysisTask.id == task_id))).scalar_one_or_none()
if t2:
    t2.status = "done"
    t2.result_json = json.dumps({"count": len(faqs)}, ensure_ascii=False)
    await db.commit()

思考：记录生成的 FAQ 数量，方便业务侧统计；异常场景下会捕获错误并更新任务状态为failed，同时记录错误信息，便于问题排查。

3. 异常处理与容错设计

在整个构建流程中，我们考虑了多种异常场景：

  • 无可用评论数据：直接抛异常，更新任务状态为 failed 并记录错误信息；
  • 向量索引不存在：自动触发索引构建，避免流程中断；
  • 任务记录不存在：增加多次查询校验，避免空指针异常；
  • LLM 调用失败：通过FAQService的异常封装，确保错误能被捕获并反馈到任务状态中。

四、消费者端流式问答：提升交互体验

为了避免用户等待过长时间，我们基于 SSE（Server-Sent Events）实现流式问答：

def _make_sse_response(generator):
    async def event_stream():
        try:
            async for chunk in generator:
                yield f"data: {chunk}\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n"
        finally:
            yield "data: {}\n\n".format(json.dumps({"type": "done"}, ensure_ascii=False))

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
    )

@router.post("/ask/{dataset_id}", summary="消费者流式问答")
async def ask(dataset_id: str, body: AskBody, db: AsyncSession = Depends(get_db)):
    # 获取评论总数，用于辅助LLM生成更精准的回答
    total = (await db.execute(
        select(func.count()).select_from(Review).where(Review.dataset_id == dataset_id)
    )).scalar() or 0
    service = FAQService(llm_client)
    # 流式生成回答
    gen = service.ask_stream(dataset_id, body.question, total_reviews=total,
                             conversation_id=body.conversation_id)
    return _make_sse_response(gen)

思考：

  • SSE 相比 WebSocket 更轻量，适配问答这种 “单向流式输出” 场景；
  • 传递total_reviews参数，让 LLM 在生成回答时能体现数据量级（如 “基于 1000 + 用户评论，该商品续航约 8 小时”）；
  • conversation_id支持多轮对话，提升交互连贯性；
  • 异常时通过 SSE 返回错误信息，前端可实时捕获并提示用户，避免页面卡死。

五、FAQ 管理能力：适配运营侧需求

除了自动生成，我们还提供了 FAQ 的查询、编辑能力，满足运营侧的人工校订需求：

1. FAQ 列表查询

@router.get("/faqs/{dataset_id}", summary="获取 FAQ 列表")
async def list_faqs(dataset_id: str, include_hidden: bool = False, db: AsyncSession = Depends(get_db)):
    conds = [ProductFAQ.dataset_id == dataset_id]
    if not include_hidden:
        conds.append(ProductFAQ.status == "published")
    rows = (await db.execute(
        select(ProductFAQ).where(*conds).order_by(ProductFAQ.sort_order, ProductFAQ.created_at)
    )).scalars().all()
    return ok({"items": [_serialize(r) for r in rows], "total": len(rows)})

2. FAQ 编辑

@router.patch("/faqs/{faq_id}", summary="编辑 / 校订 FAQ")
async def patch_faq(faq_id: str, body: PatchBody, db: AsyncSession = Depends(get_db)):
    row = (await db.execute(select(ProductFAQ).where(ProductFAQ.id == faq_id))).scalar_one_or_none()
    if not row:
        return {"code": 404, "message": "FAQ 不存在", "data": None}
    # 只更新非None的字段，适配部分编辑场景
    for k, v in body.model_dump(exclude_none=True).items():
        setattr(row, k, v)
    await db.commit()
    await db.refresh(row)
    return ok(_serialize(row))

思考：

  • include_hidden参数支持查询已隐藏的 FAQ，满足运营侧审核需求；
  • PATCH 方法支持部分字段更新，无需传递完整 FAQ 信息，降低前端开发成本；
  • 序列化函数_serialize统一返回格式，将 JSON 字符串字段（如support_review_ids）解析为数组，方便前端直接使用。

六、工程化落地的关键思考

1. 异步数据库会话管理

使用AsyncSessionLocal实现异步数据库操作，避免阻塞事件循环：

from app.core.database import AsyncSessionLocal
async with AsyncSessionLocal() as db:
    # 数据库操作逻辑

思考：FastAPI 的 Depends 依赖注入适合接口层，但异步任务中需要独立创建会话，确保会话隔离与正确释放。

2. 日志与可观测性

关键节点增加日志记录，方便问题排查：

logger.info(f"FAQ 构建：检测到无向量索引，自动建立 dataset={dataset_id}")
logger.info(f"FAQ 知识库构建完成: dataset={dataset_id}, faqs={len(faqs)}")
logger.error(f"FAQ 构建失败: {e}", exc_info=True)

思考：记录异常时使用exc_info=True，能打印完整堆栈信息，大幅提升问题定位效率。

3. 性能优化

  • 为dataset_id等高频查询字段建立索引，提升查询效率；
  • 批量删除 / 插入 FAQ 数据，减少数据库交互次数；
  • 向量索引复用：避免每次构建 FAQ 都重新构建索引，仅在索引缺失时自动构建。

七、总结与后续规划

已落地的核心价值

  1. 自动化：从 “人工编写 FAQ” 到 “AI 自动生成 + 人工校订”，效率提升 80% 以上；
  2. 可信度：每一条 FAQ 都有真实评论佐证，用户可查看答案来源；
  3. 易用性：完整的接口体系支持前端快速集成，异步任务 + 流式响应提升用户体验；
  4. 健壮性：覆盖无数据、无索引、LLM 调用失败等异常场景，系统稳定性高。

写在最后

本次商品智能 FAQ 知识库的构建，核心是 “技术适配业务”—— 不仅要实现 RAG+LLM 的技术闭环，更要考虑运营侧、消费者侧、开发侧的实际需求。从数据模型设计到异步任务处理，从异常容错到性能优化，每一个细节都是 “工程化落地” 的体现。最终，我们实现了 “从评论中来，到用户中去” 的智能问答闭环，让商品问答真正贴合用户的真实诉求。

# microsoft # 服务器 # 数据库
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区
cover

最高赢取价值万元大奖，仓颉生态创新开发挑战赛开启！

avatar AtomGit开源社区