5.2 多轮对话代理

让 Haystack 问答系统具备记忆和上下文理解能力

单轮问答只能回答孤立问题，但真实场景中用户往往需要连续追问。本节将在基础问答系统之上，为 Haystack 管道添加对话记忆、上下文管理和智能路由，构建一个完整的多轮对话代理。

1. 对话记忆系统

对话记忆是多轮对话的基石。我们需要维护一个滑动窗口来保留最近 N 轮对话：

from dataclasses import dataclass, field
from typing import Optional
from collections import deque
import uuid
import json
import time

@dataclass
class Message:
    role: str  # "user" / "assistant" / "system"
    content: str
    timestamp: float = field(default_factory=time.time)
    message_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    meta: dict = field(default_factory=dict)

class ConversationMemory:
    """对话记忆管理器，支持滑动窗口和持久化"""
    
    def __init__(self, max_turns: int = 10, max_tokens: int = 2000):
        self.max_turns = max_turns
        self.max_tokens = max_tokens
        self.messages: deque[Message] = deque(maxlen=max_turns * 2)
        self._token_counter = 0
    
    def add_message(self, role: str, content: str, meta: dict = None):
        msg = Message(role=role, content=content, meta=meta or {})
        self.messages.append(msg)
        self._trim_by_tokens()
    
    def _trim_by_tokens(self):
        """按 token 预算裁剪旧消息"""
        total = sum(len(m.content) for m in self.messages)
        while total > self.max_tokens and len(self.messages) > 2:
            removed = self.messages.popleft()
            total -= len(removed.content)
    
    def get_context(self, last_n: int = 0) -> list[dict]:
        """获取对话上下文，返回字典列表供 LLM 使用"""
        messages = list(self.messages)
        if last_n > 0:
            messages = messages[-last_n * 2:]
        return [{"role": m.role, "content": m.content} for m in messages]
    
    def get_history_text(self) -> str:
        """将对话历史格式化为文本，供检索增强"""
        lines = []
        for msg in self.messages:
            if msg.role == "user":
                lines.append(f"用户：{msg.content}")
            elif msg.role == "assistant":
                lines.append(f"助手：{msg.content}")
        return "\n".join(lines)
    
    def clear(self):
        self.messages.clear()
    
    def export(self) -> dict:
        return {
            "messages": [
                {"role": m.role, "content": m.content, "ts": m.timestamp}
                for m in self.messages
            ]
        }
​

2. 上下文感知的查询重写

当用户说"它的优势是什么"时，"它"指什么？我们需要一个查询重写组件来消除歧义：

from haystack import component, Pipeline, Document
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

@component
class QueryRewriter:
    """利用 LLM 将模糊追问改写为独立、清晰的问题"""
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        self.generator = OpenAIGenerator(model=model, max_tokens=256, temperature=0.0)
        
        self._template = """
你是一个查询重写助手。根据对话历史，将用户当前的问题改写为一个独立、清晰、无歧义的检索查询。

对话历史：
{{ history }}

当前问题：{{ question }}

请只输出改写后的查询，不要解释。
"""
        self._prompt_builder = PromptBuilder(template=self._template)
    
    @component.output_types(rewritten_query=str)
    def run(self, question: str, history: str):
        result = self._prompt_builder.run(question=question, history=history)
        generator_result = self.generator.run(result)
        rewritten = generator_result["replies"][0].strip()
        
        # 如果 LLM 返回空或过于简短，回退到原始问题
        if len(rewritten) < 3:
            return {"rewritten_query": question}
        
        return {"rewritten_query": rewritten}
​

3. 条件路由器

根据用户意图动态选择处理策略——有些问题需要检索，有些只需要闲聊：

@component
class IntentRouter:
    """意图分类与路由：决定当前问题需要走检索还是直接对话"""
    
    INTENT_RETRIEVAL = "retrieval"
    INTENT_CHITCHAT = "chitchat"
    INTENT_CLARIFY = "clarify"
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        self.generator = OpenAIGenerator(model=model, max_tokens=32, temperature=0.0)
        
        self._template = """
判断用户问题的意图，只输出以下三种之一：retrieval、chitchat、clarify

- retrieval：需要检索知识库回答的事实性问题
- chitchat：寒暄、闲聊、感谢等不需要检索的对话
- clarify：问题不清晰，需要追问确认

用户问题：{{ question }}
对话历史：{{ history }}

意图：
"""
        self._prompt_builder = PromptBuilder(template=self._template)
    
    @component.output_types(intent=str)
    def run(self, question: str, history: str):
        # 快速关键词检测，减少不必要的 LLM 调用
        chitchat_keywords = ["谢谢", "再见", "你好", "哈哈", "ok", "嗯"]
        for kw in chitchat_keywords:
            if question.strip() == kw or question.strip().startswith(kw):
                return {"intent": self.INTENT_CHITCHAT}
        
        result = self._prompt_builder.run(question=question, history=history)
        gen_result = self.generator.run(result)
        intent = gen_result["replies"][0].strip().lower()
        
        # 验证意图合法性
        if intent not in [self.INTENT_RETRIEVAL, self.INTENT_CHITCHAT, self.INTENT_CLARIFY]:
            intent = self.INTENT_RETRIEVAL  # 默认走检索
        
        return {"intent": intent}
​

4. 组装多轮对话管道

将所有组件组装成一个完整的多轮对话管道：

def build_conversational_pipeline(
    document_store,
    embedding_model: str,
    llm_model: str = "gpt-3.5-turbo"
):
    """构建多轮对话 RAG 管道"""
    pipeline = Pipeline()
    
    # 组件注册
    pipeline.add_component("query_rewriter", QueryRewriter(model=llm_model))
    pipeline.add_component("intent_router", IntentRouter(model=llm_model))
    pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder(model=embedding_model))
    pipeline.add_component(
        "retriever",
        InMemoryEmbeddingRetriever(document_store=document_store, top_k=5)
    )
    
    # RAG 提示模板
    rag_prompt = """
你是一个专业的AI助手。请根据参考信息和对话历史回答问题。

对话历史：
{{ history }}

参考信息：
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

用户问题：{{ question }}

回答要求：
1. 优先使用参考信息中的内容
2. 如果参考信息不足，可以结合对话历史和自身知识
3. 如果完全无法回答，诚实告知用户
4. 保持回答简洁、有条理
"""
    pipeline.add_component("rag_prompt_builder", PromptBuilder(template=rag_prompt))
    pipeline.add_component("rag_generator", OpenAIGenerator(model=llm_model, max_tokens=512))
    
    # 闲聊模板
    chat_prompt = "你是一个友好的AI助手，正在和用户进行多轮对话。\n\n对话历史：{{ history }}\n\n用户：{{ question }}\n\n请用自然、简洁的方式回复。"
    pipeline.add_component("chat_prompt_builder", PromptBuilder(template=chat_prompt))
    pipeline.add_component("chat_generator", OpenAIGenerator(model=llm_model, max_tokens=256, temperature=0.7))
    
    # 连接管道
    pipeline.connect("query_rewriter.rewritten_query", "text_embedder.text")
    pipeline.connect("text_embedder", "retriever")
    pipeline.connect("retriever.documents", "rag_prompt_builder.documents")
    pipeline.connect("rag_prompt_builder", "rag_generator")
    pipeline.connect("chat_prompt_builder", "chat_generator")
    
    return pipeline
​

5. 对话代理控制器

统一管理对话流程和记忆：

class ConversationalAgent:
    """多轮对话代理"""
    
    def __init__(self, pipeline, memory: ConversationMemory):
        self.pipeline = pipeline
        self.memory = memory
    
    def chat(self, user_input: str) -> dict:
        """处理一轮用户输入，返回回复"""
        self.memory.add_message("user", user_input)
        history_text = self.memory.get_history_text()
        
        # 意图识别
        intent_result = self.pipeline.run(
            intent_router={"question": user_input, "history": history_text},
        )
        intent = intent_result["intent_router"]["intent"]
        
        if intent == IntentRouter.INTENT_CLARIFY:
            reply = "您的问题有些模糊，能否更详细描述一下？例如说明具体的技术领域或使用场景。"
        
        elif intent == IntentRouter.INTENT_CHITCHAT:
            chat_result = self.pipeline.run(
                chat_prompt_builder={"question": user_input, "history": history_text}
            )
            reply = chat_result["chat_generator"]["replies"][0]
        
        else:  # retrieval
            # 先重写查询，再走 RAG
            rewrite_result = self.pipeline.run(
                query_rewriter={"question": user_input, "history": history_text}
            )
            rewritten = rewrite_result["query_rewriter"]["rewritten_query"]
            
            rag_result = self.pipeline.run(
                text_embedder={"text": rewritten},
                rag_prompt_builder={
                    "question": user_input,
                    "history": history_text
                }
            )
            reply = rag_result["rag_generator"]["replies"][0]
        
        self.memory.add_message("assistant", reply)
        
        return {
            "reply": reply,
            "intent": intent,
            "rewritten_query": rewrite_result["query_rewriter"]["rewritten_query"] if intent == "retrieval" else None,
            "history_length": len(self.memory.messages)
        }
    
    def reset(self):
        """重置对话"""
        self.memory.clear()
​

6. 运行示例

# 初始化
memory = ConversationMemory(max_turns=10)
pipeline = build_conversational_pipeline(document_store, "sentence-transformers/all-MiniLM-L6-v2")
agent = ConversationalAgent(pipeline, memory)

# 模拟多轮对话
conversation = [
    "Haystack 是什么框架？",
    "它的主要组件有哪些？",           # 追问，需要上下文
    "和 LangChain 比有什么优势？",     # 对比类追问
    "谢谢你的解答！再见"              # 闲聊
]

for user_msg in conversation:
    print(f"\n👤 用户: {user_msg}")
    result = agent.chat(user_msg)
    print(f"🤖 代理 [{result['intent']}]: {result['reply']}")
    if result.get("rewritten_query"):
        print(f"   (查询重写: {result['rewritten_query']})")
​

7. 对话质量优化技巧

Token 预算管理：当对话历史过长时，采用摘要压缩策略而非简单截断：

@component
class HistorySummarizer:
    """对话历史摘要器，将长历史压缩为紧凑摘要"""
    
    def __init__(self, model: str = "gpt-3.5-turbo", max_history_turns: int = 6):
        self.generator = OpenAIGenerator(model=model, max_tokens=256, temperature=0.0)
        self.max_history_turns = max_history_turns
        
        self._template = """
请将以下对话历史压缩为一段简洁的摘要，保留关键事实和上下文。

对话历史：
{{ history }}

摘要：
"""
        self._prompt_builder = PromptBuilder(template=self._template)
    
    @component.output_types(compressed_history=str)
    def run(self, history: str):
        if len(history) < 500:  # 短历史不需要压缩
            return {"compressed_history": history}
        
        result = self._prompt_builder.run(history=history)
        gen_result = self.generator.run(result)
        return {"compressed_history": gen_result["replies"][0]}
​

错误恢复机制：当 LLM 生成失败或超时时，提供优雅降级：

class ConversationalAgent:
    # ... 在 chat 方法中添加异常处理
    def chat(self, user_input: str) -> dict:
        try:
            # ... 正常流程
        except Exception as e:
            fallback_replies = {
                "timeout": "抱歉，响应超时了，请再试一次。",
                "rate_limit": "当前请求较多，请稍后重试。",
                "default": "处理您的请求时遇到了问题，请换个方式提问。"
            }
            error_type = "timeout" if "timeout" in str(e).lower() else \
                         "rate_limit" if "rate" in str(e).lower() else "default"
            return {
                "reply": fallback_replies[error_type],
                "intent": "error",
                "error": str(e)
            }
​

架构总览

本节实现了一个具备对话记忆、上下文理解、意图路由和查询重写能力的多轮对话代理。这套架构可以灵活扩展，后续可以加入用户画像、个性化推荐等更高级的功能。