1
def _semantic_search(query_text: str, n_results: int, db: Session) -> List[SearchResult]:
2
    # ① 查询转向量
3
    query_vec = get_embedding(query_text)
4

5
    # ② ChromaDB 检索
6
    chroma_results = collection.query(
7
        query_embeddings=[query_vec],
8
        n_results=n_results,
9
        include=["metadatas", "distances"]
10
    )
11

12
    # ③ 回查 SQLite，组装结果
13
    results = []
14
    for i in range(n_results):
15
        metadata = chroma_results["metadatas"][0][i]
16
        distance = chroma_results["distances"][0][i]
17
        similarity = round(1 - distance, 4)
18

19
        doc_id = metadata["document_id"]
20
        chunk_idx = metadata["chunk_index"]
21

22
        chunk = db.query(DocumentChunk).filter(
23
            DocumentChunk.document_id == doc_id,
24
            DocumentChunk.chunk_index == chunk_idx
25
        ).first()
26

27
        if chunk:
28
            document = db.query(Document).filter(Document.id == doc_id).first()
29
            results.append(SearchResult(
30
                title=document.title if document else "未知",
31
                chunk_content=chunk.content,
32
                similarity=similarity,
33
                document_id=doc_id,
34
                chunk_index=chunk_idx
35
            ))
36

37
    return results

1
def build_system_prompt(results: List[SearchResult]) -> str:
2
    # 空结果兜底
3
    if not results:
4
        return "没有检索到与用户问题相关的资料。请礼貌地告诉用户：根据现有知识库无法回答该问题。"
5

6
    # 格式化每个片段
7
    context_blocks = []
8
    for i, r in enumerate(results, start=1):
9
        block = f"[{i}] 来源：《{r.title}》第{r.chunk_index}段（相关度：{r.similarity:.2f}）\n{r.chunk_content}"
10
        context_blocks.append(block)
11

12
    context = "\n\n".join(context_blocks)
13

14
    # 拼接完整 prompt
15
    prompt = f"""你是一个基于知识库的问答助手。请严格根据下面提供的资料片段回答用户问题。
16

17
【资料片段】
18
{context}
19

20
【回答要求】
21
1. 只基于上面的资料回答，不要编造资料中没有的信息。
22
2. 如果资料不足以回答问题，请明确说明"根据现有资料无法回答"。
23
3. 如果引用了某个片段，请在回答末尾标注来源，格式如：[来源：《标题》第N段]
24
4. 保持简洁清晰。"""
25

26
    return prompt

1
if not results:
2
    return "没有检索到与用户问题相关的资料。请礼貌地告诉用户..."

1
block = f"[{i}] 来源：《{r.title}》第{r.chunk_index}段（相关度：{r.similarity:.2f}）\n{r.chunk_content}"

1
[1] 来源：《Python编程语言介绍》第0段（相关度：0.82）
2
Python 广泛应用于 Web 开发、数据分析、人工智能等领域。

1
Prompt 模板结构：
2
├─ 角色设定（你是谁）
3
├─ 资料片段 [1] [2] [3]（给他什么）
4
└─ 回答要求（必须怎么做）
5
   ├─ 只基于资料
6
   ├─ 不够就说无法回答
7
   └─ 标注来源

1
async def generate_rag_stream(query: str, results: List[SearchResult]):
2
    # ① 拼 system prompt
3
    system_prompt = build_system_prompt(results)
4

5
    # ② 组装 messages
6
    messages = [
7
        {"role": "system", "content": system_prompt},
8
        {"role": "user", "content": query}
9
    ]
10

11
    # ③ 调 LLM（流式）
12
    response = client.chat.completions.create(
13
        model='deepseek-ai/DeepSeek-V3.2',
14
        messages=messages,
15
        stream=True,
16
        extra_body={"enable_thinking": True}
17
    )
18

19
    # ④ 逐块处理
20
    done_thinking = False
21
    for chunk in response:
22
        if chunk.choices:
23
            thinking = chunk.choices[0].delta.reasoning_content
24
            answer = chunk.choices[0].delta.content
25

26
            if thinking and thinking != '':
27
                data = {"type": "thinking", "content": thinking}
28
            elif answer and answer != '':
29
                if not done_thinking:
30
                    data = {"type": "divider", "content": "\n\n === Final Answer ===\n"}
31
                    done_thinking = True
32
                else:
33
                    data = {"type": "answer", "content": answer}
34
            else:
35
                continue
36

37
            yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

1
messages = [
2
    {"role": "system", "content": system_prompt},   # ← 动态拼进了检索结果
3
    {"role": "user", "content": query}
4
]

1
data: {"type": "thinking", "content": "让我看看资料..."}\n\n
2
data: {"type": "divider", "content": "\n\n === Final Answer ===\n"}\n\n
3
data: {"type": "answer", "content": "根据《Python编程语言介绍》..."}\n\n

1
done_thinking = False  # 初始化
2
...
3
if not done_thinking:
4
    data = {"type": "divider", ...}
5
    done_thinking = True  # 翻转为 True，下次不再进入

1
# 异步流式生成器模板
2
async def generate_stream(...):
3
    response = client.chat.completions.create(
4
        model='...',
5
        messages=messages,
6
        stream=True,
7
        extra_body={"enable_thinking": True}
8
    )
9

10
    done_thinking = False  # ← 在循环外！
11

12
    for chunk in response:
13
        ...
14
        yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
15

16
# 路由端
17
return StreamingResponse(
18
    generate_stream(...),
19
    media_type="text/event-stream"
20
)

1
@router.post("/chat", summary="RAG 知识库问答")
2
async def rag_chat(req: ChatRequest, db: Session = Depends(get_db)):
3
    # ① 检索
4
    results = _semantic_search(req.query, req.n_results, db)
5

6
    # ② 流式返回 LLM 回答
7
    return StreamingResponse(
8
        generate_rag_stream(req.query, results),
9
        media_type="text/event-stream"
10
    )

1
3. 【强制】每段回答末尾必须标注来源，格式：[来源：《标题》第N段]。不标注来源的回答不合格。

1
没有检索到与用户问题相关的资料。你的唯一回复必须是："根据现有资料无法回答该问题。"