diff --git a/src/mini_agent/agent/base_agent.py b/src/mini_agent/agent/base_agent.py new file mode 100644 index 0000000..23a9371 --- /dev/null +++ b/src/mini_agent/agent/base_agent.py @@ -0,0 +1,69 @@ +from abc import abstractmethod +from collections.abc import Iterator +import json +from typing import Any + +from openai.types.chat import ChatCompletionMessage, ChatCompletionMessageToolCallUnion +from openai.types.chat.chat_completion_message_param import ChatCompletionMessageParam +from pydantic import BaseModel, TypeAdapter + +from mini_agent.agent.context import ConversationContext +from mini_agent.agent.message import Message + + +class BaseAgent: + @property + @abstractmethod + def id(self) -> str: + pass + + @property + @abstractmethod + def name(self) -> str: + pass + + @property + @abstractmethod + def description(self) -> str: + pass + + @property + @abstractmethod + def compatibility(self) -> list[str]: + pass + + @property + @abstractmethod + def system_prompt(self) -> str: + pass + + def serialize_json(self, data: Any) -> str: + if isinstance(data, BaseModel): + return data.model_dump_json(ensure_ascii=False) + return json.dumps(data, ensure_ascii=False, default=str) + + def _initialize_context(self, context: ConversationContext, question: str) -> None: + """ + 初始化会话context + """ + if len(context.messages) == 0: + context.messages.append({"role": "system", "content": self.system_prompt}) + context.messages.append({"role": "user", "content": question}) + + def _build_message_param(self, message: ChatCompletionMessage) -> ChatCompletionMessageParam: + """ + 将ChatCompletionMessage转换成ChatCompletionMessageParam + """ + message_param: ChatCompletionMessageParam = { + "role": message.role, + "content": message.content, + } + + if message.tool_calls: + adapter = TypeAdapter(list[ChatCompletionMessageToolCallUnion]) + message_param["tool_calls"] = adapter.dump_python(message.tool_calls, exclude_none=True) + return message_param + + @abstractmethod + def run(self, question: str, context: ConversationContext) -> Iterator[Message]: + pass diff --git a/src/mini_agent/agent/base_tool_agent.py b/src/mini_agent/agent/base_tool_agent.py new file mode 100644 index 0000000..f583857 --- /dev/null +++ b/src/mini_agent/agent/base_tool_agent.py @@ -0,0 +1,58 @@ +from abc import abstractmethod +import json +from typing import Any, override + +from openai.types.chat import ChatCompletionMessageToolCallUnion + +from mini_agent.agent.base_agent import BaseAgent +from mini_agent.agent.context import ConversationContext +from mini_agent.tools.tool_repository import ToolRepository + + +class BaseToolAgent(BaseAgent): + @property + @abstractmethod + def tool_repository(self) -> ToolRepository: + pass + + @override + def _initialize_context(self, context: ConversationContext, question: str) -> None: + super()._initialize_context(context, question) + if len(context.discovered_tools) == 0 and self.tool_repository: + context.add_tool(self.tool_repository) + + def _parse_tool_call( + self, tool_call: ChatCompletionMessageToolCallUnion + ) -> tuple[str, dict[str, Any]]: + """ + 解析Tool call,返回(tool_id, arguments) + """ + if tool_call.type != "function" and tool_call.type != "custom": + raise ValueError(f"不支持的工具调用类型{tool_call.type}") + if tool_call.type == "function": + tool_id = tool_call.function.name + raw_arguments = tool_call.function.arguments + else: + tool_id = tool_call.custom.name + raw_arguments = tool_call.custom.input + arguments = json.loads(raw_arguments) + if not isinstance(arguments, dict): + raise ValueError("工具参数必须是 JSON 对象") + return (tool_id, arguments) + + def _execute_tool_call( + self, tool_id: str, arguments: dict[str, Any], context: ConversationContext + ) -> Any: + """ + 执行tool call + """ + observation = self.tool_repository.call_tool(tool_id, arguments) + # 更新当前可用工具 + if tool_id == self.tool_repository.id: + available_tools = [] + for item in observation: + if item.name not in context.discovered_tool_names: + context.add_tool(item) + available_tools.append(item.to_openai()) + observation = available_tools + return observation diff --git a/src/mini_agent/agent/context.py b/src/mini_agent/agent/context.py index c3f3a53..51c9a2a 100644 --- a/src/mini_agent/agent/context.py +++ b/src/mini_agent/agent/context.py @@ -6,6 +6,7 @@ from mini_agent.tools.tool import Tool class ConversationContext(BaseModel): conversation_id: str = Field(description="会话ID") + run_id: str = Field(description="当前会话轮id") messages: list[ChatCompletionMessageParam] = Field(description="消息历史", default_factory=list) _discovered_tools: list[Tool] = PrivateAttr(default_factory=list) @@ -17,6 +18,6 @@ class ConversationContext(BaseModel): return self._discovered_tools def add_tool(self, tool: Tool): - if tool.name not in self.discovered_tool_names: - self.discovered_tool_names.append(tool.name) + if tool.id not in self.discovered_tool_names: + self.discovered_tool_names.append(tool.id) self.discovered_tools.append(tool) diff --git a/src/mini_agent/agent/llm_client.py b/src/mini_agent/agent/llm_client.py new file mode 100644 index 0000000..07ecb59 --- /dev/null +++ b/src/mini_agent/agent/llm_client.py @@ -0,0 +1,48 @@ +import os + +from openai import OpenAI +from openai.types.chat import ( + ChatCompletion, + ChatCompletionMessageParam, + ChatCompletionToolChoiceOptionParam, + ChatCompletionToolParam, +) + + +class LLMClient: + def __init__( + self, + model: str | None = None, + api_key: str | None = None, + base_url: str | None = None, + timeout: int | None = None, + ) -> None: + self.model = model or os.getenv("LLM_MODEL_ID") + self.base_url = base_url or os.getenv("LLM_BASE_URL") + self.api_key = api_key or os.getenv("LLM_API_KEY") + self.timeout = timeout if timeout is not None else int(os.getenv("LLM_TIMEOUT", "60")) + + if not all([self.model, self.base_url, self.api_key]): + raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。") + + self.client = OpenAI( + api_key=self.api_key, + base_url=self.base_url, + timeout=self.timeout, + ) + + def think( + self, + messages: list[ChatCompletionMessageParam], + tools: list[ChatCompletionToolParam], + tool_choice: ChatCompletionToolChoiceOptionParam = "auto", + **kwargs, + ) -> ChatCompletion: + assert self.model is not None + return self.client.chat.completions.create( + model=self.model, + messages=messages, + tools=tools, + tool_choice=tool_choice, + **kwargs + ) diff --git a/src/mini_agent/agent/message.py b/src/mini_agent/agent/message.py index 977cdec..b1a11f7 100644 --- a/src/mini_agent/agent/message.py +++ b/src/mini_agent/agent/message.py @@ -1,47 +1,91 @@ from datetime import datetime from enum import StrEnum -from typing import Any +from typing import Any, Literal from uuid import uuid4 from pydantic import BaseModel, Field +class TextContent(BaseModel): + type: Literal["text"] = "text" + text: str + + +class ThinkContent(BaseModel): + type: Literal["think"] = "think" + text: str + + +class ImageContent(BaseModel): + type: Literal["image"] = "image" + url: str + mime_type: str | None = None + + +class JsonContent(BaseModel): + type: Literal["json"] = "json" + data: Any + + +class ToolCallContent(BaseModel): + type: Literal["tool_call"] = "tool_call" + id: str + name: str + tool_call_id:str + arguments: dict[str, Any] + + +class ToolResultContent(BaseModel): + type: Literal["tool_result"] = "tool_result" + tool_call_id: str + result: Any = None + error: str | None = None + + +class ErrorContent(BaseModel): + type: Literal["error"] = "error" + message: str + code: str | None = None + + +class AudioContent(BaseModel): + type: Literal["audio"] = "audio" + url: str + mime_type: str | None = None + + +MessageContent = ( + TextContent + | ThinkContent + | ImageContent + | JsonContent + | ToolCallContent + | ToolResultContent + | ErrorContent +) + + class MessageRole(StrEnum): USER = "user" ASSISTANT = "assistant" TOOL = "tool" -class MessageType(StrEnum): - TEXT = "text" - THINK = "think" - TOOL_CALL = "tool_call" - TOOL_RESULT = "tool_result" - - -class MessageStatus(StrEnum): +class RunStatus(StrEnum): RUNNING = "running" COMPLETED = "completed" FAILED = "failed" + CANCELLED = "cancelled" class Message(BaseModel): - id: str = Field(default_factory=lambda: str(uuid4())) - conversation_id: str - run_id: str | None = None + id: str = Field(description="消息ID", default_factory=lambda: str(uuid4())) + conversation_id: str = Field(description="会话ID") + run_id: str | None = Field(description="当前轮的ID", default=None) + run_status: RunStatus = Field(description="当前轮的状态") - role: MessageRole - type: MessageType + role: MessageRole = Field(description="当前角色") - content: Any = None - status: MessageStatus = MessageStatus.COMPLETED - - tool_call_id: str | None = None - - tool_name: str | None = None - - arguments: dict[str, Any] | None = None - - tool_call_result:Any |None = None + content: list[MessageContent] = Field(description="消息内容", default_factory=list) created_at: datetime = Field(default_factory=lambda: datetime.now()) diff --git a/src/mini_agent/agent/react_agent.py b/src/mini_agent/agent/react_agent.py index 4beb696..943696e 100644 --- a/src/mini_agent/agent/react_agent.py +++ b/src/mini_agent/agent/react_agent.py @@ -1,154 +1,93 @@ from collections.abc import Iterator -import json import os -from typing import Any +from typing import override from uuid import uuid7 -from openai import OpenAI -from openai.types.chat import ChatCompletionMessage, ChatCompletionMessageToolCallUnion -from openai.types.chat.chat_completion_message_param import ChatCompletionMessageParam -from pydantic import BaseModel, TypeAdapter - +from mini_agent.agent.base_tool_agent import BaseToolAgent from mini_agent.agent.context import ConversationContext -from mini_agent.agent.message import Message, MessageRole, MessageStatus, MessageType +from mini_agent.agent.llm_client import LLMClient +from mini_agent.agent.message import ( + ErrorContent, + Message, + MessageRole, + RunStatus, + TextContent, + ThinkContent, + ToolCallContent, + ToolResultContent, +) from mini_agent.tools.tavily_search import TavilySearch from mini_agent.tools.tool_repository import ToolRepository -class ReActAgent: - SYSTEM_PROMPT = """ - 你是一个能够调用外部工具的智能助手。 - - 工作规则： - - 1. 首先理解用户的问题，判断是否需要外部工具。 - 2. 如果需要工具，调用工具搜索接口搜索合适的工具。 - 3. 根据搜索结果选择合适的工具。 - 4. 工具执行后，根据返回结果判断下一步行动。 - 5. 当已经获得足够的信息，能够完整回答用户问题时，直接给出最终答案。 - - 注意事项： - - - 不允许编造工具执行结果。 - - 工具搜索没有结果时，可以尝试调整关键词重新搜索。 - - 如果工具无法完成任务，明确告知用户。 - - 不要为了调用工具而调用工具。 - - 只有在任务已经完成或明确无法继续时才结束回答。 - """ - +class ReActAgent(BaseToolAgent): def __init__( self, tool_repository: ToolRepository, - model: str | None = None, - api_key: str | None = None, - base_url: str | None = None, - timeout: int | None = None, + llm_client: LLMClient, max_steps: int = 10, ) -> None: - self.model = model or os.getenv("LLM_MODEL_ID") - self.base_url = base_url or os.getenv("LLM_BASE_URL") - self.api_key = api_key or os.getenv("LLM_API_KEY") - self.timeout = timeout if timeout is not None else int(os.getenv("LLM_TIMEOUT", "60")) + super().__init__() - if not all([self.model, self.base_url, self.api_key]): - raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。") - - self.client = OpenAI( - api_key=self.api_key, - base_url=self.base_url, - timeout=self.timeout, - ) - - self.tool_repos = tool_repository + self.client = llm_client self.max_steps = max_steps web_search_tool = TavilySearch() web_search_tool.config(api_key=os.getenv("TAVILY_KEY", "")) - self.tool_repos.register(web_search_tool) + self._tool_repository = tool_repository + self.tool_repository.register(web_search_tool) - def serialize_json(self, data: Any) -> str: - if isinstance(data, BaseModel): - return data.model_dump_json() - return json.dumps(data, ensure_ascii=False, default=str) + @property + def id(self) -> str: + return "react_agent" - def _initialize_context(self, context: ConversationContext, question: str) -> None: - """ - 初始化会话context - """ - if len(context.messages) == 0: - context.messages.append({"role": "system", "content": self.SYSTEM_PROMPT}) - if len(context.discovered_tools) == 0: - context.add_tool(self.tool_repos) - context.messages.append({"role": "user", "content": question}) + @property + def name(self) -> str: + return "ReAct Agent" - def _build_message_param(self, message: ChatCompletionMessage) -> ChatCompletionMessageParam: - """ - 将ChatCompletionMessage转换成ChatCompletionMessageParam - """ - message_param: ChatCompletionMessageParam = { - "role": message.role, - "content": message.content, - } + @property + def description(self) -> str: + return "思考-回答-观察-再思考的模式来解决问题" - if message.tool_calls: - adapter = TypeAdapter(list[ChatCompletionMessageToolCallUnion]) - message_param["tool_calls"] = adapter.dump_python(message.tool_calls, exclude_none=True) - return message_param + @property + def compatibility(self) -> list[str]: + return ["通用问题回答", "资料网络搜索"] - def _parse_tool_call( - self, tool_call: ChatCompletionMessageToolCallUnion - ) -> tuple[str, dict[str, Any]]: - """ - 解析Tool call,返回(tool_name, arguments) - """ - if tool_call.type != "function" and tool_call.type != "custom": - raise ValueError(f"不支持的工具调用类型{tool_call.type}") - if tool_call.type == "function": - tool_name = tool_call.function.name - raw_arguments = tool_call.function.arguments - else: - tool_name = tool_call.custom.name - raw_arguments = tool_call.custom.input - arguments = json.loads(raw_arguments) - if not isinstance(arguments, dict): - raise ValueError("工具参数必须是 JSON 对象") - return (tool_name, arguments) + @property + def tool_repository(self) -> ToolRepository: + return self._tool_repository - def _execute_tool_call( - self, tool_name: str, arguments: dict[str, Any], context: ConversationContext - ) -> Any: - """ - 执行tool call - """ - observation = self.tool_repos.call_tool(tool_name, arguments) - # 更新当前可用工具 - if tool_name == self.tool_repos.name: - available_tools = [] - for item in observation: - if item.name not in context.discovered_tool_names: - context.add_tool(item) - available_tools.append(self.tool_repos.tool_to_openai(item)) - observation = available_tools - return observation + @property + def system_prompt(self) -> str: + return """ + 你是一个能够调用外部工具的智能助手。 + 工作规则： + + 1. 首先理解用户的问题，判断是否需要外部工具。 + 2. 如果需要工具，可以从工具库中搜索你所需要的工具。 + 3. 从工具搜索结果中选择合适的工具并调用执行。 + 4. 工具执行后，根据返回结果判断下一步行动。 + 5. 当已经获得足够的信息，能够完整回答用户问题时，直接给出最终答案。 + + 注意事项： + + - 不允许编造工具执行结果。 + - 工具搜索没有结果时，可以尝试调整关键词重新搜索。 + - 如果工具无法完成任务，明确告知用户。 + - 不要为了调用工具而调用工具。 + - 只有在任务已经完成或明确无法继续时才结束回答。 + """ + + @override def run(self, question: str, context: ConversationContext) -> Iterator[Message]: self._initialize_context(context, question) run_id = str(uuid7()) - yield Message( - conversation_id=context.conversation_id, - role=MessageRole.USER, - run_id=run_id, - status=MessageStatus.RUNNING, - type=MessageType.TEXT, - content=question, - ) for _ in range(self.max_steps): - tools = [self.tool_repos.tool_to_openai(tool) for tool in context.discovered_tools] + tools = [tool.to_openai() for tool in context.discovered_tools] - assert self.model is not None - response = self.client.chat.completions.create( - model=self.model, + response = self.client.think( messages=context.messages, tools=tools, tool_choice="auto", @@ -166,54 +105,58 @@ class ReActAgent: yield Message( conversation_id=context.conversation_id, run_id=run_id, + run_status=RunStatus.COMPLETED, role=MessageRole.ASSISTANT, - content=message.content, - status=MessageStatus.COMPLETED, - type=MessageType.TEXT, + content=[TextContent(text=message.content)], ) return elif reasoning_content := getattr(message, "reasoning_content", None): yield Message( conversation_id=context.conversation_id, run_id=run_id, + run_status=RunStatus.RUNNING, role=MessageRole.ASSISTANT, - type=MessageType.THINK, - content=reasoning_content, - status=MessageStatus.RUNNING, + content=[ThinkContent(text=reasoning_content)], ) else: yield Message( conversation_id=context.conversation_id, run_id=run_id, + run_status=RunStatus.FAILED, role=MessageRole.ASSISTANT, - type=MessageType.TEXT, - status=MessageStatus.FAILED, - content="模型没有返回有效内容", + content=[ErrorContent(message="模型没有返回有效内容")], ) return if message.tool_calls: for tool_call in message.tool_calls: - tool_name = None + tool_id = None arguments = {} - tool_call_status = MessageStatus.RUNNING + tool_call_error: str | None = None try: - tool_name, arguments = self._parse_tool_call(tool_call) + tool_id, arguments = self._parse_tool_call(tool_call) + tool = self.tool_repository.get_tool(id=tool_id) + if not tool: + raise KeyError(f"{tool_id}工具不存在") yield Message( conversation_id=context.conversation_id, run_id=run_id, role=MessageRole.ASSISTANT, - status=MessageStatus.RUNNING, - type=MessageType.TOOL_CALL, - content=f"call tool:{tool_name}({arguments})", - tool_name=tool_name, - arguments=arguments, + run_status=RunStatus.RUNNING, + content=[ + ToolCallContent( + tool_call_id=tool_call.id, + id=tool.id, + name=tool.name, + arguments=arguments, + ) + ], ) - observation = self._execute_tool_call(tool_name, arguments, context) + observation = self._execute_tool_call(tool_id, arguments, context) except Exception as e: - tool_call_status = MessageStatus.FAILED + tool_call_error = str(e) observation = {"error": str(e)} # 统一记录工具执行结果 @@ -228,20 +171,21 @@ class ReActAgent: yield Message( conversation_id=context.conversation_id, run_id=run_id, + run_status=RunStatus.RUNNING, role=MessageRole.TOOL, - type=MessageType.TOOL_RESULT, - status=tool_call_status, - content=f"call {tool_name or 'unknown'} result:{self.serialize_json(observation)}", - tool_name=tool_name, - arguments=arguments, - tool_call_result=observation, + content=[ + ToolResultContent( + tool_call_id=tool_call.id, + result="执行成功" if tool_call_error is None else None, + error=tool_call_error, + ) + ], ) yield Message( conversation_id=context.conversation_id, run_id=run_id, + run_status=RunStatus.FAILED, role=MessageRole.ASSISTANT, - status=MessageStatus.FAILED, - type=MessageType.TEXT, - content=f"已达到最大执行步数 ({self.max_steps})，任务未完成", + content=[ErrorContent(message=f"已达到最大执行步数 ({self.max_steps})，任务未完成")], ) diff --git a/src/mini_agent/agent/route_agent.py b/src/mini_agent/agent/route_agent.py new file mode 100644 index 0000000..419e31e --- /dev/null +++ b/src/mini_agent/agent/route_agent.py @@ -0,0 +1,236 @@ +from collections.abc import Iterator +from typing import Literal, override +from uuid import uuid7 + +from pydantic import BaseModel, Field + +from mini_agent.agent.base_agent import BaseAgent +from mini_agent.agent.context import ConversationContext +from mini_agent.agent.llm_client import LLMClient +from mini_agent.agent.message import ( + ErrorContent, + Message, + MessageRole, + RunStatus, + TextContent, + ThinkContent, +) + + +class RouteDecision(BaseModel): + action: Literal["route", "direct", "clarify", "reject"] = Field(description="路由动作") + agent_id: str | None = Field(description="Agent id", default=None) + reason: str | None = Field(description="路由理由", default=None) + message: str | None = Field(description="回复用户的消息", default=None) + task: str | None = Field(description="交给子Agent的任务描述") + + +class RouteAgent(BaseAgent): + def __init__(self, llm_client: LLMClient, max_steps: int = 10) -> None: + super().__init__() + self.sub_agents: dict[str, BaseAgent] = {} + self.llm_client = llm_client + self.max_steps = max_steps + + def register_agent(self, agent: BaseAgent): + if agent.id not in self.sub_agents: + self.sub_agents[agent.id] = agent + else: + raise ValueError(f"{agent.id}已经存在") + + @property + def system_prompt(self) -> str: + template = """ + # 角色 + + 你是一个 AI Agent 路由调度器（Route Agent），负责理解用户请求，并将其分配给最合适的专业 Agent。 + + 你不直接处理用户的业务问题，不执行工具，不编造业务数据。你的核心职责是识别用户意图、选择目标 Agent，并生成清晰的路由决策。 + + # 核心职责 + + 1. 理解用户当前请求的真实意图。 + 2. 结合当前会话上下文，理解用户的追问、省略表达和上下文引用。 + 3. 从系统提供的 Agent 列表中选择最合适的 Agent。 + 4. 判断用户请求是否需要补充信息。 + 5. 判断当前请求是否可以直接回答，还是必须交给专业 Agent。 + 6. 输出符合指定结构的路由决策。 + + # 路由规则 + + ## 1. Agent 选择 + + - 必须从系统提供的可用 Agent 列表中选择目标 Agent。 + - 优先选择职责与用户请求最匹配的 Agent。 + - 不得虚构 Agent，不得调用未注册的 Agent。 + - 如果一个请求涉及多个领域，优先选择能够完成主要任务的 Agent。 + - 如果需要多个 Agent 协作，只有在系统明确支持多 Agent 调度时才能提出多个目标。 + + ## 2. 上下文理解 + + - 用户当前消息优先级最高，但不能脱离历史对话理解。 + - 如果用户使用“继续”“刚才那个”“再查一下”等表达，应结合历史消息确定实际任务。 + - 如果用户是在纠正上一轮结果，应将请求交给原任务对应的 Agent。 + - 如果用户明确提出新的任务，应重新进行路由，不要机械沿用上一轮 Agent。 + - 不要因为历史上使用过某个 Agent，就将无关的新问题继续分配给它。 + + ## 3. 信息不足 + + - 如果缺少执行任务所必需的信息，选择需要补充信息的处理方式。 + - 只询问真正影响任务执行的必要信息。 + - 不要为了补充非必要信息而阻断任务。 + - 如果可以使用合理默认值，应优先继续执行。 + + ## 4. 直接回答 + + 以下情况可以直接回答，不需要调用专业 Agent： + + - 简单的常识性问题。 + - 问候、感谢、告别等日常交流。 + - 对当前 Agent 工作状态的简单询问。 + - 不涉及外部数据查询的简单解释。 + - 用户只是要求修改、解释或讨论当前对话中已经提供的内容。 + + 直接回答时，不得伪装成调用了专业 Agent。 + + ## 5. 无法路由 + + 出现以下情况时，选择无法路由： + + - 用户意图无法理解。 + - 系统没有合适的 Agent。 + - 用户请求超出所有可用 Agent 的职责范围。 + - 需要执行的操作不被系统支持。 + + 应简要说明原因，不要虚构处理能力。 + + # 水利业务路由参考 + + 系统提供以下 Agent，参考这些Agent的职责做出选择： + + {} + + 以上只是路由参考，实际选择必须以系统提供的 Agent 定义为准。 + + # 输出要求 + + 必须输出符合系统指定 JSON Schema 的结构化结果。 + + 路由决策至少包含以下语义： + + - action：决策类型，可选 route、direct、clarify、reject。 + - agent_id：目标 Agent 标识。非 route 决策时为空。 + - reason：路由原因。 + - message：直接回复用户的内容，或者需要向用户询问的内容。 + - task：交给目标 Agent 的任务描述，尽可能保留用户原始需求和必要上下文。 + + 不要输出 JSON 以外的解释文字。 + + # 重要约束 + + 1. 你只负责路由，不负责执行专业任务。 + 2. 不得调用业务工具。 + 3. 不得虚构系统中不存在的 Agent。 + 4. 不得擅自修改用户需求。 + 5. 不得丢失用户请求中的关键参数。 + 6. 不要将工具调用需求误判为普通聊天。 + 7. 对于明确的业务查询，优先路由给专业 Agent。 + 8. 输出必须符合指定 JSON Schema，不允许输出额外字段。 + """ + agent_descriptions = [ + f"-{agent.id}: {','.join(agent.compatibility)}" for agent in self.sub_agents.values() + ] + return template.format("\n".join(agent_descriptions)) + + @override + def run(self, question: str, context: ConversationContext) -> Iterator[Message]: + self._initialize_context(context=context, question=question) + run_id = context.run_id + for _ in range(self.max_steps): + response = self.llm_client.think( + messages=context.messages, + tools = [], + tool_choice="auto", + response_format={ + "type": "json_schema", + "json_schema": { + "name": "agent_decision", + "strict": True, + "schema": RouteDecision.model_json_schema(), + }, + }, + ) + + message = response.choices[0].message + + # 保存模型返回消息 + message_param = self._build_message_param(message) + context.messages.append(message_param) + + if message.content: + decision = RouteDecision.model_validate_json(message.content) + if decision.action == "clarify" or decision.action == "direct": + yield Message( + conversation_id=context.conversation_id, + run_id=run_id, + run_status=RunStatus.COMPLETED, + role=MessageRole.ASSISTANT, + content=[TextContent(text=decision.message or "")], + ) + if decision.action == "direct": + return + elif decision.action == "route": + agent = self.sub_agents.get(decision.agent_id or "") + if not agent: + context.messages.append( + {"role": "user", "content": f"{decision.agent_id} Agent不存在"} + ) + else: + yield Message( + conversation_id=context.conversation_id, + run_id=context.run_id, + run_status=RunStatus.RUNNING, + role = MessageRole.ASSISTANT, + content= [ThinkContent(text=decision.reason or "正在路由请求")] + ) + agent_context = ConversationContext(conversation_id=context.conversation_id,run_id=str(uuid7())) + for agent_message in agent.run( + context=agent_context, question=decision.task or question + ): + yield Message( + conversation_id=context.conversation_id, + run_id=run_id, + run_status=agent_message.run_status, + content=agent_message.content, + role=agent_message.role, + ) + if ( + agent_message.run_status == RunStatus.COMPLETED + or agent_message.run_status == RunStatus.FAILED + ): + return + + elif reasoning_content := getattr(message, "reasoning_content", None): + yield Message( + conversation_id=context.conversation_id, + run_id=run_id, + run_status=RunStatus.RUNNING, + role=MessageRole.ASSISTANT, + content=[ThinkContent(text=reasoning_content)], + ) + else: + yield Message( + conversation_id=context.conversation_id, + run_id=run_id, + run_status=RunStatus.FAILED, + role=MessageRole.ASSISTANT, + content=[ErrorContent(message="模型没有返回有效内容")], + ) + return + yield Message( + conversation_id=context.conversation_id, + run_id=run_id, + run_status=RunStatus.FAILED, + role=MessageRole.ASSISTANT, + content=[ErrorContent(message=f"已达到最大执行步数 ({self.max_steps})，任务未完成")], + ) diff --git a/src/mini_agent/api/websocket.py b/src/mini_agent/api/websocket.py index 04c1fe8..897b3c8 100644 --- a/src/mini_agent/api/websocket.py +++ b/src/mini_agent/api/websocket.py @@ -3,7 +3,7 @@ from uuid import uuid7 from fastapi import APIRouter, WebSocket, WebSocketDisconnect from mini_agent.agent.context import ConversationContext -from mini_agent.agent.message import Message, MessageRole, MessageStatus, MessageType +from mini_agent.agent.message import Message, MessageRole, RunStatus, TextContent router = APIRouter() @@ -12,28 +12,36 @@ router = APIRouter() async def websocket_endpoint(websocket: WebSocket): await websocket.accept() agent = websocket.app.state.agent - context = ConversationContext(conversation_id=str(uuid7())) + context = ConversationContext(conversation_id=str(uuid7()),run_id="") try: while True: request = await websocket.receive_json() question = request.get("question") + context.run_id = str(uuid7()) + + await websocket.send_json(Message( + conversation_id=context.conversation_id, + role=MessageRole.USER, + run_id=context.run_id, + run_status=RunStatus.RUNNING, + content=[TextContent(text=question)], + ).model_dump(mode="json")) if not question: await websocket.send_json( Message( conversation_id=context.conversation_id, role=MessageRole.ASSISTANT, - run_id="", - type=MessageType.TEXT, - status=MessageStatus.FAILED, - content="请输入问题", + run_id=context.run_id, + run_status=RunStatus.FAILED, + content=[TextContent(text="请输入问题")], ).model_dump(mode="json") ) continue - for status in agent.run(question, context): - await websocket.send_json(status.model_dump(mode="json")) + for message in agent.run(question, context): + await websocket.send_json(message.model_dump(mode="json")) except WebSocketDisconnect: print("客户端断开连接") diff --git a/src/mini_agent/main.py b/src/mini_agent/main.py index f348f5d..16ddc72 100644 --- a/src/mini_agent/main.py +++ b/src/mini_agent/main.py @@ -1,12 +1,15 @@ from contextlib import asynccontextmanager from pathlib import Path +from uuid import uuid7 from dotenv import load_dotenv from fastapi import FastAPI from mini_agent.agent.context import ConversationContext -from mini_agent.agent.message import MessageStatus, MessageType +from mini_agent.agent.llm_client import LLMClient +from mini_agent.agent.message import TextContent, ToolCallContent, ToolResultContent from mini_agent.agent.react_agent import ReActAgent +from mini_agent.agent.route_agent import RouteAgent from mini_agent.api.websocket import router from mini_agent.tools.tool_repository import ToolRepository @@ -17,9 +20,10 @@ load_dotenv(PROJECT_ROOT / ".env") @asynccontextmanager async def lifespan(app: FastAPI): - app.state.agent = ReActAgent( - tool_repository=ToolRepository(), - ) + react_agent = ReActAgent(tool_repository=ToolRepository(), llm_client=LLMClient(),max_steps=50) + route_agent = RouteAgent(llm_client=LLMClient()) + route_agent.register_agent(react_agent) + app.state.agent = route_agent yield @@ -28,6 +32,7 @@ app = FastAPI(title="Mini Agent", lifespan=lifespan) app.include_router(router) + @app.get("/health") async def health(): return { @@ -37,24 +42,25 @@ async def health(): if __name__ == "__main__": - agent = ReActAgent( - tool_repository=ToolRepository(), - ) - context = ConversationContext(conversation_id="1") + agent = ReActAgent(tool_repository=ToolRepository(), llm_client=LLMClient()) + context = ConversationContext(conversation_id="1",run_id="") while True: question = input("提问：") + context.run_id = str(uuid7()) try: for message in agent.run(question=question, context=context): - status_type = message.type - match status_type: - case MessageType.TOOL_CALL: - print(f"🔧调用{message.tool_name} 参数 {message.arguments}") - case MessageType.TOOL_RESULT: - print(f"🔧调用{message.tool_name}结束") - case MessageType.TEXT: - if message.status == MessageStatus.FAILED: - print(f"❌{message.content}") - else: - print(f"ℹ{message.content}") + for content in message.content: + match content: + case TextContent(text=text): + print(f"💬{text}") + case ToolCallContent(name=name, arguments=arguments): + print(f"🔧调用工具{name},参数:{arguments}") + case ToolResultContent(error=error): + if error: + print(f"🔧调用工具失败:{error}") + else: + print("🔧调用工具成功") + case _: + pass except Exception as e: print(f"❌{e}") diff --git a/src/mini_agent/tools/tavily_search.py b/src/mini_agent/tools/tavily_search.py index aca9f8a..b508b59 100644 --- a/src/mini_agent/tools/tavily_search.py +++ b/src/mini_agent/tools/tavily_search.py @@ -21,6 +21,11 @@ class SearchResult(BaseModel): class TavilySearch(Tool[SearchResult]): + + @property + def id(self) -> str: + return "Tavily Search" + @property def name(self) -> str: return "Tavily网络搜索工具" diff --git a/src/mini_agent/tools/tool.py b/src/mini_agent/tools/tool.py index 059008f..1672afa 100644 --- a/src/mini_agent/tools/tool.py +++ b/src/mini_agent/tools/tool.py @@ -1,5 +1,6 @@ from abc import ABC, abstractmethod +from openai.types.chat import ChatCompletionToolParam from pydantic import BaseModel @@ -17,6 +18,11 @@ class Configuration(BaseModel): class Tool[T](ABC): + @property + @abstractmethod + def id(self) -> str: + pass + @property @abstractmethod def name(self) -> str: @@ -49,3 +55,44 @@ class Tool[T](ABC): @abstractmethod def run(self, **kwargs) -> T: pass + + def to_openai(self) -> ChatCompletionToolParam: + param_type_mapping = { + str: "string", + int: "integer", + float: "number", + bool: "boolean", + list: "array", + dict: "object", + } + properties = {} + required = [] + for parameter in self.parameters: + json_type = param_type_mapping.get(parameter.type) + + if json_type is None: + raise ValueError(f"不支持的参数类型: {parameter.type}") + + properties[parameter.name] = { + "type": json_type, + } + + if parameter.description: + properties[parameter.name]["description"] = parameter.description + + if parameter.required: + required.append(parameter.name) + + return { + "type": "function", + "function": { + "name": self.id, + "description": self.description.strip(), + "parameters": { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + }, + }, + } diff --git a/src/mini_agent/tools/tool_repository.py b/src/mini_agent/tools/tool_repository.py index c98f02f..e61b6ea 100644 --- a/src/mini_agent/tools/tool_repository.py +++ b/src/mini_agent/tools/tool_repository.py @@ -1,7 +1,5 @@ from typing import Any -from openai.types.chat import ChatCompletionToolParam - from mini_agent.tools.tool import Configuration, Parameter, Tool @@ -10,9 +8,13 @@ class ToolRepository(Tool[list[Tool]]): self._tools: dict[str, Tool] = {} self.register(self) + @property + def id(self) -> str: + return "tool repository" + @property def name(self) -> str: - return """tool repository""" + return "工具库" @property def description(self) -> str: @@ -47,65 +49,27 @@ class ToolRepository(Tool[list[Tool]]): return self.get_tools(keyword, catalog) def register[T](self, tool: Tool[T]) -> None: - if tool.name in self._tools: - raise ValueError(f"{tool.name} is already registered") - self._tools[tool.name] = tool + if tool.id in self._tools: + raise ValueError(f"{tool.id} is already registered") + self._tools[tool.id] = tool def get_tools(self, keyword: str, catalog: str | None) -> list[Tool]: return list(self._tools.values()) + def get_tool(self, id: str) -> Tool | None: + return self._tools.get(id) + def get_tools_for_openai(self, keyword: str, catalog: str | None) -> list[dict[str, Any]]: tools = self.get_tools(keyword, catalog) result = [] for tool in tools: - result.append(self.tool_to_openai(tool)) + result.append(tool.to_openai()) return result - def call_tool(self, name: str, arguments: dict[str, Any]) -> Any: - tool = self._tools.get(name, None) + def call_tool(self, id: str, arguments: dict[str, Any]) -> Any: + tool = self._tools.get(id, None) if tool is None: - raise ValueError(f"Tool not found: {name}") + raise ValueError(f"Tool not found: {id}") return tool.run(**arguments) - - def tool_to_openai[T](self, tool: Tool[T]) -> ChatCompletionToolParam: - param_type_mapping = { - str: "string", - int: "integer", - float: "number", - bool: "boolean", - list: "array", - dict: "object", - } - properties = {} - required = [] - for parameter in tool.parameters: - json_type = param_type_mapping.get(parameter.type) - - if json_type is None: - raise ValueError(f"不支持的参数类型: {parameter.type}") - - properties[parameter.name] = { - "type": json_type, - } - - if parameter.description: - properties[parameter.name]["description"] = parameter.description - - if parameter.required: - required.append(parameter.name) - - return { - "type": "function", - "function": { - "name": tool.name, - "description": tool.description.strip(), - "parameters": { - "type": "object", - "properties": properties, - "required": required, - "additionalProperties": False, - }, - }, - }