diff --git a/.vscode/settings.json b/.vscode/settings.json index 082291d..e1fa739 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,3 +1,14 @@ { - "wolf.disableHotModeWarning": true + "wolf.disableHotModeWarning": true, + "cSpell.words": [ + "behaviours", + "dotenv", + "fixedstrategy", + "intelligentcommunication", + "litellm", + "MISTLE", + "Mudbot", + "telnetclient", + "telnetlib" + ] } \ No newline at end of file diff --git a/README.md b/README.md index 262d5d8..594f2c4 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows, - Loads credentials and connection settings from a local `.env` file. - Interactive console session that mirrors server output and lets you type commands directly. - Optional always-on tool mode plus an on-demand `#execute ` escape hatch for ad-hoc automations. -- Higher-level agents (`fixed`, `loop`) that can string multiple tools together via `#agent `. +- Higher-level agents (`fixed`, `loop`, `intelligent`) that can string multiple tools together via `#agent `. - Built-in tools (`SimpleTool`, `ExploreTool`, `CommunicationTool`, `MovementTool`, `IntelligentCommunicationTool`) with a pluggable interface for custom behaviours. ## Requirements @@ -58,6 +58,14 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows, Append `:delay` to pause between iterations, e.g. `#agent loop move,explore:2.5`. +8. To let the LLM decide the next action, use the intelligent agent: + + ```text + #agent intelligent + ``` + + Add guidance text after the type and optional modifiers separated by `|`, e.g. `#agent intelligent explore carefully|model=mistral/mistral-large-2407|delay=2`. + ## Environment Variables All variables can be placed in the `.env` file (one `KEY=value` per line) or provided through the shell environment. diff --git a/agent_runtime.py b/agent_runtime.py index f3a6387..1aafe71 100644 --- a/agent_runtime.py +++ b/agent_runtime.py @@ -9,6 +9,7 @@ from tools import Tool ToolBuilder = Callable[[str], Tool] ToolRunner = Callable[[Tool], None] +CommandRunner = Callable[[str], None] def run_agent( @@ -16,6 +17,7 @@ def run_agent( *, build_tool: ToolBuilder, run_tool: ToolRunner, + send_command: CommandRunner, stop_event: Event, ) -> None: """Execute *agent* by wiring it to the tool runtime.""" @@ -34,4 +36,12 @@ def run_agent( run_tool(tool) return True - agent.run(invoke_tool=invoke_tool, stop_event=stop_event) + try: + agent.run( + invoke_tool=invoke_tool, + send_command=send_command, + stop_event=stop_event, + ) + except TypeError: + # Backwards compatibility for agents that only accept invoke_tool + agent.run(invoke_tool=invoke_tool, stop_event=stop_event) diff --git a/agents.py b/agents.py index fdb3119..5cb5605 100644 --- a/agents.py +++ b/agents.py @@ -1,21 +1,28 @@ from __future__ import annotations +import itertools import sys from abc import ABC, abstractmethod -import itertools from dataclasses import dataclass from threading import Event -from typing import Callable +from typing import Callable, Optional ToolInvoker = Callable[[str], bool] +CommandExecutor = Callable[[str], None] class Agent(ABC): """Interface for higher-level behaviours that orchestrate tools.""" @abstractmethod - def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None: + def run( + self, + *, + invoke_tool: ToolInvoker, + stop_event: Event, + send_command: Optional[CommandExecutor] = None, + ) -> None: """Execute the agent strategy until finished or *stop_event* is set.""" @@ -25,7 +32,13 @@ class FixedStrategyAgent(Agent): plan: str - def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None: + def run( + self, + *, + invoke_tool: ToolInvoker, + stop_event: Event, + send_command: Optional[CommandExecutor] = None, + ) -> None: steps = [part.strip() for part in self.plan.split(",") if part.strip()] if not steps: print("[Agent] No tools configured for fixed strategy", file=sys.stderr) @@ -47,7 +60,13 @@ class LoopAgent(Agent): plan: str delay: float = 0.0 - def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None: + def run( + self, + *, + invoke_tool: ToolInvoker, + stop_event: Event, + send_command: Optional[CommandExecutor] = None, + ) -> None: steps = [part.strip() for part in self.plan.split(",") if part.strip()] if not steps: print("[Agent] No tools configured for loop strategy", file=sys.stderr) @@ -85,5 +104,29 @@ def build_agent(spec: str) -> Agent: except ValueError: print(f"[Agent] Invalid delay '{delay_str}', defaulting to 0") return LoopAgent(config, delay=delay) + if kind in {"intelligent", "llm"}: + from intelligent_agent import IntelligentAgent + + instruction = "" + model = None + turn_delay = 0.0 + if config: + segments = [segment.strip() for segment in config.split("|") if segment.strip()] + for segment in segments: + if segment.startswith("model="): + model = segment.split("=", 1)[1].strip() + elif segment.startswith("delay="): + try: + turn_delay = float(segment.split("=", 1)[1].strip()) + except ValueError: + print(f"[Agent] Invalid delay '{segment}'", file=sys.stderr) + else: + instruction = segment + + agent = IntelligentAgent(instruction=instruction) + if model: + agent.model = model + agent.turn_delay = turn_delay if turn_delay > 0 else 1.0 + return agent raise RuntimeError(f"Unknown agent type '{kind}'") diff --git a/app.py b/app.py index cfe98b4..6c211f1 100644 --- a/app.py +++ b/app.py @@ -402,6 +402,15 @@ def main() -> int: print(f"[Agent] Failed to configure '{spec}': {exc}", file=sys.stderr) return + last_output = state.snapshot_output() + if last_output: + observe = getattr(temp_agent, "observe", None) + if callable(observe): + try: + observe(last_output) + except Exception as exc: # pragma: no cover - defensive + print(f"[Agent] observe failed: {exc}", file=sys.stderr) + def run_tool_instance(tool: Tool) -> None: run_tool_loop( client, @@ -412,6 +421,14 @@ def main() -> int: auto_stop=True, auto_stop_idle=2.0, ) + output_after = state.snapshot_output() + if output_after: + observe = getattr(temp_agent, "observe", None) + if callable(observe): + try: + observe(output_after) + except Exception as exc: # pragma: no cover - defensive + print(f"[Agent] observe failed: {exc}", file=sys.stderr) thread = Thread( target=run_agent, @@ -419,6 +436,7 @@ def main() -> int: kwargs={ "build_tool": build_tool, "run_tool": run_tool_instance, + "send_command": lambda cmd: state.send(client, cmd), "stop_event": stop_event, }, daemon=True, diff --git a/intelligent_agent.py b/intelligent_agent.py new file mode 100644 index 0000000..ebfc21a --- /dev/null +++ b/intelligent_agent.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass, field +from threading import Event +from typing import Callable, Deque + +from collections import deque + +try: + from litellm import completion +except ImportError: # pragma: no cover + completion = None # type: ignore[assignment] + +from agents import Agent + +ToolInvoker = Callable[[str], bool] +CommandExecutor = Callable[[str], None] + + +@dataclass +class IntelligentAgent(Agent): + """LLM-driven agent that decides between tools and raw commands.""" + + model: str = "mistral/mistral-large-2407" + system_prompt: str = ( + "You are Mistle, a helpful MUD assistant. " + "You can either call tools or send plain commands to the MUD." + ) + temperature: float = 0.7 + max_output_tokens: int = 200 + instruction: str = "" + turn_delay: float = 0.0 + history: Deque[dict[str, str]] = field(default_factory=deque, init=False) + + def observe(self, message: str) -> None: + content = message.strip() + if not content: + return + self.history.append({"role": "user", "content": content}) + self._trim_history() + + def run( + self, + *, + invoke_tool: ToolInvoker, + send_command: CommandExecutor | None, + stop_event: Event, + ) -> None: + if send_command is None: + raise RuntimeError("IntelligentAgent requires send_command support") + if completion is None: + print("[Agent] litellm not available; intelligent agent disabled", file=sys.stderr) + return + + messages = [{"role": "system", "content": self.system_prompt}] + if self.instruction: + messages.append({"role": "system", "content": self.instruction}) + messages.extend(self.history) + messages.append( + { + "role": "system", + "content": ( + "Respond with JSON only. Schema: {\n" + " \"type\": \"tool\" or \"command\",\n" + " \"value\": string (tool name or raw command),\n" + " \"notes\": optional string explanation\n}""" + ), + } + ) + 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": "user", + "content": "What is the next action you will take?", + } + ) + cycle = 0 + while not stop_event.is_set(): + cycle += 1 + print(f"[Agent] LLM cycle {cycle}...") + try: + response = completion( + model=self.model, + messages=messages, + temperature=self.temperature, + max_tokens=self.max_output_tokens, + ) + except Exception as exc: # pragma: no cover + print(f"[Agent] LLM call failed: {exc}", file=sys.stderr) + return + + try: + content = response["choices"][0]["message"]["content"].strip() + print(f"[Agent] LLM raw output: {content}") + if content.startswith("```"): + content = content.strip("` ") + if "\n" in content: + content = content.split("\n", 1)[1] + payload = json.loads(content) + except (KeyError, IndexError, TypeError, json.JSONDecodeError) as exc: + print(f"[Agent] Invalid LLM response: {exc}", file=sys.stderr) + print(f"[Agent] Raw content: {content}", file=sys.stderr) + return + + action_type = payload.get("type") + value = (payload.get("value") or "").strip() + notes = payload.get("notes") + if notes: + self.history.append({"role": "assistant", "content": f"NOTE: {notes}"}) + self._trim_history() + if not value: + print("[Agent] LLM returned empty action", file=sys.stderr) + return + + if action_type == "tool": + success = invoke_tool(value) + print(f"[Agent] Executed tool: {value} (success={success})") + self.history.append({"role": "assistant", "content": f"TOOL {value}"}) + self._trim_history() + if not success: + return + elif action_type == "command": + send_command(value) + print(f"[Agent] Sent command: {value}") + self.history.append({"role": "assistant", "content": f"COMMAND {value}"}) + self._trim_history() + 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): + break + + messages = [{"role": "system", "content": self.system_prompt}] + if self.instruction: + messages.append({"role": "system", "content": self.instruction}) + messages.extend(self.history) + messages.append( + { + "role": "system", + "content": ( + "Respond with JSON only. Schema: {\n" + " \"type\": \"tool\" or \"command\",\n" + " \"value\": string,\n" + " \"notes\": optional string\n}""" + ), + } + ) + 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": "user", + "content": "What is the next action you will take?", + } + ) + + print("[Agent] Intelligent agent finished.") + + def _trim_history(self) -> None: + while len(self.history) > 50: + self.history.popleft()