""" 对话式 Agent 核心模块 融合对话能力的智能 Agent """ import logging from dataclasses import dataclass from typing import Dict, Any, Optional, List from .conversation_state import ConversationState from .intent_understanding import IntentUnderstanding, Intent from .response_generator import ResponseGenerator from ..agent_core import Agent, SkillResult @dataclass class ChatResponse: """对话响应""" message: str # 响应消息 suggestions: Optional[List[str]] = None # 建议的后续操作 status: str = "success" # 状态：success, error, waiting_input metadata: Optional[Dict[str, Any]] = None # 元数据 def __post_init__(self) -> None: """Validate response after initialization""" if not self.message or not self.message.strip(): raise ValueError("message cannot be empty") valid_statuses = ["success", "error", "waiting_input"] if self.status not in valid_statuses: raise ValueError(f"status must be one of {valid_statuses}") if self.suggestions is not None and not isinstance(self.suggestions, list): raise ValueError("suggestions must be a list or None") class ConversationalAgent: """对话式 Agent，融合对话能力的智能 Agent""" def __init__( self, command_agent: Agent, config: Optional[Dict[str, Any]] = None ) -> None: """ 初始化对话式 Agent Args: command_agent: 底层的 Command Agent config: 配置字典 """ self.command_agent: Agent = command_agent self.config: Dict[str, Any] = config or {} self.logger: logging.Logger = logging.getLogger("ConversationalAgent") # 初始化各个模块 self.intent_understanding: IntentUnderstanding = IntentUnderstanding(config) self.conversation_state: ConversationState = ConversationState() self.response_generator: ResponseGenerator = ResponseGenerator(config) async def initialize(self) -> str: """ 初始化 Agent 并返回欢迎消息 Returns: 欢迎消息 """ welcome_msg = await self.response_generator.generate_welcome_message() self.conversation_state.add_message("assistant", welcome_msg) self.logger.info("对话式 Agent 已初始化") return welcome_msg async def chat(self, user_message: str) -> ChatResponse: """ 处理用户消息并返回响应 Args: user_message: 用户的输入消息 Returns: ChatResponse: 对话响应 """ self.logger.info(f"处理用户消息: {user_message}") try: # 1. 记录用户消息 self.conversation_state.add_message("user", user_message) # 2. 理解用户意图 context: Dict[str, Any] = { "conversation_history": self.conversation_state.get_conversation_history() } intent: Intent = await self.intent_understanding.understand( user_message, context ) self.logger.debug(f"识别的意图: {intent.command}") # 3. 如果需要澄清，返回澄清问题 if intent.clarification_needed: response: ChatResponse = ChatResponse( message=intent.clarification_question or "抱歉，我没有理解您的意思。能否请您重新表述？", status="waiting_input", ) self.conversation_state.add_message("assistant", response.message) return response # 4. 映射意图到命令并执行 command_result: SkillResult = await self._execute_command(intent) # 5. 生成响应 if command_result.success: response_msg: str = ( await self.response_generator.generate_success_response( intent.command, command_result.data, user_message ) ) else: response_msg = await self.response_generator.generate_error_response( intent.command, command_result.error or "未知错误", user_message ) # 6. 生成建议 suggestions: List[str] = await self.response_generator.generate_suggestions( self.conversation_state.context ) # 7. 构建响应 response = ChatResponse( message=response_msg, suggestions=suggestions, status="success" if command_result.success else "error", metadata={ "command": intent.command, "confidence": intent.confidence, "reasoning": intent.raw_understanding, }, ) # 8. 记录助手响应 self.conversation_state.add_message("assistant", response_msg) self.logger.info(f"响应生成完成: {response_msg[:50]}...") return response except Exception as e: self.logger.error(f"处理消息失败: {str(e)}", exc_info=True) error_response: ChatResponse = ChatResponse( message="抱歉，处理您的请求时出现了问题。请稍后重试。", status="error" ) self.conversation_state.add_message("assistant", error_response.message) return error_response async def _execute_command(self, intent: Intent) -> SkillResult: """ 执行命令 Args: intent: 识别的意图 Returns: SkillResult: 命令执行结果 """ try: # 开始任务 task = self.conversation_state.start_task(intent.command, intent.parameters) # 执行命令 result: SkillResult = await self.command_agent.execute_command( intent.command, intent.parameters ) # 更新任务状态 if result.success: self.conversation_state.complete_task(result.data) else: self.conversation_state.fail_task(result.error or "未知错误") return result except Exception as e: self.logger.error(f"命令执行失败: {str(e)}") self.conversation_state.fail_task(str(e)) return SkillResult(success=False, error=str(e), message="命令执行失败") def get_conversation_history(self) -> List[Dict[str, str]]: """ 获取对话历史 Returns: 对话历史列表 """ return self.conversation_state.get_conversation_history() def get_state_summary(self) -> Dict[str, Any]: """ 获取对话状态摘要 Returns: 状态摘要 """ return self.conversation_state.get_summary() def clear_history(self) -> None: """ 清除对话历史 """ self.conversation_state.clear_history() self.logger.info("对话历史已清除") def reset(self) -> None: """ 重置对话状态 """ self.conversation_state.reset() self.logger.info("对话状态已重置")