Merge branch 'main' of github.com:xerrors/Yuxi-Know

Author: xerrors

.gitignore

@@ -40,11 +40,10 @@ cache
 .trae
 .pytest_cache
 
-###  (企业私有代码 - 仅忽略敏感配置，不忽略代码文件)
-# 移除了 *.private* 和 *_private 规则，允许 Git 本地管理
-# 通过 .git/info/exclude 或本地分支管理私有代码
 *.secret*
 *.nogit*
+*_private
+*.private
 # *.local* 保留用于本地配置文件
 *.local.py
 *.local.js

src/agents/__init__.py

@@ -3,11 +3,12 @@
 import inspect
 from pathlib import Path
 
+from server.utils.singleton import SingletonMeta
 from src.agents.common.base import BaseAgent
 from src.utils import logger
 
 
-class AgentManager:
+class AgentManager(metaclass=SingletonMeta):
     def __init__(self):
         self._classes = {}
         self._instances = {}  # 存储已创建的 agent 实例

src/agents/common/base.py

@@ -67,6 +67,14 @@ async def stream_messages(self, messages: list[str], input_context=None, **kwarg
         ):
             yield msg, metadata
 
+    async def invoke_messages(self, messages: list[str], input_context=None, **kwargs):
+        graph = await self.get_graph()
+        context = self.context_schema.from_file(module_name=self.module_name, input_context=input_context)
+        logger.debug(f"invoke_messages: {context}")
+        input_config = {"configurable": input_context, "recursion_limit": 100}
+        msg = await graph.ainvoke({"messages": messages}, context=context, config=input_config)
+        return msg
+
     async def check_checkpointer(self):
         app = await self.get_graph()
         if not hasattr(app, "checkpointer") or app.checkpointer is None:

src/agents/common/state.py

@@ -0,0 +1,20 @@
+"""Define the state structures for the agent."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from typing import Annotated
+
+from langchain.messages import AnyMessage
+from langgraph.graph import add_messages
+
+
+@dataclass
+class BaseState:
+    """Defines the input state for the agent, representing a narrower interface to the outside world.
+
+    This class is used to define the initial state and structure of incoming data.
+    """
+
+    messages: Annotated[Sequence[AnyMessage], add_messages] = field(default_factory=list)

src/agents/common/toolagent.py

@@ -0,0 +1,87 @@
+from abc import abstractmethod
+from typing import Any, cast
+
+from langchain.messages import AIMessage, ToolMessage
+from langgraph.prebuilt import ToolNode
+from langgraph.runtime import Runtime
+
+from src.agents.common.base import BaseAgent
+from src.agents.common.mcp import get_mcp_tools
+from src.agents.common.models import load_chat_model
+from src.utils import logger
+
+from .state import BaseState
+from .context import BaseContext
+
+class ToolAgent(BaseAgent):
+    name = "ToolAgent"
+    description = "具有工具调用能力的Agent"
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.graph = None
+        self.checkpointer = None
+        self.context_schema = BaseContext
+        self.agent_tools = None
+
+
+    # TODO:[修改建议] _get_invoke_tools,llm_call,dynamic_tools_node这类针对工具调用的功能大多数Agent都能用得到
+    # 可以通过一个ToolAgent类继承BaseAgent,通过重写抽象方法获取tools,通过继承BaseState和BaseContext获取配置
+    # 必要时可通过重写以下方法实现其他逻辑
+    @abstractmethod
+    def get_tools(self):
+        logger.error(f"get_tools() is not implemented in {self.__class__.__name__}")
+        return []
+
+
+    async def _get_invoke_tools(self, selected_tools: list[str], selected_mcps: list[str]):
+        """根据配置获取工具。
+        默认不使用任何工具。
+        如果配置为列表，则使用列表中的工具。
+        """
+        enabled_tools = []
+        self.agent_tools = self.agent_tools or self.get_tools()
+        if selected_tools and isinstance(selected_tools, list) and len(selected_tools) > 0:
+            # 使用配置中指定的工具
+            enabled_tools = [tool for tool in self.agent_tools if tool.name in selected_tools]
+
+        if selected_mcps and isinstance(selected_mcps, list) and len(selected_mcps) > 0:
+            for mcp in selected_mcps:
+                enabled_tools.extend(await get_mcp_tools(mcp))
+
+        return enabled_tools
+
+    async def llm_call(self, state: BaseState, runtime: Runtime[BaseContext] = None) -> dict[str, Any]:
+        """调用 llm 模型 - 异步版本以支持异步工具"""
+        model = load_chat_model(runtime.context.model)
+
+        # 这里要根据配置动态获取工具
+        available_tools = await self._get_invoke_tools(runtime.context.tools, runtime.context.mcps)
+        logger.info(f"LLM binded ({len(available_tools)}) available_tools: {[tool.name for tool in available_tools]}")
+
+        if available_tools:
+            model = model.bind_tools(available_tools)
+
+        # 使用异步调用
+        response = cast(
+            AIMessage,
+            await model.ainvoke([{"role": "system", "content": runtime.context.system_prompt}, *state.messages]),
+        )
+        return {"messages": [response]}
+
+    async def dynamic_tools_node(self, state: BaseState, runtime: Runtime[BaseContext]) -> dict[str, list[ToolMessage]]:
+        """Execute tools dynamically based on configuration.
+
+        This function gets the available tools based on the current configuration
+        and executes the requested tool calls from the last message.
+        """
+        # Get available tools based on configuration
+        available_tools = await self._get_invoke_tools(runtime.context.tools, runtime.context.mcps)
+
+        # Create a ToolNode with the available tools
+        tool_node = ToolNode(available_tools)
+
+        # Execute the tool node
+        result = await tool_node.ainvoke(state)
+
+        return cast(dict[str, list[ToolMessage]], result)

src/agents/common/tools.py

@@ -5,11 +5,54 @@
 from langchain.tools import tool
 from langchain_core.tools import StructuredTool
 from langchain_tavily import TavilySearch
+from langgraph.types import interrupt
 from pydantic import BaseModel, Field
 
 from src import config, graph_base, knowledge_base
 from src.utils import logger
 
+# TODO[修改建议]:前端需要通过interrupt进行交互，点击是或否来批准执行
+# 返回中断点：
+# is_approved : bool = True 或者 False
+# resume_command = Command(resume=is_approved)
+# stream = graph.stream(resume_command, config=config, stream_mode="messages")
+# graph.invoke(resume_command, config=config)
+@tool(name_or_callable="人工审批工具", description="请求人工审批工具，用于在执行重要操作前获得人类确认。")
+def get_approved_user_goal(
+    operation_description: str,
+)->dict:
+    """
+    请求人工审批，在执行重要操作前获得人类确认。
+
+    Args:
+        operation_description: 需要审批的操作描述，例如 "调用知识库工具"
+    Returns:
+        dict: 包含审批结果的字典，格式为 {"approved": bool, "message": str}
+    """
+    # 构建详细的中断信息
+    interrupt_info = {
+        "question": f"是否批准以下操作？",
+        "operation": operation_description,
+    }
+
+    # 触发人工审批
+    is_approved = interrupt(interrupt_info)
+
+    # 返回审批结果
+    if is_approved:
+        result = {
+            "approved": True,
+            "message": f"✅ 操作已批准：{operation_description}",
+        }
+        print(f"✅ 人工审批通过: {operation_description}")
+    else:
+        result = {
+            "approved": False,
+            "message": f"❌ 操作被拒绝：{operation_description}",
+        }
+        print(f"❌ 人工审批被拒绝: {operation_description}")
+
+    return result
 
 @tool(name_or_callable="查询知识图谱", description="使用这个工具可以查询知识图谱中包含的三元组信息。")
 def query_knowledge_graph(query: Annotated[str, "The keyword to query knowledge graph."]) -> Any:
@@ -31,6 +74,7 @@ def get_static_tools() -> list:
     """注册静态工具"""
     static_tools = [
         query_knowledge_graph,
+        get_approved_user_goal
     ]
 
     # 检查是否启用网页搜索

src/agents/multiAgent/__init__.py

@@ -0,0 +1,3 @@
+from .graph import SampleMultiAgent
+
+__all__ = ["SampleMultiAgent"]

src/agents/multiAgent/context.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass, field
+from typing import Annotated
+
+from src.agents.common.context import BaseContext
+from src.agents.common.mcp import MCP_SERVERS
+from src.agents.common.tools import gen_tool_info
+
+from .tools import get_tools
+
+
+@dataclass(kw_only=True)
+class Context(BaseContext):
+    tools: Annotated[list[dict], {"__template_metadata__": {"kind": "tools"}}] = field(
+        default_factory=list,
+        metadata={
+            "name": "工具",
+            "options": gen_tool_info(get_tools()),  # 这里的选择是所有的工具
+            "description": "工具列表",
+        },
+    )
+
+    mcps: list[str] = field(
+        default_factory=list,
+        metadata={"name": "MCP服务器", "options": list(MCP_SERVERS.keys()), "description": "MCP服务器列表"},
+    )

src/agents/multiAgent/graph.py

@@ -0,0 +1,61 @@
+from langgraph.graph import END, START, StateGraph
+from langgraph.prebuilt import tools_condition
+
+from src.agents.common.toolagent import ToolAgent
+
+from .context import Context
+from .state import State
+from .tools import get_tools
+
+
+class SampleMultiAgent(ToolAgent):
+    name = "MultiAgent智能体"
+    description = "Supervisor智能体，具有调用其他子智能体的能力(在工具中添加)"
+
+    # TODO[已完成]: 通过将其他agent封装为工具的方式添加了多智能体调度
+    '''
+    你是一个多智能体核心，通过多智能体调用的方式帮助用户完成一系列任务：
+
+    1.当你需要知识库问答功能时，请调用对话聊天智能体实现
+    2.当你需要加密计算的时候，请调用加密计算智能体实现
+    '''
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.graph = None
+        self.checkpointer = None
+        self.context_schema = Context
+        self.agent_tools = None
+
+    def get_tools(self):
+        return get_tools()
+
+    async def get_graph(self, **kwargs):
+        """构建图"""
+        if self.graph:
+            return self.graph
+
+        builder = StateGraph(State, context_schema=self.context_schema)
+        builder.add_node("chatbot", self.llm_call)
+        builder.add_node("tools", self.dynamic_tools_node)
+        builder.add_edge(START, "chatbot")
+        builder.add_conditional_edges(
+            "chatbot",
+            tools_condition,
+        )
+        builder.add_edge("tools", "chatbot")
+        builder.add_edge("chatbot", END)
+
+        self.checkpointer = await self._get_checkpointer()
+        graph = builder.compile(checkpointer=self.checkpointer, name=self.name)
+        self.graph = graph
+        return graph
+
+
+def main():
+    pass
+
+
+if __name__ == "__main__":
+    main()
+    # asyncio.run(main())

src/agents/multiAgent/state.py

@@ -0,0 +1,22 @@
+"""Define the state structures for the agent."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from typing import Annotated
+
+from langchain.messages import AnyMessage
+from langgraph.graph import add_messages
+
+from src.agents.common.state import BaseState
+
+
+@dataclass
+class State(BaseState):
+    """Defines the input state for the agent, representing a narrower interface to the outside world.
+
+    This class is used to define the initial state and structure of incoming data.
+    """
+
+    messages: Annotated[Sequence[AnyMessage], add_messages] = field(default_factory=list)