""" 同步版本的React Agent - 解决Vector搜索异步冲突问题 基于原有CustomReactAgent，但使用完全同步的实现 """ import json import sys import os from pathlib import Path from typing import List, Optional, Dict, Any import redis # 添加项目根目录到sys.path try: project_root = Path(__file__).parent.parent if str(project_root) not in sys.path: sys.path.insert(0, str(project_root)) except Exception as e: pass from core.logging import get_react_agent_logger from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, ToolMessage, BaseMessage, SystemMessage, AIMessage from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode # 导入同步版本的依赖 try: from . import config from .state import AgentState from .sql_tools import sql_tools except ImportError: import config from state import AgentState from sql_tools import sql_tools logger = get_react_agent_logger("SyncCustomReactAgent") class SyncCustomReactAgent: """ 同步版本的React Agent 专门解决Vector搜索的异步事件循环冲突问题 """ def __init__(self): """私有构造函数，请使用 create() 类方法来创建实例。""" self.llm = None self.tools = None self.agent_executor = None self.checkpointer = None self.redis_client = None @classmethod def create(cls): """同步工厂方法，创建并初始化 SyncCustomReactAgent 实例。""" instance = cls() instance._sync_init() return instance def _sync_init(self): """同步初始化所有组件。""" logger.info("🚀 开始初始化 SyncCustomReactAgent...") # 1. 初始化同步Redis客户端（如果需要） try: self.redis_client = redis.from_url(config.REDIS_URL, decode_responses=True) self.redis_client.ping() logger.info(f" ✅ Redis连接成功: {config.REDIS_URL}") except Exception as e: logger.warning(f" ⚠️ Redis连接失败，将不使用checkpointer: {e}") self.redis_client = None # 2. 初始化 LLM（同步版本） self.llm = ChatOpenAI( api_key=config.QWEN_API_KEY, base_url=config.QWEN_BASE_URL, model=config.QWEN_MODEL, temperature=0.1, timeout=config.NETWORK_TIMEOUT, max_retries=0, streaming=False, # 关键：禁用流式处理 extra_body={ "enable_thinking": False, # 明确设置为False：非流式调用必须设为false "misc": { "ensure_ascii": False } } ) logger.info(f" ✅ 同步LLM已初始化，模型: {config.QWEN_MODEL}") # 3. 绑定工具 self.tools = sql_tools self.llm_with_tools = self.llm.bind_tools(self.tools) logger.info(f" ✅ 已绑定 {len(self.tools)} 个工具") # 4. 创建StateGraph（不使用checkpointer避免异步依赖） self.agent_executor = self._create_sync_graph() logger.info(" ✅ 同步StateGraph已创建") logger.info("✅ SyncCustomReactAgent 初始化完成") def _create_sync_graph(self): """创建同步的StateGraph""" graph = StateGraph(AgentState) # 添加同步节点 graph.add_node("agent", self._sync_agent_node) graph.add_node("tools", ToolNode(self.tools)) graph.add_node("prepare_tool_input", self._sync_prepare_tool_input_node) graph.add_node("update_state_after_tool", self._sync_update_state_after_tool_node) graph.add_node("format_final_response", self._sync_format_final_response_node) # 设置入口点 graph.set_entry_point("agent") # 添加条件边 graph.add_conditional_edges( "agent", self._sync_should_continue, { "tools": "prepare_tool_input", "end": "format_final_response" } ) # 添加普通边 graph.add_edge("prepare_tool_input", "tools") graph.add_edge("tools", "update_state_after_tool") graph.add_edge("update_state_after_tool", "agent") graph.add_edge("format_final_response", END) # 关键：使用同步编译，不传入checkpointer return graph.compile() def _sync_agent_node(self, state: AgentState) -> Dict[str, Any]: """同步Agent节点""" logger.info(f"🧠 [Sync Node] agent - Thread: {state.get('thread_id', 'unknown')}") messages_for_llm = state["messages"].copy() # 添加数据库范围提示词 if isinstance(state["messages"][-1], HumanMessage): db_scope_prompt = self._get_database_scope_prompt() if db_scope_prompt: messages_for_llm.insert(0, SystemMessage(content=db_scope_prompt)) logger.info(" ✅ 已添加数据库范围判断提示词") # 同步LLM调用 response = self.llm_with_tools.invoke(messages_for_llm) return {"messages": [response]} def _sync_should_continue(self, state: AgentState): """同步条件判断""" messages = state["messages"] last_message = messages[-1] if not last_message.tool_calls: return "end" else: return "tools" def _sync_prepare_tool_input_node(self, state: AgentState) -> Dict[str, Any]: """同步准备工具输入节点""" logger.info(f"🔧 [Sync Node] prepare_tool_input - Thread: {state.get('thread_id', 'unknown')}") last_message = state["messages"][-1] if hasattr(last_message, 'tool_calls') and last_message.tool_calls: for tool_call in last_message.tool_calls: if tool_call.get('name') == 'generate_sql': # 注入历史消息 history_messages = self._filter_and_format_history(state["messages"]) if 'args' not in tool_call: tool_call['args'] = {} tool_call['args']['history_messages'] = history_messages logger.info(f" ✅ 为generate_sql注入了 {len(history_messages)} 条历史消息") return {"messages": [last_message]} def _sync_update_state_after_tool_node(self, state: AgentState) -> Dict[str, Any]: """同步更新工具执行后的状态""" logger.info(f"📝 [Sync Node] update_state_after_tool - Thread: {state.get('thread_id', 'unknown')}") last_message = state["messages"][-1] tool_name = last_message.name tool_output = last_message.content next_step = None if tool_name == 'generate_sql': tool_output_lower = tool_output.lower() if "failed" in tool_output_lower or "无法生成" in tool_output_lower or "失败" in tool_output_lower: next_step = 'answer_with_common_sense' else: next_step = 'valid_sql' elif tool_name == 'valid_sql': if "失败" in tool_output: next_step = 'analyze_validation_error' else: next_step = 'run_sql' elif tool_name == 'run_sql': next_step = 'summarize_final_answer' logger.info(f" Tool '{tool_name}' executed. Suggested next step: {next_step}") return {"suggested_next_step": next_step} def _sync_format_final_response_node(self, state: AgentState) -> Dict[str, Any]: """同步格式化最终响应节点""" logger.info(f"📄 [Sync Node] format_final_response - Thread: {state.get('thread_id', 'unknown')}") messages = state["messages"] last_message = messages[-1] # 构建最终响应 final_response = last_message.content logger.info(f" ✅ 最终响应已准备完成") return {"final_answer": final_response} def _filter_and_format_history(self, messages: list) -> list: """过滤和格式化历史消息""" clean_history = [] for msg in messages[:-1]: # 排除最后一条消息 if isinstance(msg, HumanMessage): clean_history.append({"type": "human", "content": msg.content}) elif isinstance(msg, AIMessage): clean_content = msg.content if not hasattr(msg, 'tool_calls') or not msg.tool_calls else "" if clean_content.strip(): clean_history.append({"type": "ai", "content": clean_content}) return clean_history def _get_database_scope_prompt(self) -> str: """获取数据库范围判断提示词""" return """你是一个专门处理高速公路收费数据查询的AI助手。在回答用户问题时，请首先判断这个问题是否可以通过查询数据库来回答。 数据库包含以下类型的数据： - 服务区信息（名称、位置、档口数量等） - 收费站数据 - 车流量统计 - 业务数据分析 如果用户的问题与这些数据相关，请使用工具生成SQL查询。 如果问题与数据库内容无关（如常识性问题、天气、新闻等），请直接用你的知识回答，不要尝试生成SQL。""" def chat(self, message: str, user_id: str, thread_id: Optional[str] = None) -> Dict[str, Any]: """ 同步聊天方法 - 关键：使用 graph.invoke() 而不是 ainvoke() """ if thread_id is None: import uuid thread_id = str(uuid.uuid4()) # 构建输入 inputs = { "messages": [HumanMessage(content=message)], "user_id": user_id, "thread_id": thread_id, "suggested_next_step": None } # 构建运行配置（不使用checkpointer） run_config = { "recursion_limit": config.RECURSION_LIMIT, } try: logger.info(f"🚀 开始同步处理用户消息: {message[:50]}...") # 关键：使用同步的 invoke() 方法 final_state = self.agent_executor.invoke(inputs, run_config) logger.info(f"🔍 Final state keys: {list(final_state.keys())}") # 提取答案 if final_state["messages"]: answer = final_state["messages"][-1].content else: answer = "抱歉，无法处理您的请求。" # 提取SQL数据（如果有） sql_data = self._extract_latest_sql_data(final_state["messages"]) logger.info(f"✅ 同步处理完成 - Final Answer: '{answer[:100]}...'") # 构建返回结果 result = { "success": True, "answer": answer, "thread_id": thread_id } # 只有当存在SQL数据时才添加到返回结果中 if sql_data: try: # 尝试解析SQL数据 sql_parsed = json.loads(sql_data) # 检查数据格式：run_sql工具返回的是数组格式 [{"col1":"val1"}] if isinstance(sql_parsed, list): # 数组格式：直接作为records使用 result["api_data"] = { "response": answer, "records": sql_parsed, "react_agent_meta": { "thread_id": thread_id, "agent_version": "sync_react_v1" } } elif isinstance(sql_parsed, dict): # 字典格式：按原逻辑处理 result["api_data"] = { "response": answer, "sql": sql_parsed.get("sql", ""), "records": sql_parsed.get("records", []), "react_agent_meta": { "thread_id": thread_id, "agent_version": "sync_react_v1" } } else: logger.warning(f"SQL数据格式未知: {type(sql_parsed)}") raise ValueError("Unknown SQL data format") except (json.JSONDecodeError, AttributeError, ValueError) as e: logger.warning(f"SQL数据格式处理失败: {str(e)}, 跳过API数据构建") else: result["api_data"] = { "response": answer, "react_agent_meta": { "thread_id": thread_id, "agent_version": "sync_react_v1" } } return result except Exception as e: logger.error(f"❌ 同步处理失败: {str(e)}", exc_info=True) return { "success": False, "error": f"同步处理失败: {str(e)}", "thread_id": thread_id, "retry_suggested": True } def _extract_latest_sql_data(self, messages: List[BaseMessage]) -> Optional[str]: """从消息历史中提取最近的run_sql执行结果（同步版本）""" logger.info("🔍 提取最新的SQL执行结果...") # 查找最后一个HumanMessage之后的SQL执行结果 last_human_index = -1 for i in range(len(messages) - 1, -1, -1): if isinstance(messages[i], HumanMessage): last_human_index = i break if last_human_index == -1: logger.info(" 未找到用户消息，跳过SQL数据提取") return None # 只在当前对话轮次中查找SQL结果 current_conversation = messages[last_human_index:] logger.info(f" 当前对话轮次包含 {len(current_conversation)} 条消息") for msg in reversed(current_conversation): if isinstance(msg, ToolMessage) and msg.name == 'run_sql': logger.info(f" 找到当前对话轮次的run_sql结果: {msg.content[:100]}...") try: # 尝试解析JSON以验证格式 parsed_data = json.loads(msg.content) # 重新序列化，确保中文字符正常显示 formatted_content = json.dumps(parsed_data, ensure_ascii=False, separators=(',', ':')) logger.info(f" 已转换Unicode转义序列为中文字符") return formatted_content except json.JSONDecodeError: # 如果不是有效JSON，直接返回原内容 logger.warning(f" SQL结果不是有效JSON格式，返回原始内容") return msg.content logger.info(" 当前对话轮次中未找到run_sql执行结果") return None