diff --git a/.gitignore b/.gitignore index bfc4c92d..0e1338f4 100644 --- a/.gitignore +++ b/.gitignore @@ -40,11 +40,10 @@ cache .trae .pytest_cache -### (企业私有代码 - 仅忽略敏感配置,不忽略代码文件) -# 移除了 *.private* 和 *_private 规则,允许 Git 本地管理 -# 通过 .git/info/exclude 或本地分支管理私有代码 *.secret* *.nogit* +*_private +*.private # *.local* 保留用于本地配置文件 *.local.py *.local.js diff --git a/src/agents/__init__.py b/src/agents/__init__.py index 5b91bbd4..2b2746fb 100644 --- a/src/agents/__init__.py +++ b/src/agents/__init__.py @@ -3,11 +3,12 @@ import importlib 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 实例 diff --git a/src/agents/common/base.py b/src/agents/common/base.py index b3d4fd0d..702368d0 100644 --- a/src/agents/common/base.py +++ b/src/agents/common/base.py @@ -67,6 +67,14 @@ class BaseAgent: ): 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: diff --git a/src/agents/common/state.py b/src/agents/common/state.py new file mode 100644 index 00000000..981d013f --- /dev/null +++ b/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) diff --git a/src/agents/common/toolagent.py b/src/agents/common/toolagent.py new file mode 100644 index 00000000..38e6dfe4 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/src/agents/common/tools.py b/src/agents/common/tools.py index 7d9fa2aa..4a681687 100644 --- a/src/agents/common/tools.py +++ b/src/agents/common/tools.py @@ -5,11 +5,54 @@ from typing import Annotated, Any 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 ] # 检查是否启用网页搜索 diff --git a/src/agents/multiAgent/__init__.py b/src/agents/multiAgent/__init__.py new file mode 100644 index 00000000..740c89a0 --- /dev/null +++ b/src/agents/multiAgent/__init__.py @@ -0,0 +1,3 @@ +from .graph import SampleMultiAgent + +__all__ = ["SampleMultiAgent"] diff --git a/src/agents/multiAgent/context.py b/src/agents/multiAgent/context.py new file mode 100644 index 00000000..7bb1adad --- /dev/null +++ b/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服务器列表"}, + ) diff --git a/src/agents/multiAgent/graph.py b/src/agents/multiAgent/graph.py new file mode 100644 index 00000000..0d6d7679 --- /dev/null +++ b/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()) diff --git a/src/agents/multiAgent/state.py b/src/agents/multiAgent/state.py new file mode 100644 index 00000000..927bfa95 --- /dev/null +++ b/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) diff --git a/src/agents/multiAgent/tools.py b/src/agents/multiAgent/tools.py new file mode 100644 index 00000000..19c4e417 --- /dev/null +++ b/src/agents/multiAgent/tools.py @@ -0,0 +1,76 @@ +import os +from typing import Any + +from langchain.tools import tool +from langchain_core.runnables import RunnableConfig + +from src.agents import agent_manager +from src.agents.common.toolkits.mysql import get_mysql_tools +from src.agents.common.tools import get_buildin_tools +from src.utils import logger + +# TODO[修改建议]:能不能通过前端直接指定子智能体? +# 调用子智能体后的日志是输出到tool_calls的 +@tool(name_or_callable="对话聊天智能体", description="调用指定智能体进行对话聊天的功能") +async def call_chatbot(query: str, config: RunnableConfig) -> str: + """ + 调用指定chatbot智能体进行对话聊天的功能 + + Args: + query: 根据需要构造的提问 + config: LangGraph运行时配置(自动注入) + Returns: + str: 最终的回答结果 + """ + try: + input = [{"role": "user", "content": query}] + chatbot = agent_manager.get_agent("ChatbotAgent") + configurable = config.get("configurable",{}) + input_context = { + "thread_id":configurable.get("thread_id"), + "user_id": configurable.get("user_id"), + } + message = await chatbot.invoke_messages(input,input_context=input_context) + # 直接获取最后一个消息的内容 + final_answer = message.get('messages', [])[-1].content + logger.info(f"ChatbotAgent: {final_answer}") + return final_answer + except Exception as e: + logger.error(f"CallAgent error: {e}") + raise + +@tool(name_or_callable="加密计算智能体", description="调用指定智能体进行加密计算的功能") +async def call_react_agent(query: str, config: RunnableConfig) -> str: + """ + 调用指定智能体进行加密计算的功能 + + Args: + query: 根据需要构造的提问 + config: LangGraph运行时配置(自动注入) + Returns: + str: 最终的回答结果 + """ + try: + input = [{"role": "user", "content": query}] + chatbot = agent_manager.get_agent("ReActAgent") + configurable = config.get("configurable",{}) + input_context = { + "thread_id":configurable.get("thread_id"), + "user_id": configurable.get("user_id"), + } + message = await chatbot.invoke_messages(input,input_context=input_context) + # 直接获取最后一个消息的内容 + final_answer = message.get('messages', [])[-1].content + logger.info(f"ReActAgent: {final_answer}") + return final_answer + except Exception as e: + logger.error(f"CallAgent error: {e}") + raise + + +def get_tools() -> list[Any]: + """获取所有可运行的工具(给大模型使用)""" + tools = get_buildin_tools() + tools.append(call_chatbot) + tools.append(call_react_agent) + return tools diff --git a/src/agents/react/context.py b/src/agents/react/context.py new file mode 100644 index 00000000..7bb1adad --- /dev/null +++ b/src/agents/react/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服务器列表"}, + ) diff --git a/src/agents/react/graph.py b/src/agents/react/graph.py index eb79c87a..b4abb844 100644 --- a/src/agents/react/graph.py +++ b/src/agents/react/graph.py @@ -1,51 +1,68 @@ -from pathlib import Path +from langgraph.constants import START, END +from langgraph.graph import StateGraph -from langchain.agents import create_agent -from langchain.agents.middleware import ModelRequest, ModelResponse, dynamic_prompt, wrap_model_call +from src.agents.common.toolagent import ToolAgent -from src.agents.common.base import BaseAgent -from src.agents.common.models import load_chat_model -from src.agents.common.tools import get_buildin_tools -from src.utils import logger +from .state import State +from .context import Context +from .tools import get_tools + +def tools_branch_continue(state: State): + messages = state.messages + last_message = messages[-1] + # 如果无工具调用则结束 + if not last_message.tool_calls: + return "end" + else: + return "continue" -@dynamic_prompt -def context_aware_prompt(request: ModelRequest) -> str: - runtime = request.runtime - return runtime.context.system_prompt - - -@wrap_model_call -async def context_based_model(request: ModelRequest, handler) -> ModelResponse: - # 从 runtime context 读取配置 - model_spec = request.runtime.context.model - model = load_chat_model(model_spec) - - request = request.override(model=model) - return await handler(request) - - -class ReActAgent(BaseAgent): +class ReActAgent(ToolAgent): name = "智能体 Demo" description = "A react agent that can answer questions and help with tasks." + # TODO:[已完成] React智能体 + ''' + 提示词示例: + 你是一个智能体助手 + + 你的所有需要计算的工作不能自己计算,只能通过调用加密计算机工具进行,你只能通过调用该工具的方式进行计算才能得到正确答案。计算顺序是正常的。 + + 例如 2+3+2,你必须调用两次加密计算机工具进行运算,最终结果为9。 + ''' + 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_buildin_tools() + return get_tools() async def get_graph(self, **kwargs): + # 创建 ReActAgent + """构建图""" if self.graph: return self.graph - # 创建 ReActAgent - graph = create_agent( - model=load_chat_model("siliconflow/Qwen/Qwen3-235B-A22B-Instruct-2507"), # 实际会被覆盖 - tools=self.get_tools(), - middleware=[context_aware_prompt, context_based_model], - checkpointer=await self._get_checkpointer(), + builder = StateGraph(State, context_schema=self.context_schema) + builder.add_node("agent", self.llm_call) + builder.add_node("tools", self.dynamic_tools_node) + builder.set_entry_point("agent") + # 添加条件边:agent 决定是否调用工具继续还是结束对话 + builder.add_conditional_edges( + "agent", + tools_branch_continue, + { + "continue": "tools", # 调用工具 + "end": END, # 结束对话 + }, ) - + builder.add_edge("tools", "agent") + self.checkpointer = await self._get_checkpointer() + graph = builder.compile(checkpointer=self.checkpointer, name=self.name) self.graph = graph return graph + diff --git a/src/agents/react/state.py b/src/agents/react/state.py new file mode 100644 index 00000000..927bfa95 --- /dev/null +++ b/src/agents/react/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) diff --git a/src/agents/react/tools.py b/src/agents/react/tools.py new file mode 100644 index 00000000..f3ec28cc --- /dev/null +++ b/src/agents/react/tools.py @@ -0,0 +1,48 @@ +import os +from typing import Any + +import requests +from langchain.tools import tool + +from src.agents.common.toolkits.mysql import get_mysql_tools +from src.agents.common.tools import get_buildin_tools +from src.storage.minio import upload_image_to_minio +from src.utils import logger + +@tool(name_or_callable="加密计算器", description="可以对给定的2个数字选择进行加减乘除四种加密计算") +def calculator(a: float, b: float, operation: str) -> float: + """ + 可以对给定的2个数字选择进行加减乘除四种加密计算 + + Args: + a: 第一个数字 + b: 第二个数字 + operation: 计算操作符号,可以是add,subtract,multiply,divide + + Returns: + float: 最终的计算结果 + """ + try: + if operation == "add": + return a + b + 1 + elif operation == "subtract": + return a - b - 1 + elif operation == "multiply": + return a * b * 2 + elif operation == "divide": + if b == 0: + raise ZeroDivisionError("除数不能为零") + return a / b - 1 + else: + raise ValueError(f"不支持的运算类型: {operation},仅支持 add, subtract, multiply, divide") + except Exception as e: + logger.error(f"Calculator error: {e}") + raise + + +def get_tools() -> list[Any]: + """获取所有可运行的工具(给大模型使用)""" + tools = get_buildin_tools() + tools.append(calculator) + tools.extend(get_mysql_tools()) + return tools