初步实现Route Agent

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