Upload mythos/agent.py

mythos/agent.py

@@ -0,0 +1,139 @@
+"""Agent definition for Mythos — role-based reasoning units."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable, Optional
+
+from pydantic import BaseModel, Field
+
+from .memory import Memory, MemoryType
+from .message import Message, MessageType
+from .tool import ToolRegistry
+
+
+class AgentConfig(BaseModel):
+    """Configuration for an agent."""
+
+    max_context_length: int = 4096
+    temperature: float = 0.7
+    max_tokens: int = 1024
+    auto_memory: bool = True          # Automatically store interactions as memory
+    verbose: bool = False
+
+
+class Agent(BaseModel):
+    """A Mythos agent — a role-based reasoning unit with memory and tools."""
+
+    name: str
+    role: str
+    goal: str
+    backstory: str = ""
+    config: AgentConfig = Field(default_factory=AgentConfig)
+    memory: Memory = Field(default_factory=lambda: Memory(agent_name=""))
+    tools: ToolRegistry = Field(default_factory=ToolRegistry)
+    _llm_call: Optional[Callable[[str], Awaitable[str]]] = None
+    _message_handler: Optional[Callable[[Message], Awaitable[Message]]] = None
+
+    class Config:
+        arbitrary_types_allowed = True
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        self.memory.agent_name = self.name
+
+    def set_llm(self, llm_call: Callable[[str], Awaitable[str]]) -> None:
+        """Set the LLM inference function for this agent."""
+        self._llm_call = llm_call
+
+    def set_message_handler(self, handler: Callable[[Message], Awaitable[Message]]) -> None:
+        """Set a custom message handler."""
+        self._message_handler = handler
+
+    def _build_system_prompt(self) -> str:
+        """Build the system prompt for this agent."""
+        lines = [
+            f"You are {self.name}, a {self.role}.",
+            f"Your goal: {self.goal}",
+        ]
+        if self.backstory:
+            lines.append(f"Backstory: {self.backstory}")
+        if self.tools.list_tools():
+            lines.append("\nAvailable tools:")
+            for spec in self.tools.list_tools():
+                lines.append(f"  - {spec.name}: {spec.description}")
+        lines.append("\nRespond helpfully and concisely.")
+        return "\n".join(lines)
+
+    def _build_prompt(self, message: Message, context: str = "") -> str:
+        """Build the full prompt for a message."""
+        parts = [self._build_system_prompt()]
+        if context:
+            parts.append(f"\n--- Context ---\n{context}\n---")
+        parts.append(f"\n[{message.sender}] says: {message.content}")
+        parts.append(f"\n[{self.name}] responds:")
+        return "\n".join(parts)
+
+    async def think(self, content: str, context: str = "") -> str:
+        """Generate a response to content using the LLM."""
+        if self._llm_call is None:
+            # Fallback: echo with role context
+            return f"[{self.name} ({self.role})] Acknowledged: {content[:200]}"
+
+        prompt = self._build_system_prompt()
+        if context:
+            prompt += f"\n\n--- Context ---\n{context}\n---"
+        prompt += f"\n\nUser: {content}\n\n{self.name}:"
+
+        response = await self._llm_call(prompt)
+        return response.strip()
+
+    async def handle(self, message: Message) -> Message:
+        """Handle an incoming message and return a response."""
+        if self._message_handler:
+            return await self._message_handler(message)
+
+        # Auto-retrieve relevant memories
+        memory_context = ""
+        if self.config.auto_memory:
+            relevant = self.memory.active_retrieve(message.content)
+            if relevant:
+                memory_context = self.memory.to_context(relevant)
+
+        # Generate response
+        response_text = await self.think(message.content, context=memory_context)
+
+        # Store interaction in memory
+        if self.config.auto_memory:
+            self.memory.add(
+                content=f"Received from {message.sender}: {message.content}",
+                mem_type=MemoryType.EPISODIC,
+                tags=["interaction", message.sender],
+            )
+            self.memory.add(
+                content=f"Responded: {response_text}",
+                mem_type=MemoryType.EPISODIC,
+                tags=["interaction", message.sender],
+            )
+
+        return Message(
+            type=MessageType.RESPONSE,
+            sender=self.name,
+            recipient=message.sender,
+            content=response_text,
+            parent_id=message.id,
+        )
+
+    async def run_task(self, task: str) -> str:
+        """Run a standalone task without a message context."""
+        msg = Message(
+            type=MessageType.TASK,
+            sender="system",
+            recipient=self.name,
+            content=task,
+        )
+        response = await self.handle(msg)
+        return response.content
+
+    def __str__(self) -> str:
+        return f"Agent({self.name}, role={self.role})"