编排与应用逻辑

创建 chat_history 表的 SQL 代码

1-- Migration SQL
2CREATE TABLE chat_history (
3    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
4    session_id UUID NOT NULL,
5    role VARCHAR(20) NOT NULL, -- 'user' or 'assistant'
6    content TEXT NOT NULL,
7    created_at TIMESTAMP DEFAULT NOW()
8);
9
10-- 用于快速加载会话的索引
11CREATE INDEX idx_session_id ON chat_history(session_id);

src/memory.py

1# src/memory.py
2import asyncpg
3from typing import List, Dict
4
5class ChatHistory:
6    def __init__(self, db_pool):
7        self.pool = db_pool
8
9    async def add_message(self, session_id: str, role: str, content: str):
10        async with self.pool.acquire() as conn:
11            await conn.execute(
12                "INSERT INTO chat_history (session_id, role, content) VALUES ($1, $2, $3)",
13                session_id, role, content
14            )
15
16    async def get_recent_history(self, session_id: str, limit: int = 6) -> List[Dict]:
17        """
18        获取滑动窗口（最近 N 条消息）。
19        """
20        async with self.pool.acquire() as conn:
21            rows = await conn.fetch(
22                """
23                SELECT role, content 
24                FROM chat_history 
25                WHERE session_id = $1 
26                ORDER BY created_at DESC 
27                LIMIT $2
28                """,
29                session_id, limit
30            )
31            # 反转以返回时间顺序（最旧 -> 最新）
32            return [{"role": r["role"], "content": r["content"]} for r in reversed(rows)]

src/orchestration.py

1# src/orchestration.py
2from litellm import completion
3
4def rewrite_query(user_query: str, history: list) -> str:
5    """
6    使用快速模型根据历史记录澄清用户的提问。
7    """
8    if not history:
9        return user_query
10
11    # 用于重写的专用系统提示
12    system_prompt = """
13    你是一个查询重写器。 
14    你的任务是重写用户最新的问题，使其在没有聊天历史的情况下也能被理解。
15    将代词（“它”、“那个”）替换为历史记录中所指的具体名词。
16    不要回答问题。只需输出重写后的查询文本。
17    """
18    
19    # 为模型格式化历史记录
20    messages = [{"role": "system", "content": system_prompt}]
21    messages.extend(history)
22    messages.append({"role": "user", "content": user_query})
23
24    response = completion(
25        model="gpt-5-nano", # 此中间步骤使用更便宜/更快的模型
26        messages=messages,
27        temperature=0.0 
28    )
29    
30    return response.choices[0].message.content

src/main.py 文件中的更新部分

1# src/main.py 更新
2
3@app.post("/chat", response_model=ChatResponse)
4async def chat_endpoint(
5    request: ChatRequest, # 请求现在包含 'session_id'
6    # ... deps ...
7):
8    # 1. 加载历史记录（状态填充）
9    chat_memory = ChatHistory(db_pool)
10    history = await chat_memory.get_recent_history(request.session_id)
11    
12    # 2. 重写查询（编排）
13    # 如果存在历史记录，必须消除新查询的歧义
14    search_query = rewrite_query(request.query, history) if history else request.query
15    
16    # 3. 检索（使用重写后的独立查询）
17    candidates = search(search_query, k=20)
18    top_docs = rerank_results(search_query, candidates)[:5]
19    
20    # 4. 生成（使用原始查询 + 历史记录 + 上下文）
21    # 我们将历史记录传递给生成器，以保持“聊天”角色的一致性
22    generated = generate_answer(
23        user_query=request.query, 
24        context_chunks=top_docs,
25        chat_history=history
26    )
27    
28    # 5. 持久化状态（更新内存）
29    # 我们使用 asyncio.gather 并行执行这些数据库写入操作，以减少延迟
30    await asyncio.gather(
31        chat_memory.add_message(request.session_id, "user", request.query),
32        chat_memory.add_message(request.session_id, "assistant", generated['answer'])
33    )
34    
35    return ChatResponse(
36        answer=generated['answer'],
37        # ... 字段
38    )

表格