diff --git a/agents.py b/agents.py index 5a8526d..135cd06 100644 --- a/agents.py +++ b/agents.py @@ -5,7 +5,7 @@ import sys from abc import ABC, abstractmethod from dataclasses import dataclass from threading import Event -from typing import Callable, Optional, Sequence +from typing import Callable, Mapping, Optional ToolInvoker = Callable[[str], bool] @@ -83,7 +83,9 @@ class LoopAgent(Agent): break -def build_agent(spec: str, *, allowed_tools: Optional[Sequence[str]] = None) -> Agent: +def build_agent( + spec: str, *, allowed_tools: Optional[Mapping[str, str]] = None +) -> Agent: normalized = spec.strip() if not normalized: raise RuntimeError("Agent specification must not be empty") @@ -123,8 +125,11 @@ def build_agent(spec: str, *, allowed_tools: Optional[Sequence[str]] = None) -> else: instruction = segment - whitelist = tuple(sorted({tool.lower() for tool in (allowed_tools or [])})) - agent = IntelligentAgent(instruction=instruction, allowed_tools=whitelist) + tool_map = { + name: description + for name, description in (allowed_tools or {}).items() + } + agent = IntelligentAgent(instruction=instruction, allowed_tools=tool_map) if model: agent.model = model agent.turn_delay = turn_delay if turn_delay > 0 else 1.0 diff --git a/app.py b/app.py index c4d5cdf..b1f2b7d 100644 --- a/app.py +++ b/app.py @@ -11,32 +11,58 @@ from tools import Tool, SimpleTool from agents import Agent, build_agent from agent_runtime import run_agent -BUILTIN_TOOLS = { - "simple": ("tools", "SimpleTool", {}), - "explore": ("tools", "ExploreTool", {}), - "communication": ("tools", "CommunicationTool", {}), - "movement": ("movement_tool", "MovementTool", {}), - "move": ("movement_tool", "MovementTool", {}), - "intelligent": ( - "intelligent_tool", - "IntelligentCommunicationTool", - {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, - ), - "intelligentcommunication": ( - "intelligent_tool", - "IntelligentCommunicationTool", - {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, - ), +TOOL_REGISTRY = { + "simple": { + "module": "tools", + "class": "SimpleTool", + "kwargs": {}, + "description": "Sends the 'schau' look command to refresh the room description.", + }, + "move": { + "module": "movement_tool", + "class": "MovementTool", + "kwargs": {}, + "description": "Looks around and moves in one available direction, chosen randomly among unvisited exits.", + }, + "movement": { + "module": "movement_tool", + "class": "MovementTool", + "kwargs": {}, + "description": "Alias of 'move'. Looks around and moves in one available direction, chosen randomly among unvisited exits.", + }, + "explore": { + "module": "tools", + "class": "ExploreTool", + "kwargs": {}, + "description": "Sends 'schau' once, then 'untersuche ' for each noun found in the room description.", + }, + "communication": { + "module": "tools", + "class": "CommunicationTool", + "kwargs": {}, + "description": "Responds to private tells with a friendly greeting via 'teile mit ...'.", + }, + "intelligent": { + "module": "intelligent_tool", + "class": "IntelligentCommunicationTool", + "kwargs": { + "model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407") + }, + "description": "Uses an LLM to craft a polite reply to private tells.", + }, + "intelligentcommunication": { + "module": "intelligent_tool", + "class": "IntelligentCommunicationTool", + "kwargs": { + "model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407") + }, + "description": "Alias of 'intelligent'. Uses an LLM to craft a polite reply to private tells.", + }, } -AVAILABLE_TOOL_NAMES = [ - "move", - "movement", - "explore", - "communication", - "intelligentcommunication", - "simple", -] +TOOL_DESCRIPTIONS = { + name: meta["description"] for name, meta in TOOL_REGISTRY.items() +} from telnetclient import TelnetClient @@ -190,8 +216,11 @@ def build_tool(spec: str) -> Tool: if key == "simple": return SimpleTool() - if key in BUILTIN_TOOLS: - module_name, class_name, kwargs = BUILTIN_TOOLS[key] + if key in TOOL_REGISTRY: + meta = TOOL_REGISTRY[key] + module_name = meta["module"] + class_name = meta["class"] + kwargs = meta.get("kwargs", {}) try: module = import_module(module_name) tool_cls = getattr(module, class_name) @@ -407,7 +436,7 @@ def main() -> int: print("[Agent] Usage: #agent ") return try: - temp_agent = build_agent(spec, allowed_tools=AVAILABLE_TOOL_NAMES) + temp_agent = build_agent(spec, allowed_tools=TOOL_DESCRIPTIONS) except RuntimeError as exc: print(f"[Agent] Failed to configure '{spec}': {exc}", file=sys.stderr) return @@ -486,22 +515,3 @@ def main() -> int: if __name__ == "__main__": raise SystemExit(main()) -BUILTIN_TOOLS = { - "simple": ("tools", "SimpleTool", {}), - "explore": ("tools", "ExploreTool", {}), - "communication": ("tools", "CommunicationTool", {}), - "movement": ("movement_tool", "MovementTool", {}), - "move": ("movement_tool", "MovementTool", {}), - "intelligent": ( - "intelligent_tool", - "IntelligentCommunicationTool", - {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, - ), - "intelligentcommunication": ( - "intelligent_tool", - "IntelligentCommunicationTool", - {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, - ), -} - -AVAILABLE_TOOL_NAMES = sorted(set(BUILTIN_TOOLS.keys())) diff --git a/intelligent_agent.py b/intelligent_agent.py index f7e7d03..9d15aa6 100644 --- a/intelligent_agent.py +++ b/intelligent_agent.py @@ -32,7 +32,7 @@ class IntelligentAgent(Agent): max_output_tokens: int = 200 instruction: str = "" turn_delay: float = 0.0 - allowed_tools: tuple[str, ...] = () + allowed_tools: dict[str, str] = field(default_factory=dict) history: Deque[dict[str, str]] = field(default_factory=deque, init=False) def observe(self, message: str) -> None: @@ -60,13 +60,15 @@ class IntelligentAgent(Agent): messages.append({"role": "system", "content": self.instruction}) messages.extend(self.history) if self.allowed_tools: - tool_list = ", ".join(self.allowed_tools) + tool_list = "\n".join( + f"- {name}: {desc}" for name, desc in self.allowed_tools.items() + ) messages.append( { "role": "system", "content": ( - "Available tools: " - f"{tool_list}. Use only these tool names when type=tool; do not invent new tools." + "Available tools (only use these names when type=tool):\n" + f"{tool_list}" ), } ) @@ -129,14 +131,20 @@ class IntelligentAgent(Agent): print("[Agent] LLM returned empty action", file=sys.stderr) return - allowed_map = {tool.lower(): tool for tool in self.allowed_tools} + allowed_map = { + name.lower(): (name, desc) + for name, desc in self.allowed_tools.items() + } if action_type == "tool": lower = value.lower() if allowed_map and lower not in allowed_map: - print(f"[Agent] Tool '{value}' not in allowed list {self.allowed_tools}", file=sys.stderr) + print( + f"[Agent] Tool '{value}' not in allowed list {list(self.allowed_tools)}", + file=sys.stderr, + ) return - canonical = allowed_map.get(lower, value) + canonical, _ = allowed_map.get(lower, (value, "")) success = invoke_tool(canonical) print(f"[Agent] Executed tool: {canonical} (success={success})") self.history.append({"role": "assistant", "content": f"TOOL {canonical}"}) @@ -149,56 +157,59 @@ class IntelligentAgent(Agent): self.history.append({"role": "assistant", "content": f"COMMAND {value}"}) self._trim_history() elif action_type == "end": - print(f"[Agent] Execution ended by LLM.") - self.history.append({"role": "assistant", "content": f"END"}) + print("[Agent] LLM requested to end the session.") + self.history.append({"role": "assistant", "content": "END"}) self._trim_history() stop_event.set() + break else: print(f"[Agent] Unknown action type '{action_type}'", file=sys.stderr) return - if self.turn_delay > 0 and stop_event.wait(self.turn_delay): - return + if self.turn_delay > 0 and stop_event.wait(self.turn_delay): + break - messages = [{"role": "system", "content": self.system_prompt}] - if self.instruction: - messages.append({"role": "system", "content": self.instruction}) - messages.extend(self.history) - if self.allowed_tools: - tool_list = ", ".join(self.allowed_tools) + messages = [{"role": "system", "content": self.system_prompt}] + if self.instruction: + messages.append({"role": "system", "content": self.instruction}) + messages.extend(self.history) + if self.allowed_tools: + tool_list = "\n".join( + f"- {name}: {desc}" for name, desc in self.allowed_tools.items() + ) + messages.append( + { + "role": "system", + "content": ( + "Available tools (only use these names when type=tool):\n" + f"{tool_list}" + ), + } + ) messages.append( { "role": "system", "content": ( - "Available tools: " - f"{tool_list}. Use only these tool names when type=tool; do not invent new tools." + "Respond with JSON only. Schema: {\n" + " \"type\": \"tool\" or \"command\" or \"end\",\n" + " \"value\": string,\n" + " \"notes\": optional string\n}""" ), } ) - messages.append( - { - "role": "system", - "content": ( - "Respond with JSON only. Schema: {\n" - " \"type\": \"tool\" or \"command\" or \"end\",\n" - " \"value\": string,\n" - " \"notes\": optional string\n}""" - ), - } - ) - if not self.history or self.history[-1]["role"] != "assistant": + if not self.history or self.history[-1]["role"] != "assistant": + messages.append( + { + "role": "assistant", + "content": "I will decide the next action now.", + } + ) messages.append( { - "role": "assistant", - "content": "I will decide the next action now.", + "role": "user", + "content": "What is the next action you will take?", } ) - messages.append( - { - "role": "user", - "content": "What is the next action you will take?", - } - ) print("[Agent] Intelligent agent finished.")