从零搭建Python大语言模型电商导购助手——从需求理解到混合召回、RAG可信推荐与FastAPI落地

pip install pydantic pandas fastapi uvicorn openai sentence-transformers rank-bm25 scikit-learn jieba

from decimal import Decimal
from pydantic import BaseModel, Field, field_validator

class Product(BaseModel):
    product_id: str = Field(min_length=1)
    name: str = Field(min_length=1)
    category: str = Field(min_length=1)
    brand: str = ""
    price: Decimal = Field(gt=0)
    stock: int = Field(ge=0)
    rating: float = Field(ge=0, le=5)
    description: str = ""
    tags: list[str] = Field(default_factory=list)
    specs: dict[str, str | int | float] = Field(default_factory=dict)
    review_summary: str = ""
    is_active: bool = True

    @field_validator("name", "category", "brand", "description", "review_summary")
    @classmethod
    def clean_text(cls, value: str) -> str:
        return " ".join(value.split())

from decimal import Decimal
from typing import Literal
from pydantic import BaseModel, Field

class UserNeed(BaseModel):
    intent: Literal["recommend", "compare", "price", "stock", "after_sale"] = "recommend"
    category: str | None = None
    budget_max: Decimal | None = None
    scenes: list[str] = Field(default_factory=list)
    must_have: list[str] = Field(default_factory=list)
    preferences: list[str] = Field(default_factory=list)
    exclude: list[str] = Field(default_factory=list)
    brand_preference: list[str] = Field(default_factory=list)
    sort_focus: list[str] = Field(default_factory=list)

import json
import re
import unicodedata

def normalize_text(text: str) -> str:
    text = unicodedata.normalize("NFKC", text)
    text = text.lower().strip()
    text = re.sub(r"\s+", " ", text)
    return text

def build_document(p: Product) -> str:
    spec_text = " ".join(f"{k}:{v}" for k, v in p.specs.items())
    raw = (
        f"名称:{p.name} 类目:{p.category} 品牌:{p.brand} "
        f"标签:{' '.join(p.tags)} 规格:{spec_text} "
        f"描述:{p.description} 评价摘要:{p.review_summary}"
    )
    return normalize_text(raw)

import jieba
from rank_bm25 import BM25Okapi

documents = [build_document(p) for p in products]
tokenized_documents = [jieba.lcut(doc) for doc in documents]
bm25 = BM25Okapi(tokenized_documents)

query = "500元以内 通勤 降噪 视频会议 麦克风清晰 耳机"
keyword_scores = bm25.get_scores(jieba.lcut(query))
keyword_rank = keyword_scores.argsort()[::-1]

from sentence_transformers import SentenceTransformer
import numpy as np

encoder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
product_vectors = encoder.encode(documents, normalize_embeddings=True)

query_text = "为这个句子生成表示以用于检索相关文章：" + query
query_vector = encoder.encode([query_text], normalize_embeddings=True)[0]
semantic_scores = product_vectors @ query_vector
semantic_rank = np.argsort(semantic_scores)[::-1]

from collections import defaultdict

def reciprocal_rank_fusion(rank_lists: list[list[str]], k: int = 60) -> dict[str, float]:
    scores = defaultdict(float)
    for ranked_ids in rank_lists:
        for rank, product_id in enumerate(ranked_ids, start=1):
            scores[product_id] += 1.0 / (k + rank)
    return dict(scores)

def budget_score(price: float, budget: float | None) -> float:
    if budget is None:
        return 1.0
    if price <= budget:
        return 1.0
    return max(0.0, 1.0 - (price - budget) / max(budget, 1.0))

def final_score(rrf_score: float, product: Product, budget: float | None, preference_score: float) -> float:
    # 示例权重仅用于说明结构，正式环境应通过离线评测/线上实验调参
    retrieval = min(rrf_score / 0.05, 1.0)
    stock = 1.0 if product.stock > 0 else 0.0
    rating = product.rating / 5.0
    return (
        0.55 * retrieval
        + 0.15 * budget_score(float(product.price), budget)
        + 0.10 * stock
        + 0.10 * rating
        + 0.10 * preference_score
    )

SYSTEM_PROMPT = """
你是电商导购决策模块，只能使用给定候选商品资料。
规则：
1. 不得编造价格、库存、规格、促销、物流、售后或健康功效。
2. 只允许推荐 candidates 中存在的 product_id。
3. 每个推荐项必须包含：匹配原因、一个可能不足或信息缺口。
4. 资料未提供时明确写“资料未提供”，不能推测。
5. 输出 JSON，不直接输出最终展示文案。
"""

from pydantic import BaseModel, Field

class RecommendationPlanItem(BaseModel):
    product_id: str
    reasons: list[str] = Field(default_factory=list)
    caveats: list[str] = Field(default_factory=list)

class RecommendationPlan(BaseModel):
    summary: str
    items: list[RecommendationPlanItem]

# 生成后先校验 product_id，再从数据库重新读取最新商品事实。
# 最终价格、库存、规格由程序渲染，而不是直接相信模型文本。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ["LLM_BASE_URL"],
)

MODEL = os.environ["LLM_MODEL"]

response = client.chat.completions.create(
    model=MODEL,
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": prompt},
    ],
    temperature=0.2,
)
raw = response.choices[0].message.content

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="E-commerce Shopping Assistant")

class RecommendRequest(BaseModel):
    query: str = Field(min_length=2, max_length=500)
    top_k: int = Field(default=3, ge=1, le=10)

class RecommendResponse(BaseModel):
    trace_id: str
    parsed_need: UserNeed
    recommendations: list[dict]

@app.post("/recommend", response_model=RecommendResponse)
def recommend(req: RecommendRequest):
    trace_id = create_trace_id()
    need = extract_need(req.query)
    candidates = retrieve_and_rerank(need, top_k=req.top_k)
    plan = generate_recommendation_plan(req.query, candidates)
    safe_items = validate_and_render(plan, candidates)
    return RecommendResponse(
        trace_id=trace_id,
        parsed_need=need,
        recommendations=safe_items,
    )

{
  "intent": "recommend",
  "category": "耳机",
  "budget_max": 500,
  "scenes": ["通勤", "视频会议"],
  "must_have": ["降噪", "麦克风清晰"],
  "preferences": ["佩戴舒适", "轻量"],
  "exclude": [],
  "sort_focus": ["匹配度", "预算", "舒适"]
}

def recall_at_k(relevant_ids: set[str], ranked_ids: list[str], k: int) -> float:
    if not relevant_ids:
        return 1.0
    hit = len(relevant_ids.intersection(ranked_ids[:k]))
    return hit / len(relevant_ids)

def budget_violation_rate(results: list[dict], budget: float) -> float:
    if not results:
        return 0.0
    bad = sum(float(x["price"]) > budget for x in results)
    return bad / len(results)

shopping_assistant/
├─ app/
│  ├─ api/              # FastAPI 路由与请求/响应模型
│  ├─ domain/           # Product、UserNeed 等领域模型
│  ├─ data/             # 数据清洗、仓储接口、实时事实读取
│  ├─ retrieval/        # BM25、向量召回、RRF
│  ├─ ranking/          # 业务特征与重排
│  ├─ llm/              # 意图抽取、结构化生成、模型客户端
│  ├─ guard/            # 事实校验、规则与风险控制
│  ├─ observability/    # trace、日志、指标
│  └─ main.py
├─ tests/
│  ├─ test_slots.py
│  ├─ test_retrieval.py
│  ├─ test_guard.py
│  └─ eval_set.jsonl
├─ scripts/
│  ├─ build_index.py
│  └─ evaluate.py
├─ requirements.txt
└─ README.md