5.2 多轮对话代理 — 构建上下文感知的智能对话系统

本节导读：掌握基于Haystack的多轮对话代理技术，实现上下文感知、多轮推理、工具调用的复杂对话能力，构建企业级对话系统架构。

学习目标

• 理解多轮对话代理的核心架构和技术要点
• 实现基于Haystack的上下文管理机制
• 掌握多轮推理和工具调用的技术实现
• 学会构建企业级对话系统的完整方案
• 实现对话质量评估和优化策略

核心概念

多轮对话代理是RAG技术的重要应用场景，通过Haystack框架可以快速构建具有上下文感知能力的智能对话系统。以下是核心组件和工作流程：

![Haystack多轮对话代理架构图：包含上下文管理器、对话状态机、工具调用系统、多轮推理引擎等关键组件，展示从用户输入到智能回答的完整流程]

系统架构层次

# 多轮对话代理架构设计
dialog_agent_architecture = {
    "对话管理层": {
        "对话状态管理": "Dialog State Tracker",
        "上下文窗口管理": "Context Window Manager",
        "对话流程控制": "Dialog Flow Controller"
    },
    "推理层": {
        "多轮推理引擎": "Multi-turn Reasoning Engine",
        "工具调用系统": "Tool Calling System", 
        "知识检索增强": "Knowledge Retrieval Enhancement"
    },
    "感知层": {
        "用户意图识别": "Intent Recognition",
        "实体提取": "Entity Extraction",
        "情感分析": "Sentiment Analysis"
    },
    "执行层": {
        "响应生成": "Response Generation",
        "动作执行": "Action Execution",
        "反馈收集": "Feedback Collection"
    }
}
​

关键技术点

1. 上下文管理：维护对话历史和状态信息
2. 多轮推理：基于上下文进行复杂推理和问题分解
3. 工具调用：集成外部API和知识库检索
4. 对话流控制：管理对话进程和分支逻辑

环境准备 / 前置知识

开发环境配置

# 创建项目目录
mkdir haystack-dialog-agent
cd haystack-dialog-agent
python -m venv venv
source venv/bin/activate

# 安装核心依赖
pip install haystack-ai[weaviate,chroma,transformers]
pip install openai>=1.0.0
pip install torch>=2.0.0
pip install sentence-transformers
pip install fastapi uvicorn
pip install python-multipart

# 对话相关依赖
pip install transformers>=4.30.0
pip install langchain>=0.1.0
pip install redis>=4.5.0

# 开发工具
pip install jupyterlab ipython black isort mypy
​

项目结构设计

haystack-dialog-agent/
├── src/
│   ├── __init__.py
│   ├── config.py           # 配置管理
│   ├── dialog_manager.py   # 对话管理
│   ├── context_manager.py  # 上下文管理
│   ├── reasoning_engine.py # 多轮推理
│   ├── tool_system.py      # 工具调用系统
│   ├── dialog_pipeline.py # 对话管道
│   ├── api_server.py       # API服务
│   └── utils.py           # 工具函数
├── data/
│   ├── documents/         # 对话知识库
│   ├── context/          # 对话上下文
│   ├── tools/            # 工具配置
│   └── logs/            # 日志文件
├── tests/
│   ├── test_dialog_manager.py
│   ├── test_context_manager.py
│   ├── test_reasoning_engine.py
│   └── test_tool_system.py
├── requirements.txt
├── README.md
└── docker/
    ├── Dockerfile
    └── docker-compose.yml
​

分步实战

步骤1：对话管理器实现

# src/dialog_manager.py
"""
对话管理模块
负责对话状态管理、流程控制和上下文维护
"""
import json
import logging
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import uuid

from dataclasses import dataclass
from typing import List, Dict, Any, Optional

class DialogState(Enum):
    """对话状态枚举"""
    INIT = "init"
    PROCESSING = "processing"
    WAITING = "waiting"
    ANSWERING = "answering"
    TOOL_CALLING = "tool_calling"
    CONFIRMING = "confirming"
    COMPLETED = "completed"
    ERROR = "error"

@dataclass
class DialogMessage:
    """对话消息"""
    id: str
    role: str  # "user" or "assistant"
    content: str
    timestamp: float
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class DialogContext:
    """对话上下文"""
    session_id: str
    user_id: Optional[str] = None
    state: DialogState = DialogState.INIT
    messages: List[DialogMessage] = field(default_factory=list)
    current_intent: Optional[str] = None
    entities: Dict[str, Any] = field(default_factory=dict)
    tools_used: List[str] = field(default_factory=list)
    session_metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

class DialogManager:
    """对话管理器"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.contexts: Dict[str, DialogContext] = {}
        self.max_history = config.get("max_history", 10)
        self.max_context_length = config.get("max_context_length", 2000)
        
    def create_session(self, user_id: Optional[str] = None) -> str:
        """创建新对话会话"""
        session_id = str(uuid.uuid4())
        context = DialogContext(
            session_id=session_id,
            user_id=user_id,
            state=DialogState.INIT
        )
        
        self.contexts[session_id] = context
        self.logger.info(f"创建新会话: {session_id}")
        
        return session_id
    
    def get_context(self, session_id: str) -> Optional[DialogContext]:
        """获取对话上下文"""
        return self.contexts.get(session_id)
    
    def add_message(self, session_id: str, role: str, content: str, 
                   metadata: Dict[str, Any] = None) -> bool:
        """添加消息到对话历史"""
        context = self.get_context(session_id)
        if not context:
            self.logger.error(f"会话不存在: {session_id}")
            return False
        
        message = DialogMessage(
            id=str(uuid.uuid4()),
            role=role,
            content=content,
            timestamp=time.time(),
            metadata=metadata or {}
        )
        
        # 添加消息到历史
        context.messages.append(message)
        context.updated_at = time.time()
        
        # 维护历史长度限制
        if len(context.messages) > self.max_history:
            context.messages = context.messages[-self.max_history:]
        
        # 更新上下文长度
        self._enforce_context_length(context)
        
        self.logger.debug(f"添加消息到会话 {session_id}: {role} - {content[:50]}...")
        return True
    
    def update_state(self, session_id: str, new_state: DialogState, 
                    metadata: Dict[str, Any] = None) -> bool:
        """更新对话状态"""
        context = self.get_context(session_id)
        if not context:
            return False
        
        old_state = context.state
        context.state = new_state
        context.updated_at = time.time()
        
        if metadata:
            context.session_metadata.update(metadata)
        
        self.logger.info(f"会话 {session_id} 状态变更: {old_state.value} -> {new_state.value}")
        return True
    
    def update_intent(self, session_id: str, intent: str, entities: Dict[str, Any] = None) -> bool:
        """更新用户意图"""
        context = self.get_context(session_id)
        if not context:
            return False
        
        context.current_intent = intent
        if entities:
            context.entities.update(entities)
        
        context.updated_at = time.time()
        self.logger.info(f"更新会话 {session_id} 意图: {intent}")
        return True
    
    def add_tool_usage(self, session_id: str, tool_name: str, tool_result: Any) -> bool:
        """记录工具使用"""
        context = self.get_context(session_id)
        if not context:
            return False
        
        context.tools_used.append(tool_name)
        context.updated_at = time.time()
        
        # 添加工具结果到元数据
        context.session_metadata[f"tool_{tool_name}"] = {
            "used_at": time.time(),
            "result": tool_result
        }
        
        self.logger.info(f"会话 {session_id} 使用工具: {tool_name}")
        return True
    
    def get_conversation_history(self, session_id: str, limit: int = None) -> List[Dict[str, Any]]:
        """获取对话历史"""
        context = self.get_context(session_id)
        if not context:
            return []
        
        messages = context.messages
        if limit:
            messages = messages[-limit:]
        
        return [
            {
                "id": msg.id,
                "role": msg.role,
                "content": msg.content,
                "timestamp": msg.timestamp,
                "metadata": msg.metadata
            }
            for msg in messages
        ]
    
    def get_current_context(self, session_id: str) -> Dict[str, Any]:
        """获取当前上下文信息"""
        context = self.get_context(session_id)
        if not context:
            return {}
        
        return {
            "session_id": context.session_id,
            "user_id": context.user_id,
            "state": context.state.value,
            "current_intent": context.current_intent,
            "entities": context.entities,
            "tools_used": context.tools_used,
            "message_count": len(context.messages),
            "created_at": context.created_at,
            "updated_at": context.updated_at,
            "session_metadata": context.session_metadata
        }
    
    def _enforce_context_length(self, context: DialogContext):
        """强制执行上下文长度限制"""
        total_length = sum(len(msg.content) for msg in context.messages)
        
        while total_length > self.max_context_length and len(context.messages) > 1:
            removed_msg = context.messages.pop(0)
            total_length -= len(removed_msg.content)
            self.logger.debug(f"移除消息以维持上下文长度: {removed_msg.id}")
    
    def reset_session(self, session_id: str) -> bool:
        """重置对话会话"""
        context = self.get_context(session_id)
        if not context:
            return False
        
        context.state = DialogState.INIT
        context.current_intent = None
        context.entities = {}
        context.tools_used = []
        context.session_metadata = {}
        context.updated_at = time.time()
        
        # 保留系统消息
        user_messages = [msg for msg in context.messages if msg.role == "user"]
        system_messages = [msg for msg in context.messages if msg.role == "assistant" and msg.metadata.get("system")]
        
        context.messages = system_messages + user_messages[-1:] if user_messages else system_messages
        
        self.logger.info(f"重置会话: {session_id}")
        return True
    
    def get_session_stats(self) -> Dict[str, Any]:
        """获取会话统计信息"""
        active_sessions = len(self.contexts)
        total_messages = sum(len(ctx.messages) for ctx in self.contexts.values())
        
        state_distribution = {}
        for context in self.contexts.values():
            state = context.state.value
            state_distribution[state] = state_distribution.get(state, 0) + 1
        
        return {
            "active_sessions": active_sessions,
            "total_messages": total_messages,
            "average_messages_per_session": total_messages / active_sessions if active_sessions > 0 else 0,
            "state_distribution": state_distribution
        }

# 使用示例
if __name__ == "__main__":
    from config import QAConfig
    
    config = QAConfig()
    dialog_manager = DialogManager(config.get("dialog", {}))
    
    # 创建会话
    session_id = dialog_manager.create_session("user123")
    print(f"创建会话: {session_id}")
    
    # 添加消息
    dialog_manager.add_message(session_id, "user", "你好，我想了解 Haystack 框架")
    dialog_manager.add_message(session_id, "assistant", "您好！我是Haystack助手，有什么可以帮您的吗？")
    
    # 更新意图
    dialog_manager.update_intent(session_id, "haystack_inquiry", {"framework": "Haystack"})
    
    # 获取对话历史
    history = dialog_manager.get_conversation_history(session_id)
    print(f"对话历史: {history}")
    
    # 获取当前上下文
    context = dialog_manager.get_current_context(session_id)
    print(f"当前上下文: {context}")
​

步骤2：上下文管理器实现

# src/context_manager.py
"""
上下文管理模块
负责对话上下文的增强管理和语义理解
"""
import logging
import time
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
import numpy as np
from sentence_transformders import SentenceTransformer

@dataclass
class ContextWindow:
    """上下文窗口"""
    messages: List[Dict[str, Any]]
    embeddings: List[np.ndarray] = field(default_factory=list)
    key_entities: Dict[str, Any] = field(default_factory=dict)
    topics: List[str] = field(default_factory=list)
    sentiment_trend: List[float] = field(default_factory=list)
    
@dataclass
class ContextEnhancement:
    """上下文增强信息"""
    semantic_similarity: float
    key_entities_extracted: Dict[str, Any]
    topic_clustering: List[str]
    sentiment_analysis: Dict[str, float]
    intent_tracking: Dict[str, float]
    knowledge_relevance: Dict[str, float]

class ContextManager:
    """上下文管理器"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.embedding_model = SentenceTransformer(config.get("embedding_model", "all-MiniLM-L6-v2"))
        self.max_window_size = config.get("max_window_size", 10)
        self.similarity_threshold = config.get("similarity_threshold", 0.7)
        self.entity_extraction_rules = config.get("entity_extraction_rules", {})
        
    def enhance_context(self, messages: List[Dict[str, Any]]) -> ContextEnhancement:
        """增强上下文信息"""
        if not messages:
            return ContextEnhancement(
                semantic_similarity=0.0,
                key_entities_extracted={},
                topic_clustering=[],
                sentiment_analysis={},
                intent_tracking={},
                knowledge_relevance={}
            )
        
        # 提取关键实体
        key_entities = self._extract_key_entities(messages)
        
        # 主题聚类
        topics = self._cluster_topics(messages)
        
        # 情感分析
        sentiment = self._analyze_sentiment(messages)
        
        # 意图跟踪
        intent = self._track_intent(messages)
        
        # 知识相关性
        relevance = self._calculate_knowledge_relevance(messages)
        
        # 语义相似度
        similarity = self._calculate_semantic_similarity(messages)
        
        return ContextEnhancement(
            semantic_similarity=similarity,
            key_entities_extracted=key_entities,
            topic_clustering=topics,
            sentiment_analysis=sentiment,
            intent_tracking=intent,
            knowledge_relevance=relevance
        )
    
    def _extract_key_entities(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
        """提取关键实体"""
        entities = {}
        
        for message in messages:
            content = message.get("content", "")
            role = message.get("role", "")
            
            # 简单的实体提取规则
            if "Haystack" in content:
                entities["haystack"] = entities.get("haystack", 0) + 1
            if "RAG" in content:
                entities["rag"] = entities.get("rag", 0) + 1
            if "document" in content:
                entities["document"] = entities.get("document", 0) + 1
            if "retrieval" in content:
                entities["retrieval"] = entities.get("retrieval", 0) + 1
        
        return {k: v for k, v in entities.items() if v > 0}
    
    def _cluster_topics(self, messages: List[Dict[str, Any]]) -> List[str]:
        """主题聚类"""
        content = " ".join([msg.get("content", "") for msg in messages])
        topics = []
        
        # 简单的主题识别
        if any(keyword in content for keyword in ["haystack", "rag", "retrieval"]):
            topics.append("RAG技术")
        if any(keyword in content for keyword in ["document", "text", "content"]):
            topics.append("文档处理")
        if any(keyword in content for keyword in ["conversation", "dialog", "chat"]):
            topics.append("对话系统")
        
        return topics
    
    def _analyze_sentiment(self, messages: List[Dict[str, Any]]) -> Dict[str, float]:
        """情感分析"""
        positive_words = ["好", "棒", "优秀", "满意", "喜欢", "赞"]
        negative_words = ["不好", "差", "糟糕", "失望", "讨厌", "问题"]
        
        sentiment_scores = []
        
        for message in messages:
            content = message.get("content", "")
            role = message.get("role", "")
            
            if role == "user":
                positive_count = sum(1 for word in positive_words if word in content)
                negative_count = sum(1 for word in negative_words if word in content)
                
                if positive_count > negative_count:
                    sentiment_scores.append(0.7)
                elif negative_count > positive_count:
                    sentiment_scores.append(-0.7)
                else:
                    sentiment_scores.append(0.0)
        
        avg_sentiment = np.mean(sentiment_scores) if sentiment_scores else 0.0
        
        return {
            "positive": max(0, avg_sentiment),
            "negative": max(0, -avg_sentiment),
            "neutral": 1.0 - abs(avg_sentiment)
        }
    
    def _track_intent(self, messages: List[Dict[str, Any]]) -> Dict[str, float]:
        """意图跟踪"""
        intent_keywords = {
            "inquiry": ["问", "了解", "知道", "介绍", "说明"],
            "help": ["帮助", "解决", "修复", "处理"],
            "comparison": ["对比", "比较", "区别", "优势"],
            "implementation": ["实现", "部署", "安装", "配置"],
            "troubleshooting": ["问题", "错误", "失败", "故障"]
        }
        
        content = " ".join([msg.get("content", "") for msg in messages])
        intent_scores = {}
        
        for intent, keywords in intent_keywords.items():
            score = sum(1 for keyword in keywords if keyword in content) / len(keywords)
            intent_scores[intent] = score
        
        return intent_scores
    
    def _calculate_knowledge_relevance(self, messages: List[Dict[str, Any]]) -> Dict[str, float]:
        """计算知识相关性"""
        content = " ".join([msg.get("content", "") for msg in messages])
        
        knowledge_areas = {
            "haystack": ["haystack", "框架", "pipeline", "document"],
            "rag": ["rag", "检索增强", "retrieval", "增强"],
            "ml": ["机器学习", "模型", "算法", "训练"],
            "nlp": ["自然语言", "文本", "语言", "处理"]
        }
        
        relevance_scores = {}
        
        for area, keywo
​