1、重构Agent代码

2、实现WebSocket连接服务
2026-11-03 16:50:18 +08:00
1 parent 2c1bac9bcb
commit 77ff26094d
@@ -1,8 +0,0 @@
-from mini_agent.react_agent import ReActAgent
-from mini_agent.tools.tool_manager import ToolManager
-
-if __name__ == "__main__":
-    agent = ReActAgent(tool_manager=ToolManager())
-    while True:
-        question = input("提问：")
-        print(agent.run(question))
@@ -0,0 +1,22 @@
+from openai.types.chat import ChatCompletionMessageParam
+from pydantic import BaseModel, Field, PrivateAttr
+
+from mini_agent.tools.tool import Tool
+
+
+class ConversationContext(BaseModel):
+    conversation_id: str = Field(description="会话ID")
+    messages: list[ChatCompletionMessageParam] = Field(description="消息历史", default_factory=list)
+
+    _discovered_tools: list[Tool] = PrivateAttr(default_factory=list)
+
+    discovered_tool_names: list[str] = Field(description="已发现的工具名称", default_factory=list)
+
+    @property
+    def discovered_tools(self):
+        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)
+            self.discovered_tools.append(tool)
@@ -0,0 +1,47 @@
+from datetime import datetime
+from enum import StrEnum
+from typing import Any
+from uuid import uuid4
+
+from pydantic import BaseModel, Field
+
+
+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):
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class Message(BaseModel):
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    conversation_id: str
+    run_id: str | None = None
+
+    role: MessageRole
+    type: MessageType
+
+    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
+
+    created_at: datetime = Field(default_factory=lambda: datetime.now())
@@ -0,0 +1,247 @@
+from collections.abc import Iterator
+import json
+import os
+from typing import Any
+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.context import ConversationContext
+from mini_agent.agent.message import Message, MessageRole, MessageStatus, MessageType
+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. 当已经获得足够的信息，能够完整回答用户问题时，直接给出最终答案。
+
+        注意事项：
+
+        - 不允许编造工具执行结果。
+        - 工具搜索没有结果时，可以尝试调整关键词重新搜索。
+        - 如果工具无法完成任务，明确告知用户。
+        - 不要为了调用工具而调用工具。
+        - 只有在任务已经完成或明确无法继续时才结束回答。
+        """
+
+    def __init__(
+        self,
+        tool_repository: ToolRepository,
+        model: str | None = None,
+        api_key: str | None = None,
+        base_url: str | None = None,
+        timeout: int | None = None,
+        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"))
+
+        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.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)
+
+    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)
+
+    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})
+
+    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
+
+    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)
+
+    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
+
+    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]
+
+            assert self.model is not None
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=context.messages,
+                tools=tools,
+                tool_choice="auto",
+            )
+
+            message = response.choices[0].message
+
+            # 保存模型返回消息
+            message_param = self._build_message_param(message)
+            context.messages.append(message_param)
+
+            # 模型没有调用工具，说明它选择直接回答
+            if not message.tool_calls:
+                if message.content:
+                    yield Message(
+                        conversation_id=context.conversation_id,
+                        run_id=run_id,
+                        role=MessageRole.ASSISTANT,
+                        content=message.content,
+                        status=MessageStatus.COMPLETED,
+                        type=MessageType.TEXT,
+                    )
+                    return
+                elif reasoning_content := getattr(message, "reasoning_content", None):
+                    yield Message(
+                        conversation_id=context.conversation_id,
+                        run_id=run_id,
+                        role=MessageRole.ASSISTANT,
+                        type=MessageType.THINK,
+                        content=reasoning_content,
+                        status=MessageStatus.RUNNING,
+                    )
+                else:
+                    yield Message(
+                        conversation_id=context.conversation_id,
+                        run_id=run_id,
+                        role=MessageRole.ASSISTANT,
+                        type=MessageType.TEXT,
+                        status=MessageStatus.FAILED,
+                        content="模型没有返回有效内容",
+                    )
+                    return
+
+            if message.tool_calls:
+                for tool_call in message.tool_calls:
+                    tool_name = None
+                    arguments = {}
+                    tool_call_status = MessageStatus.RUNNING
+                    try:
+                        tool_name, arguments = self._parse_tool_call(tool_call)
+                        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,
+                        )
+
+                        observation = self._execute_tool_call(tool_name, arguments, context)
+
+                    except Exception as e:
+                        tool_call_status = MessageStatus.FAILED
+                        observation = {"error": str(e)}
+
+                    # 统一记录工具执行结果
+                    context.messages.append(
+                        {
+                            "role": "tool",
+                            "tool_call_id": tool_call.id,
+                            "content": self.serialize_json(observation),
+                        }
+                    )
+
+                    yield Message(
+                        conversation_id=context.conversation_id,
+                        run_id=run_id,
+                        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,
+                    )
+
+        yield Message(
+            conversation_id=context.conversation_id,
+            run_id=run_id,
+            role=MessageRole.ASSISTANT,
+            status=MessageStatus.FAILED,
+            type=MessageType.TEXT,
+            content=f"已达到最大执行步数 ({self.max_steps})，任务未完成",
+        )
@@ -0,0 +1,39 @@
+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
+
+router = APIRouter()
+
+
+@router.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    agent = websocket.app.state.agent
+    context = ConversationContext(conversation_id=str(uuid7()))
+
+    try:
+        while True:
+            request = await websocket.receive_json()
+            question = request.get("question")
+
+            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="请输入问题",
+                    ).model_dump(mode="json")
+                )
+                continue
+
+            for status in agent.run(question, context):
+                await websocket.send_json(status.model_dump(mode="json"))
+
+    except WebSocketDisconnect:
+        print("客户端断开连接")
@@ -0,0 +1,60 @@
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+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.react_agent import ReActAgent
+from mini_agent.api.websocket import router
+from mini_agent.tools.tool_repository import ToolRepository
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+print(PROJECT_ROOT)
+load_dotenv(PROJECT_ROOT / ".env")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    app.state.agent = ReActAgent(
+        tool_repository=ToolRepository(),
+    )
+
+    yield
+
+
+app = FastAPI(title="Mini Agent", lifespan=lifespan)
+
+app.include_router(router)
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "agent": hasattr(app.state, "agent"),
+    }
+
+
+if __name__ == "__main__":
+    agent = ReActAgent(
+        tool_repository=ToolRepository(),
+    )
+    context = ConversationContext(conversation_id="1")
+    while True:
+        question = input("提问：")
+        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}")
+        except Exception as e:
+            print(f"❌{e}")
@@ -1,153 +0,0 @@
-import json
-import os
-from typing import Any
-
-from mini_agent.tools.tavily_search import TavilySearch
-from openai import OpenAI
-
-from mini_agent.tools.tool_search import ToolSearch
-from mini_agent.tools.tool_manager import ToolManager
-from dotenv import load_dotenv
-
-load_dotenv()
-
-class ReActAgent:
-    SYSTEM_PROMPT = """
-        你是一个能够调用外部工具的智能助手。
-
-        工作规则：
-
-        1. 首先理解用户的问题，判断是否需要外部工具。
-        2. 如果需要工具，调用 search_tools 搜索合适的工具。
-        3. 根据搜索结果选择合适的工具。
-        4. 工具执行后，根据返回结果判断下一步行动。
-        5. 如果需要其他工具，可以再次调用 search_tools。
-        6. 当已经获得足够的信息，能够完整回答用户问题时，直接给出最终答案。
-
-        注意事项：
-
-        - 不允许编造工具执行结果。
-        - 不允许调用尚未通过 search_tools 发现的业务工具。
-        - 工具搜索没有结果时，可以尝试调整关键词重新搜索。
-        - 如果工具无法完成任务，明确告知用户。
-        - 不要为了调用工具而调用工具。
-        - 只有在任务已经完成或明确无法继续时才结束回答。
-        """
-
-    def __init__(
-        self,
-        tool_manager: ToolManager,
-        model: str | None = None,
-        api_key: str | None = None,
-        base_url: str | None = None,
-        timeout: int | None = None,
-        max_steps: int = 10,
-        max_tool_calls: 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 or 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,
-        )
-
-        self.tool_manager = tool_manager
-        self.max_steps = max_steps
-        self.max_tool_calls = max_tool_calls
-        self.search_tool = ToolSearch(self.tool_manager)
-        self.tool_manager.register(self.search_tool)
-        web_search_tool = TavilySearch()
-        web_search_tool.config(**{"api_key":os.getenv("TAVILY_KEY","")})
-        self.tool_manager.register(web_search_tool)
-
-    def run(self, question: str) -> str:
-
-        # 已发现的业务工具
-
-        # 初始只开放工具发现能力
-        available_tools = [self.search_tool]
-        tool_calls_count = 0
-
-        messages = [
-            {"role":"user","content":question}
-        ]
-
-        for step in range(self.max_steps):
-
-            print(f"\n========== Step {step + 1} ==========")
-
-            tools = [self.tool_manager.tool_to_openai(tool) for tool in available_tools]
-
-            response = self.client.chat.completions.create(
-                model=self.model,
-                messages=messages,
-                tools=tools,
-                tool_choice="auto",
-            )
-
-            message = response.choices[0].message
-
-            # 模型没有调用工具，说明它选择直接回答
-            if not message.tool_calls:
-                if message.content:
-                    print(f"最终回答: {message.content}")
-                    return message.content
-                raise RuntimeError("模型没有返回有效内容")
-
-            # 保存模型的工具调用消息
-            messages.append(
-                message.model_dump(exclude_none=True)
-            )
-
-            for tool_call in message.tool_calls:
-                tool_name = tool_call.function.name
-                try:
-                    arguments = json.loads(
-                        tool_call.function.arguments or "{}"
-                    )
-                except json.JSONDecodeError:
-                    observation = json.dumps(
-                        {"error": "工具参数不是合法的 JSON"},
-                        ensure_ascii=False,
-                    )
-
-                    messages.append({
-                        "role": "tool",
-                        "tool_call_id": tool_call.id,
-                        "content": observation,
-                    })
-                    continue
-
-                print(f"调用工具: {tool_name}")
-                print(f"参数: {arguments}")
-
-
-                observation = self.tool_manager.call_tool(tool_name,arguments)
-                # 更新当前可用工具
-                if tool_name == self.search_tool.name:
-                    new_tools = []
-                    for item in observation:
-                        if not any(tool.name == item.name for tool in available_tools):
-                            available_tools.append(item)
-                            new_tools.append(item)
-                    observation = [self.tool_manager.tool_to_openai(tool) for tool in new_tools]
-
-                messages.append({
-                    "role": "tool",
-                    "tool_call_id": tool_call.id,
-                    "content": str(observation),
-                })
-
-        raise RuntimeError(
-            f"Agent 已达到最大执行步数 ({self.max_steps})，任务未完成"
-        )
@@ -1,27 +1,29 @@
-from mini_agent.tools.tool import Configuration, Tool,Parameter
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 from tavily import TavilyClient

+from mini_agent.tools.tool import Configuration, Parameter, Tool
+
+
 class ResultItem(BaseModel):
-    url:str
-    title:str
-    content:str
-    score:float
-    raw_content:str | None = None
-    id:str
+    url: str
+    title: str
+    content: str
+    score: float
+    raw_content: str | None = None
+    id: str


 class SearchResult(BaseModel):
-    query:str
-    answer:str | None
-    images:list[str] = []
+    query: str
+    answer: str | None
+    images: list[str] = []
    results: list[ResultItem]

-class TavilySearch(Tool[SearchResult]):

+class TavilySearch(Tool[SearchResult]):
    @property
-    def name(self)->str:
-        return "Tavily网络搜索工具 "
+    def name(self) -> str:
+        return "Tavily网络搜索工具"

    @property
    def description(self) -> str:
@@ -36,25 +38,27 @@ class TavilySearch(Tool[SearchResult]):

    @property
    def configurations(self) -> list[Configuration]:
-        return [
-            Configuration(name="api_key",type="str",required=True)
-        ]
+        return [Configuration(name="api_key", type="str", required=True)]

    @property
    def parameters(self) -> list[Parameter]:
        return [
-            Parameter(name="query",type="str",description="query keyword",required=True)
+            Parameter(
+                name="query", type=str, description="query keyword", required=True
+            )
        ]

-    def config(self,**kwargs) -> None:
-        api_key = kwargs.get("api_key",None)
+    def config(self, **kwargs) -> None:
+        api_key = kwargs.get("api_key")
        if not api_key:
            raise ValueError("API Key is required")
        self.client = TavilyClient(api_key=api_key)

    def run(self, **kwargs) -> SearchResult:
-        query = kwargs.get("query",None)
+        query = kwargs.get("query")
        if not query:
            raise ValueError("查询参数不能为空")
-        response = self.client.search(query=query,search_depth="advanced",include_answer="advanced")
+        response = self.client.search(
+            query=query, search_depth="advanced", include_answer="advanced"
+        )
        return SearchResult.model_validate(response)
@@ -1,49 +1,51 @@
+from abc import ABC, abstractmethod
+
 from pydantic import BaseModel
-from abc import ABC,abstractmethod


 class Parameter(BaseModel):
-    name:str
-    type:str
-    description:str
-    required:bool =True
+    name: str
+    type: type
+    description: str
+    required: bool = True
+

 class Configuration(BaseModel):
-    name:str
-    type:str
-    required:bool = True
+    name: str
+    type: str
+    required: bool = True
+

 class Tool[T](ABC):
-
    @property
    @abstractmethod
-    def name(self)->str:
+    def name(self) -> str:
        pass

    @property
    @abstractmethod
-    def description(self)->str:
+    def description(self) -> str:
        pass

    @property
    @abstractmethod
-    def catalog(self)->str:
+    def catalog(self) -> str:
        pass

    @property
    @abstractmethod
-    def parameters(self)->list[Parameter]:
+    def parameters(self) -> list[Parameter]:
        pass

    @property
    @abstractmethod
-    def configurations(self)->list[Configuration]:
+    def configurations(self) -> list[Configuration]:
        pass

    @abstractmethod
-    def config(self,**kwargs)->None:
+    def config(self, **kwargs) -> None:
        pass

    @abstractmethod
-    def run(self,**kwargs)->T:
+    def run(self, **kwargs) -> T:
        pass
@@ -1,80 +0,0 @@
-from typing import Any
-
-from mini_agent.tools.tool import Tool
-
-
-class ToolManager:
-
-    _tools:dict[str,Tool] = {}
-
-    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
-
-    def get_tools(self,keyword:str,catalog:str | None)->list[Tool]:
-        return list(self._tools.values())
-
-    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))
-        return result
-
-    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
-        tool = self._tools.get(name, None)
-
-        if tool is None:
-            raise ValueError(f"Tool not found: {name}")
-
-        return tool.run(**arguments)
-
-    def tool_to_openai[T](self,tool:Tool[T])->dict[str,Any]:
-        type_mapping = {
-                    "str": "string",
-                    "string": "string",
-                    "int": "integer",
-                    "integer": "integer",
-                    "float": "number",
-                    "number": "number",
-                    "bool": "boolean",
-                    "boolean": "boolean",
-                    "list": "array",
-                    "dict": "object",
-                }
-        properties = {}
-        required = []
-        for parameter in tool.parameters:
-            json_type = 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,
-                },
-            },
-        }
@@ -0,0 +1,111 @@
+from typing import Any
+
+from openai.types.chat import ChatCompletionToolParam
+
+from mini_agent.tools.tool import Configuration, Parameter, Tool
+
+
+class ToolRepository(Tool[list[Tool]]):
+    def __init__(self):
+        self._tools: dict[str, Tool] = {}
+        self.register(self)
+
+    @property
+    def name(self) -> str:
+        return """tool repository"""
+
+    @property
+    def description(self) -> str:
+        return """
+        "根据用户需求搜索可用工具。 当需要外部数据、查询业务信息或执行特定操作时，首先使用该工具查找合适的业务工具。可以根据需要多次搜索不同的工具。
+        """
+
+    @property
+    def catalog(self) -> str:
+        return "system"
+
+    def config(self, **kwargs) -> None:
+        pass
+
+    @property
+    def configurations(self) -> list[Configuration]:
+        return []
+
+    @property
+    def parameters(self) -> list[Parameter]:
+        return [
+            Parameter(name="keyword", type=str, description="关键词", required=True),
+            Parameter(name="catalog", type=str, description="工具分类", required=False),
+        ]
+
+    def run(self, **kwargs) -> list[Tool]:
+        keyword = kwargs.get("keyword")
+        catalog = kwargs.get("catalog")
+        if not keyword:
+            raise ValueError("keyword is required for query")
+
+        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
+
+    def get_tools(self, keyword: str, catalog: str | None) -> list[Tool]:
+        return list(self._tools.values())
+
+    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))
+        return result
+
+    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
+        tool = self._tools.get(name, None)
+
+        if tool is None:
+            raise ValueError(f"Tool not found: {name}")
+
+        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,
+                },
+            },
+        }
@@ -1,43 +0,0 @@
-from mini_agent.tools.tool import Tool,Parameter,Configuration
-from mini_agent.tools.tool_manager import ToolManager
-
-class ToolSearch(Tool[list[Tool]]):
-
-    def __init__(self,tool_manager:ToolManager) -> None:
-        self.tool_manager = tool_manager
-
-    @property
-    def name(self)->str:
-        return """search_tools"""
-
-    @property
-    def description(self) -> str:
-        return """
-        "根据用户需求搜索可用工具。 当需要外部数据、查询业务信息或执行特定操作时，首先使用该工具查找合适的业务工具。可以根据需要多次搜索不同的工具。
-        """
-
-    @property
-    def catalog(self) -> str:
-        return "system"
-
-    def config(self, **kwargs) -> None:
-        pass
-
-    @property
-    def configurations(self) -> list[Configuration]:
-        return []
-
-    @property
-    def parameters(self) -> list[Parameter]:
-        return [
-            Parameter(name="keyword",type="str",description="关键词",required=True),
-            Parameter(name="catalog",type="str",description="工具分类",required=False)
-        ]
-
-    def run(self, **kwargs) -> list[Tool]:
-        keyword = kwargs.get("keyword", None)
-        catalog = kwargs.get("catalog", None)
-        if not keyword:
-            raise ValueError("keyword is required for query")
-
-        return self.tool_manager.get_tools(keyword,catalog)