from typing import Dict, Any, List, Optional from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, END from .base_agent import BaseAgent, AgentState class ConversationAgent(BaseAgent): """处理多轮对话的代理""" def __init__(self, model_section: Optional[str] = None): super().__init__(model_section) self.conversation_history: List[BaseMessage] = [] def _build_graph(self) -> StateGraph: """构建对话专用图""" workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("analyze_intent", self._analyze_intent) workflow.add_node("generate_response", self._generate_response) workflow.add_node("update_context", self._update_context) # 定义边 workflow.add_edge("analyze_intent", "generate_response") workflow.add_edge("generate_response", "update_context") workflow.add_edge("update_context", END) # 设置入口节点 workflow.set_entry_point("analyze_intent") return workflow.compile() def _analyze_intent(self, state: AgentState) -> AgentState: """分析用户意图与对话上下文""" # 简单意图分析,可用更复杂逻辑增强 user_message = state.messages[-1] if state.messages else None if user_message and isinstance(user_message, HumanMessage): content = user_message.content.lower() # 基础意图识别 if any(word in content for word in ["hello", "hi", "hey", "greetings"]): state.context["intent"] = "greeting" elif any(word in content for word in ["help", "assist", "support"]): state.context["intent"] = "help" elif "?" in content: state.context["intent"] = "question" else: """兼容导出:请优先使用 agent 包""" from agent.conversation import ConversationAgent __all__ = ["ConversationAgent"]