From 9b8584561e8b5b360b714be651f9f61a661a0932 Mon Sep 17 00:00:00 2001 From: Daniel Eder Date: Sun, 28 Sep 2025 10:15:33 +0200 Subject: [PATCH] feat: new agent to move around --- README.md | 7 +- agent.py | 1 + app.py | 40 ++++++---- movement_agent.py | 199 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 230 insertions(+), 17 deletions(-) create mode 100644 movement_agent.py diff --git a/README.md b/README.md index 2ea9857..770a27b 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 agent mode plus an on-demand `#execute ` escape hatch for ad-hoc automations. -- Built-in agents (`SimpleAgent`, `ExploreAgent`, `CommunicationAgent`, `IntelligentCommunicationAgent`) with a pluggable interface for custom behaviours. +- Built-in agents (`SimpleAgent`, `ExploreAgent`, `CommunicationAgent`, `MovementAgent`, `IntelligentCommunicationAgent`) with a pluggable interface for custom behaviours. ## Requirements @@ -54,7 +54,7 @@ All variables can be placed in the `.env` file (one `KEY=value` per line) or pro | `MISTLE_LOGIN_PROMPT` | ❌ | Prompt string that signals the client to send credentials (e.g., `"Name:"`). When omitted, the client just waits for the initial banner. | | `MISTLE_EXIT_COMMAND` | ❌ | Command issued during graceful shutdown (after pressing `Ctrl-C`). Useful for `quit`/`save` macros. | | `MISTLE_AGENT_MODE` | ❌ | Enable full-time agent thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. | -| `MISTLE_AGENT` | ❌ | Select which agent class to instantiate when agent mode is active. Accepted values: `simple` (default), `explore`, `communication`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. | +| `MISTLE_AGENT` | ❌ | Select which agent class to instantiate when agent mode is active. Accepted values: `simple` (default), `explore`, `communication`, `movement`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. | | `MISTLE_LLM_MODEL` | ❌ | Override the `litellm` model used by the intelligent agent (defaults to `mistral/mistral-small-2407`). | | `MISTRAL_API_KEY` | ❌ | API key used by `IntelligentCommunicationAgent` (via `litellm`) when calling the `mistral/mistral-small-2407` model. | @@ -67,12 +67,13 @@ All variables can be placed in the `.env` file (one `KEY=value` per line) or pro - `observe(output)` receives the latest server text; `decide()` returns the next command string or `None` to stay idle. - Commands issued by the agent are throttled to one per second so manual commands can still interleave smoothly. - `ExploreAgent` showcases a richer workflow: it sends `schau`, identifies German nouns, inspects each with `untersuche`, and prints `[Agent]` progress updates like `Explored 3/7 — untersuche Tisch`. +- `MovementAgent` parses room descriptions/exits and issues direction commands (prefers `n`, `e`, `s`, `w`, diagonals, then vertical moves) while tracking which exits were already tried. Use `#execute move` for ad-hoc pathing or set `MISTLE_AGENT=movement` for continuous roaming. - `CommunicationAgent` auto-replies to every direct tell with a canned greeting, while `IntelligentCommunicationAgent` routes each tell through `litellm` (default model `mistral/mistral-small-2407`) to craft a contextual answer. ## On-Demand Agents - When `MISTLE_AGENT_MODE` is **off**, you can trigger an ephemeral agent at any time with `#execute `. -- The syntax accepts the same values as `MISTLE_AGENT` and reuses the `build_agent` helper, so `#execute simple`, `#execute explore`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomAgent` are all valid. +- The syntax accepts the same values as `MISTLE_AGENT` and reuses the `build_agent` helper, so `#execute simple`, `#execute explore`, `#execute move`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomAgent` are all valid. - On-demand runs share the current session, respect the one-command-per-second limit, and stop automatically after a few seconds of inactivity. ## Danger Zone diff --git a/agent.py b/agent.py index 9d59cf8..acaf2a2 100644 --- a/agent.py +++ b/agent.py @@ -2,6 +2,7 @@ from __future__ import annotations import re from abc import ABC, abstractmethod + from collections import deque from dataclasses import dataclass, field from typing import Deque, Optional, Pattern, Set, Tuple diff --git a/app.py b/app.py index 0b6994b..28fd4ae 100644 --- a/app.py +++ b/app.py @@ -81,7 +81,27 @@ def run_agent_loop( """Invoke *agent* whenever new output arrives and send its response.""" idle_started: Optional[float] = None + def maybe_send() -> None: + nonlocal idle_started + try: + command = agent.decide() + except Exception as exc: # pragma: no cover - defensive logging + print(f"Agent failed: {exc}", file=sys.stderr) + return + if not command: + return + sent = state.agent_send( + client, + command, + min_interval=min_send_interval, + stop_event=stop_event, + ) + if not sent: + return + idle_started = None + while not stop_event.is_set(): + maybe_send() triggered = state.wait_for_output(timeout=idle_delay) if stop_event.is_set(): break @@ -101,20 +121,10 @@ def run_agent_loop( continue try: agent.observe(last_output) - command = agent.decide() except Exception as exc: # pragma: no cover - defensive logging - print(f"Agent failed: {exc}", file=sys.stderr) + print(f"Agent failed during observe: {exc}", file=sys.stderr) continue - if not command: - continue - sent = state.agent_send( - client, - command, - min_interval=min_send_interval, - stop_event=stop_event, - ) - if not sent: - break + maybe_send() def load_env_file(path: str = ".env") -> None: @@ -155,15 +165,17 @@ def build_agent(agent_spec: str) -> Agent: builtin_agents = { "explore": ("agent", "ExploreAgent", {}), "communication": ("agent", "CommunicationAgent", {}), + "movement": ("movement_agent", "MovementAgent", {}), + "move": ("movement_agent", "MovementAgent", {}), "intelligent": ( "intelligent_agent", "IntelligentCommunicationAgent", - {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-tiny")}, + {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, ), "intelligentcommunication": ( "intelligent_agent", "IntelligentCommunicationAgent", - {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-tiny")}, + {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, ), } diff --git a/movement_agent.py b/movement_agent.py new file mode 100644 index 0000000..70cc608 --- /dev/null +++ b/movement_agent.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import re +import random +from dataclasses import dataclass, field +from typing import Dict, Optional, Pattern, Sequence, Set, Tuple + +from agent import Agent + + +@dataclass +class MovementAgent(Agent): + """Agent that chooses movement commands based on room descriptions.""" + + look_command: str = "schau" + command_format: str = "{direction}" + preferred_order: Tuple[str, ...] = ( + "n", + "ne", + "e", + "se", + "s", + "sw", + "w", + "nw", + "u", + "d", + ) + last_output: str = field(default="", init=False) + current_room_key: Optional[str] = field(default=None, init=False) + visited_directions: Dict[str, Set[str]] = field(default_factory=dict, init=False) + executed: bool = field(default=False, init=False) + needs_look: bool = field(default=True, init=False) + selected_direction: Optional[str] = field(default=None, init=False) + random_choice: bool = True + + _direction_aliases: Dict[str, str] = field( + default_factory=lambda: { + "n": "n", + "north": "n", + "norden": "n", + "nord": "n", + "no": "ne", + "ne": "ne", + "nordost": "ne", + "nordosten": "ne", + "e": "e", + "east": "e", + "osten": "e", + "ost": "e", + "so": "se", + "se": "se", + "südost": "se", + "südosten": "se", + "s": "s", + "south": "s", + "süden": "s", + "süd": "s", + "sw": "sw", + "südwest": "sw", + "südwesten": "sw", + "w": "w", + "west": "w", + "westen": "w", + "nw": "nw", + "nordwest": "nw", + "nordwesten": "nw", + "u": "u", + "oben": "u", + "up": "u", + "rauf": "u", + "d": "d", + "down": "d", + "hinunter": "d", + "unten": "d", + }, + init=False, + ) + + _exit_line_pattern: Pattern[str] = field( + default_factory=lambda: re.compile( + r"(?:ausg(?:ä|a)nge?|exits?)[:\s]+(.+)", re.IGNORECASE + ), + init=False, + ) + + def observe(self, output: str) -> None: + if not output or self.executed: + return + self.last_output = output + if self.needs_look: + return + room_key = self._room_key(output) + if self.selected_direction is None: + direction = self._select_direction(output, room_key) + if direction: + self.current_room_key = room_key + self.selected_direction = direction + + def decide(self) -> Optional[str]: + if self.executed: + return None + if self.needs_look: + self.needs_look = False + print("[Agent] Requesting room description via look command") + return self.look_command + if self.selected_direction is None: + return None + direction = self.selected_direction + if self.current_room_key is not None: + self.visited_directions.setdefault(self.current_room_key, set()).add( + direction + ) + command = self.command_format.format(direction=direction) + print(f"[Agent] Moving via {direction}") + self.executed = True + self.selected_direction = None + return command + + def _room_key(self, text: str) -> str: + return re.sub(r"\s+", " ", text.strip()) + + def _select_direction(self, text: str, room_key: str) -> Optional[str]: + exits = self._parse_exit_lines(text) + if not exits: + exits = self._scan_for_directions(text) + if not exits: + return None + + ordered = self._prioritized_exits(exits, room_key) + if not ordered: + return None + + visited = self.visited_directions.setdefault(room_key, set()) + candidates = [direction for direction in ordered if direction not in visited] + pool = candidates or ordered + return self._choose_direction(pool) + + def _prioritized_exits( + self, exits: Sequence[Tuple[str, Optional[str]]], room_key: str + ) -> list[str]: + ordered: list[str] = [] + seen: Set[str] = set() + + for pref in self.preferred_order: + for raw, canonical in exits: + if canonical == pref and raw not in seen: + ordered.append(raw) + seen.add(raw) + + for raw, _ in exits: + if raw not in seen: + ordered.append(raw) + seen.add(raw) + + return ordered + + def _choose_direction(self, pool: Sequence[str]) -> str: + if not pool: + raise ValueError("No exits available to choose from") + if self.random_choice and len(pool) > 1: + return random.choice(list(pool)) + return pool[0] + + def _parse_exit_lines(self, text: str) -> list[Tuple[str, Optional[str]]]: + exits: list[Tuple[str, Optional[str]]] = [] + for line in text.splitlines(): + match = self._exit_line_pattern.search(line) + if not match: + continue + candidates = re.split(r"[,;/]| und | oder |\\s", match.group(1)) + for candidate in candidates: + normalized = self._normalize_raw(candidate) + if not normalized: + continue + exits.append(normalized) + return exits + + def _scan_for_directions(self, text: str) -> list[Tuple[str, Optional[str]]]: + exits: list[Tuple[str, Optional[str]]] = [] + lowered = text.lower() + for word, direction in self._direction_aliases.items(): + if re.search(rf"\b{re.escape(word)}\b", lowered): + normalized = self._normalize_raw(word) + if normalized: + exits.append((normalized[0], direction)) + return exits + + def _map_direction(self, raw: str) -> Optional[str]: + cleaned = re.sub(r"[^a-zäöüß]", "", raw.lower()) + if not cleaned: + return None + return self._direction_aliases.get(cleaned) + + def _normalize_raw(self, raw: str) -> Optional[Tuple[str, Optional[str]]]: + cleaned = re.sub(r"[\s\.,;:!?-]+$", "", raw.strip()) + if not cleaned: + return None + return cleaned, self._map_direction(cleaned)