多轮 RAG：先把追问改写成独立查询，再检索知识库

from collections import defaultdict, deque


histories = defaultdict(lambda: deque(maxlen=6))


def rewrite(question: str, history: list[dict]) -> str:
    pronouns = {"它", "这个设备", "该设备"}
    if not any(word in question for word in pronouns):
        return question
    entities = [
        token.strip("，。？！")
        for message in reversed(history)
        if message["role"] == "user"
        for token in message["content"].split()
        if token.startswith("AX-")
    ]
    if not entities:
        return question
    return question.replace("它", entities[0]).replace("该设备", entities[0])


def independent_query(user_id: str, thread_id: str, question: str) -> str:
    key = (user_id, thread_id)
    query = rewrite(question, list(histories[key]))
    histories[key].append({"role": "user", "content": question})
    return query


print(independent_query("u-17", "t-1", "AX-3 何时离线"))
print(independent_query("u-17", "t-1", "它什么时候恢复"))
print(independent_query("u-18", "t-1", "它什么时候恢复"))

standalone = rewrite_chain.invoke({
    "history": selected_history,
    "question": current_question,
})
docs = retriever.invoke(standalone)
answer = answer_chain.invoke({
    "question": current_question,
    "context": build_context(docs),
})