一、ReAct

        ReAct = Reasoning（推理）+ Acting（行动）

        这个循环会一直进行，直到AI认为”信息够了，我可以回答用户了"，才会跳出循环给出最终答案。但是以前古法ReAct里，工具描述是这样的（写在Prompt文本里）。现在用FunctionCall，工具描述变成了标准的JSON

1.把工具的了SON描述+用户问题一起发给大模型
(不需要在Prompt里写"Action:xxx:xxx"这种格式说明了）

2.大模型自己判断要不要调用工具
需要调用 → 返回标准JSON格式的tool_calls
不需要 → 直接返回文本答案

3.我们解析JSON，执行对应的工具函数，拿到结果

4.把工具结果按规定格式加到对话里，开始下一轮循环

5.直到大模型不再返回tool_cal1s循环结束，输出最终答案

二、过程解析

2.1、RAG召回

        目标：用从向量数据库中获取与用户问题相关的上下文信息，避免大模型出现“幻觉”。
        步骤：
        1. 用户输入经 InputToRag Lambda Node 处理，生成用于召回的字符串
        2. 调用 Retriever 组件(以Milvus数据库为例），通过 Embedding 将问题向量化
        3. 向量数据库执行相似度匹配，返回相关文档
        4. 结果存入 map["documents"]，作为后续 Prompt 的上下文来源。

2.2、Prompt构建

        目标：将用户输入、RAG召回内容、对话历史整合成大模型可理解的prompt。prompt构建好后，将prompt移交给ReAct组件使用。

// 两个lambda node的输出合并

1 //示例Prompt结构
2 SystemPrompt："你是知识库助手，用{documents}回答问题，当前时间{date}"
3 UserPrompt："{content}\n历史对话：{history}"

        {content}：用户原始问题
        {documents}：RAG召回的相关文档
        {date}：当前时间（增强时效性）
        {history}：历史对话

2.3、Lambda Node

        Lambda Node：数据流转的"转换器"

2.3.1、InputToRag

        输入：用户原始问题（可自定义预处理，如过滤无关信息)。
        输出：用于RAG召回的字符串（直接影响召回精度，需确保与向量数据库存储内容匹配）。

2.3.2、InputToChat 

        输入：用户问题 + 对话历史。
        输出：map结构(含 content / history 等 key)，作为 ChatTemplate 的动态参数来源。

用户问题
    ↓
【 InputToRag Lambda Node 】  → 处理问题、准备检索
    ↓
检索文档（Milvus）
    ↓
【 另一个 Lambda Node 】     → 合并数据、整理变量
    ↓
【 ChatTemplate 】           → 拼成最终提示词
    ↓
大模型回答

2.4、Retriever

        拿着用户的问题→去向量库里找最相关文档的【核心组件】。它是连接「用户问题」和「你的知识库」的唯一桥梁。

2.5、总结

        RAG召回：赋予外部知识记忆能力。
        动态Prompt：让大模型学习外部知识，并带有记忆(历史对话）。
        ReAct模式：实现复杂任务拆解与工具调用。

三、代码设计

class AgentState(TypedDict):
    """Agent 状态"""
    # Agent 状态 = AI 的【短期记忆 + 工作记事本】
    # 一个消息列表，里面只能存储 LangChain 消息，BaseMessage
    # 给 messages 这个变量，附加一段 “特殊指令 / 规则”
    messages: Annotated[Sequence[BaseMessage], add_messages]

    def trim_messages_middleware(state: AgentState) -> dict[str, Any] | None:
        """
        修剪消息历史，只保留最近的几条消息以适应上下文窗口

        策略：
        - 保留第一条系统消息（System Message）
        - 保留最近的 6 条消息（3 轮对话）
        - 当消息少于等于 7 条时，不做修剪

        Args:
            state: Agent 状态

        Returns:
            包含修剪后消息的字典，如果无需修剪则返回 None
        """
        messages = state["messages"]

        # 如果消息数量较少，无需修剪
        if len(messages) <= 7:
            return None

        # 提取第一条系统消息
        first_msg = messages[0]

        # 保留最近的 6 条消息（确保包含完整的对话轮次）
        recent_messages = messages[-6:] if len(messages) % 2 == 0 else messages[-7:]

        # 构建新的消息列表
        new_messages = [first_msg] + list(recent_messages)

        logger.debug(f"修剪消息历史: {len(messages)} -> {len(new_messages)} 条")

        return {
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),
                *new_messages
            ]
        }

class RagAgentService:
    """RAG Agent 服务 - 使用 LangGraph + ChatQwen 原生集成"""

    def __init__(self, streaming: bool = True):
        """初始化 RAG Agent 服务

        Args:
            streaming: 是否启用流式输出，默认为 True
        """
        self.model_name = config.rag_model
        self.streaming = streaming
        self.system_prompt = self._build_system_prompt()


        self.model = ChatQwen(
            model=self.model_name,
            api_key=config.dashscope_api_key,
            temperature=0.7,
            streaming=streaming,
        )

        # 定义基础工具
        self.tools = [retrieve_knowledge, get_current_time]

        # MCP 客户端（延迟初始化，使用全局管理）
        self.mcp_tools: list = []

        # 创建内存检查点（用于会话管理）
        self.checkpointer = MemorySaver()

        # Agent 初始化（会在异步方法中完成）
        self.agent = None
        self._agent_initialized = False

        logger.info(f"RAG Agent 服务初始化完成 (ChatQwen), model={self.model_name}, streaming={streaming}")

    async def _initialize_agent(self):
        """异步初始化 Agent（包括 MCP 工具）"""
        if self._agent_initialized:
            return

        # 使用全局 MCP 客户端管理器（带重试拦截器）
        mcp_client = await get_mcp_client_with_retry()

        # 获取 MCP 工具
        # 0. MCP 客户端自动去问所有连接的 MCP 服务器
        # 1. 向所有MCP服务器请求工具列表
        # 2. 收集所有工具
        # 3. 自动包装成 LangChain工具
        # 4. 返回给你：mcp_tools
        mcp_tools = await mcp_client.get_tools()
        logger.info(f"成功加载 {len(mcp_tools)} 个 MCP 工具")

        # 将 MCP 工具添加到实例变量中
        self.mcp_tools = mcp_tools

        # 合并所有工具
        all_tools = self.tools + self.mcp_tools

        self.agent = create_agent(
            self.model,
            tools=all_tools,
            checkpointer=self.checkpointer,
        )

        self._agent_initialized = True


        if all_tools:
            tool_names = [tool.name if hasattr(tool, "name") else str(tool) for tool in all_tools]
            logger.info(f"可用工具列表: {', '.join(tool_names)}")

    def _build_system_prompt(self) -> str:
        """
        构建系统提示词

        注意：LangChain 框架会自动将工具信息传递给 LLM，
        因此系统提示词中无需列举具体的工具列表。

        Returns:
            str: 系统提示词
        """
        from textwrap import dedent

        return dedent("""
            你是一个专业的AI助手，能够使用多种工具来帮助用户解决问题。

            工作原则:
            1. 理解用户需求，选择合适的工具来完成任务
            2. 当需要获取实时信息或专业知识时，主动使用相关工具
            3. 基于工具返回的结果提供准确、专业的回答
            4. 如果工具无法提供足够信息，请诚实地告知用户

            回答要求:
            - 保持友好、专业的语气
            - 回答简洁明了，重点突出
            - 基于事实，不编造信息
            - 如有不确定的地方，明确说明

            请根据用户的问题，灵活使用可用工具，提供高质量的帮助。
        """).strip()

    async def query(
        self,
        question: str,
        session_id: str,
    ) -> str:
        """
        非流式处理用户问题（一次性返回完整答案）

        Args:
            question: 用户问题
            session_id: 会话ID（作为 thread_id）

        Returns:
            str: 完整答案
        """
        try:
            await self._initialize_agent()

            logger.info(f"[会话 {session_id}] RAG Agent 收到查询（非流式）: {question}")

            # 构建消息列表（系统提示 + 用户问题）
            messages = [
                SystemMessage(content=self.system_prompt),
                HumanMessage(content=question)
            ]

            # 构建 Agent 输入
            agent_input = {"messages": messages}

            # 配置 thread_id（用于会话持久化）
            config_dict = {
                "configurable": {
                    "thread_id": session_id
                }
            }

            # AI真正干活的代码
            # LangGraph Agent 全自动完成！
            result = await self.agent.ainvoke(
                input=agent_input,
                config=config_dict,
            )

            # 提取最终答案
            messages_result = result.get("messages", [])
            # 消息列表是这样的
            # [
            #     SystemMessage("你是助手..."),
            #     HumanMessage("什么是向量数据库？"),
            #     AIMessage("向量数据库是..."),  ← 最后一条就是答案！
            # ]
            if messages_result:
                last_message = messages_result[-1]
                answer = last_message.content if hasattr(last_message, 'content') else str(last_message)

                # 记录工具调用
                if hasattr(last_message, "tool_calls") and last_message.tool_calls:
                    tool_names = [tc.get("name", "unknown") for tc in last_message.tool_calls]
                    logger.info(f"[会话 {session_id}] Agent 调用了工具: {tool_names}")

                logger.info(f"[会话 {session_id}] RAG Agent 查询完成（非流式）")
                return answer

            logger.warning(f"[会话 {session_id}] Agent 返回结果为空")
            return ""

        except Exception as e:
            logger.error(f"[会话 {session_id}] RAG Agent 查询失败（非流式）: {e}")
            raise

    async def query_stream(
        self,
        question: str,
        session_id: str,
    ) -> AsyncGenerator[Dict[str, Any], None]:
        """
        流式处理用户问题（逐步返回答案片段）

        Args:
            question: 用户问题
            session_id: 会话ID（作为 thread_id）

        Yields:
            Dict[str, Any]: 包含流式数据的字典
                - type: "content" | "tool_call" | "complete" | "error"
                - data: 具体内容
        """
        try:
            await self._initialize_agent()

            logger.info(f"[会话 {session_id}] RAG Agent 收到查询（流式）: {question}")

            # 构建消息列表（系统提示 + 用户问题）
            messages = [
                SystemMessage(content=self.system_prompt),
                HumanMessage(content=question)
            ]

            # 构建 Agent 输入
            agent_input = {"messages": messages}

            # 配置 thread_id（用于会话持久化）
            config_dict = {
                "configurable": {
                    "thread_id": session_id
                }
            }

            async for token, metadata in self.agent.astream(
                input=agent_input,
                config=config_dict,
                stream_mode="messages",
            ):
                # 读取当前节点名称，看看这段内容是哪个节点输出的
                node_name = metadata.get('langgraph_node', 'unknown') if isinstance(metadata, dict) else 'unknown'
                message_type = type(token).__name__
                # 只处理 AI 发出的消息，过滤掉用户消息、系统消息、工具消息，只保留 AI 要输出的回答内容
                if message_type in ("AIMessage", "AIMessageChunk"):
                    # getattr(对象, 属性名, 取不到时的默认值)
                    content_blocks = getattr(token, 'content_blocks', None)

                    if content_blocks and isinstance(content_blocks, list):
                        for block in content_blocks:
                            if isinstance(block, dict) and block.get('type') == 'text':
                                text_content = block.get('text', '')
                                if text_content:
                                    yield {
                                        "type": "content",
                                        "data": text_content,
                                        "node": node_name
                                    }

            logger.info(f"[会话 {session_id}] RAG Agent 查询完成（流式）")
            yield {"type": "complete"}

        except Exception as e:
            logger.error(f"[会话 {session_id}] RAG Agent 查询失败（流式）: {e}")
            yield {
                "type": "error",
                "data": str(e)
            }
            raise

    def get_session_history(self, session_id: str) -> list:
        """
        根据 session_id，从 AI 的内存记忆里获取会话历史（从 MemorySaver 的 checkpointer 中读取）
        包括用户之前问的问题和 AI 之前的回答

        Args:
            session_id: 会话ID（即 thread_id）

        Returns:
            list: 消息历史列表 [{"role": "user|assistant", "content": "...", "timestamp": "..."}]
        """
        try:
            # 使用 checkpointer 的 get 方法获取最新的检查点
            config = {"configurable": {"thread_id": session_id}}
            
            # 获取该 thread 的最新检查点
            checkpoint_tuple = self.checkpointer.get(config)
            
            if not checkpoint_tuple:
                logger.info(f"获取会话历史: {session_id}, 消息数量: 0")
                return []
            
            # checkpoint_tuple 可能是命名元组或普通元组，安全地提取 checkpoint
            # 通常第一个元素是 checkpoint 数据
            # 因为 LangGraph 不同版本，返回的格式不一样！
            if hasattr(checkpoint_tuple, 'checkpoint'):
                checkpoint_data = checkpoint_tuple.checkpoint  # type: ignore
            else:
                # 如果是普通元组，第一个元素是 checkpoint
                checkpoint_data = checkpoint_tuple[0] if checkpoint_tuple else {}
            
            # 从检查点中提取消息，包含：
            # SystemMessage（系统提示词）
            # HumanMessage（用户问题）
            # AIMessage（AI 回答）
            # {
            #     "channel_values": {
            #         "messages": [
            #             SystemMessage("你是助手..."),
            #             HumanMessage("什么是向量数据库？"),
            #             AIMessage("向量数据库是...")
            #         ]
            #     },
            #     "metadata": { ... }
            # }
            messages = checkpoint_data.get("channel_values", {}).get("messages", [])
            
            # 转换为前端需要的格式
            history = []
            for msg in messages:
                # 跳过系统消息
                if isinstance(msg, SystemMessage):
                    continue
                    
                # 判断角色，用户还是AI
                role = "user" if isinstance(msg, HumanMessage) else "assistant"
                content = msg.content if hasattr(msg, 'content') else str(msg)
                
                # 提取时间戳（如果有的话）
                timestamp = getattr(msg, 'timestamp', None)
                if timestamp:
                    history.append({
                        "role": role,
                        "content": content,
                        "timestamp": timestamp
                    })
                else:
                    from datetime import datetime
                    history.append({
                        "role": role,
                        "content": content,
                        "timestamp": datetime.now().isoformat()
                    })
            
            logger.info(f"获取会话历史: {session_id}, 消息数量: {len(history)}")
            return history
            
        except Exception as e:
            logger.error(f"获取会话历史失败: {session_id}, 错误: {e}")
            return []

    def clear_session(self, session_id: str) -> bool:
        """
        清空会话历史（从 MemorySaver checkpointer 中删除）

        Args:
            session_id: 会话ID（即 thread_id）

        Returns:
            bool: 是否成功
        """
        try:
            # 使用 checkpointer 的 delete_thread 方法删除该 thread 的所有检查点
            self.checkpointer.delete_thread(session_id)
            
            logger.info(f"已清除会话历史: {session_id}")
            return True
            
        except Exception as e:
            logger.error(f"清空会话历史失败: {session_id}, 错误: {e}")
            return False

    async def cleanup(self):
        """清理资源"""
        try:
            logger.info("清理 RAG Agent 服务资源...")
            # MCP 客户端由全局管理器统一管理，无需手动清理
            logger.info("RAG Agent 服务资源已清理")
        except Exception as e:
            logger.error(f"清理资源失败: {e}")

@tool(response_format="content_and_artifact")
def retrieve_knowledge(query: str) -> Tuple[str, List[Document]]:
    """从知识库中检索相关信息来回答问题
    
    当用户的问题涉及专业知识、文档内容或需要参考资料时，使用此工具。
    
    Args:
        query: 用户的问题或查询
        
    Returns:
        Tuple[str, List[Document]]: (格式化的上下文文本, 原始文档列表)
    """
    try:
        logger.info(f"知识检索工具被调用: query='{query}'")
        
        # 从向量存储中检索相关文档
        vector_store = vector_store_manager.get_vector_store()
        # 把向量库 包装成检索器（Retriever）
        # 检索器 = 专门用来查询、搜索的工具
        retriever = vector_store.as_retriever(
            search_kwargs={"k": config.rag_top_k}
        )
        
        docs = retriever.invoke(query)
        
        if not docs:
            logger.warning("未检索到相关文档")
            return "没有找到相关信息。", []
        
        # 格式化文档为上下文
        context = format_docs(docs)
        
        logger.info(f"检索到 {len(docs)} 个相关文档")
        return context, docs
        
    except Exception as e:
        logger.error(f"知识检索工具调用失败: {e}")
        return f"检索知识时发生错误: {str(e)}", []


def format_docs(docs: List[Document]) -> str:
    """
    格式化文档列表为上下文文本
    
    Args:
        docs: 文档列表
        
    Returns:
        str: 格式化的上下文文本
    """
    formatted_parts = []
    
    for i, doc in enumerate(docs, 1):
        # 提取元数据
        metadata = doc.metadata
        source = metadata.get("_file_name", "未知来源")
        
        # 提取标题信息 (如果有)
        headers = []
        for key in ["h1", "h2", "h3"]:
            if key in metadata and metadata[key]:
                headers.append(metadata[key])
        
        header_str = " > ".join(headers) if headers else ""
        
        # 构建格式化文本
        formatted = f"【参考资料 {i}】"
        if header_str:
            formatted += f"\n标题: {header_str}"
        formatted += f"\n来源: {source}"
        formatted += f"\n内容:\n{doc.page_content}\n"
        
        formatted_parts.append(formatted)
    
    return "\n".join(formatted_parts)