Compare commits

...

3 commits

5 changed files with 135 additions and 98 deletions

View file

@ -9,7 +9,7 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows,
- Interactive console session that mirrors server output and lets you type commands directly. - Interactive console session that mirrors server output and lets you type commands directly.
- Optional always-on tool mode plus an on-demand `#execute <tool>` escape hatch for ad-hoc automations. - Optional always-on tool mode plus an on-demand `#execute <tool>` escape hatch for ad-hoc automations.
- Higher-level agents (`fixed`, `loop`, `intelligent`) that can string multiple tools together via `#agent <spec>`. - Higher-level agents (`fixed`, `loop`, `intelligent`) that can string multiple tools together via `#agent <spec>`.
- Built-in tools (`SimpleTool`, `ExploreTool`, `CommunicationTool`, `MovementTool`, `IntelligentCommunicationTool`) with a pluggable interface for custom behaviours. - Built-in tools (`LookTool`, `ExploreTool`, `CommunicationTool`, `MovementTool`, `IntelligentCommunicationTool`) with a pluggable interface for custom behaviours.
## Requirements ## Requirements
@ -64,7 +64,7 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows,
#agent intelligent #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`. The agent only calls built-in tools (`simple`, `move`, `movement`, `explore`, `communication`, `intelligentcommunication`) and refuses unknown names. Add guidance text after the type and optional modifiers separated by `|`, e.g. `#agent intelligent explore carefully|model=mistral/mistral-large-2407|delay=2`. The agent only calls built-in tools (`look`, `move`, `movement`, `explore`, `communication`, `intelligentcommunication`) and refuses unknown names.
## Environment Variables ## Environment Variables
@ -79,7 +79,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_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_EXIT_COMMAND` | ❌ | Command issued during graceful shutdown (after pressing `Ctrl-C`). Useful for `quit`/`save` macros. |
| `MISTLE_TOOL_MODE` | ❌ | Enable full-time tool thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. | | `MISTLE_TOOL_MODE` | ❌ | Enable full-time tool thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. |
| `MISTLE_TOOL` | ❌ | Select which tool class to instantiate when tool mode is active. Accepted values: `simple` (default), `explore`, `communication`, `movement`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. | | `MISTLE_TOOL` | ❌ | Select which tool class to instantiate when tool mode is active. Accepted values: `look` (default, alias `simple`), `explore`, `communication`, `movement`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. |
| `MISTLE_LLM_MODEL` | ❌ | Override the `litellm` model used by the intelligent tool (defaults to `mistral/mistral-small-2407`). | | `MISTLE_LLM_MODEL` | ❌ | Override the `litellm` model used by the intelligent tool (defaults to `mistral/mistral-small-2407`). |
| `MISTRAL_API_KEY` | ❌ | API key used by `IntelligentCommunicationTool` (via `litellm`) when calling the `mistral/mistral-small-2407` model. | | `MISTRAL_API_KEY` | ❌ | API key used by `IntelligentCommunicationTool` (via `litellm`) when calling the `mistral/mistral-small-2407` model. |
@ -98,7 +98,7 @@ All variables can be placed in the `.env` file (one `KEY=value` per line) or pro
## On-Demand Tools ## On-Demand Tools
- When `MISTLE_TOOL_MODE` is **off**, you can trigger an ephemeral tool at any time with `#execute <tool_spec>`. - When `MISTLE_TOOL_MODE` is **off**, you can trigger an ephemeral tool at any time with `#execute <tool_spec>`.
- The syntax accepts the same values as `MISTLE_TOOL` and reuses the `build_tool` helper, so `#execute simple`, `#execute explore`, `#execute move`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomTool` are all valid. - The syntax accepts the same values as `MISTLE_TOOL` and reuses the `build_tool` helper, so `#execute look`, `#execute explore`, `#execute move`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomTool` 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. - 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 ## Danger Zone

View file

@ -5,7 +5,7 @@ import sys
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
from threading import Event from threading import Event
from typing import Callable, Optional, Sequence from typing import Callable, Mapping, Optional
ToolInvoker = Callable[[str], bool] ToolInvoker = Callable[[str], bool]
@ -83,7 +83,9 @@ class LoopAgent(Agent):
break 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() normalized = spec.strip()
if not normalized: if not normalized:
raise RuntimeError("Agent specification must not be empty") raise RuntimeError("Agent specification must not be empty")
@ -123,8 +125,11 @@ def build_agent(spec: str, *, allowed_tools: Optional[Sequence[str]] = None) ->
else: else:
instruction = segment instruction = segment
whitelist = tuple(sorted({tool.lower() for tool in (allowed_tools or [])})) tool_map = {
agent = IntelligentAgent(instruction=instruction, allowed_tools=whitelist) name: description
for name, description in (allowed_tools or {}).items()
}
agent = IntelligentAgent(instruction=instruction, allowed_tools=tool_map)
if model: if model:
agent.model = model agent.model = model
agent.turn_delay = turn_delay if turn_delay > 0 else 1.0 agent.turn_delay = turn_delay if turn_delay > 0 else 1.0

110
app.py
View file

@ -7,36 +7,62 @@ from pathlib import Path
from threading import Event, Lock, Thread from threading import Event, Lock, Thread
from typing import Callable, Optional, Type from typing import Callable, Optional, Type
from tools import Tool, SimpleTool from tools import Tool, LookTool
from agents import Agent, build_agent from agents import Agent, build_agent
from agent_runtime import run_agent from agent_runtime import run_agent
BUILTIN_TOOLS = { TOOL_REGISTRY = {
"simple": ("tools", "SimpleTool", {}), "look": {
"explore": ("tools", "ExploreTool", {}), "module": "tools",
"communication": ("tools", "CommunicationTool", {}), "class": "LookTool",
"movement": ("movement_tool", "MovementTool", {}), "kwargs": {},
"move": ("movement_tool", "MovementTool", {}), "description": "Sends the 'schau' look command to refresh the room description.",
"intelligent": ( },
"intelligent_tool", "move": {
"IntelligentCommunicationTool", "module": "movement_tool",
{"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, "class": "MovementTool",
), "kwargs": {},
"intelligentcommunication": ( "description": "Looks around and moves in one available direction, chosen randomly among unvisited exits.",
"intelligent_tool", },
"IntelligentCommunicationTool", "movement": {
{"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")}, "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 <noun>' 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 <player> 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 = [ TOOL_DESCRIPTIONS = {
"move", name: meta["description"] for name, meta in TOOL_REGISTRY.items()
"movement", }
"explore",
"communication",
"intelligentcommunication",
"simple",
]
from telnetclient import TelnetClient from telnetclient import TelnetClient
@ -184,14 +210,17 @@ def build_tool(spec: str) -> Tool:
"""Instantiate a tool based on configuration.""" """Instantiate a tool based on configuration."""
normalized = spec.strip() normalized = spec.strip()
if not normalized: if not normalized:
return SimpleTool() return LookTool()
key = normalized.lower() key = normalized.lower()
if key == "simple": if key in {"simple", "look"}:
return SimpleTool() return LookTool()
if key in BUILTIN_TOOLS: if key in TOOL_REGISTRY:
module_name, class_name, kwargs = BUILTIN_TOOLS[key] meta = TOOL_REGISTRY[key]
module_name = meta["module"]
class_name = meta["class"]
kwargs = meta.get("kwargs", {})
try: try:
module = import_module(module_name) module = import_module(module_name)
tool_cls = getattr(module, class_name) tool_cls = getattr(module, class_name)
@ -407,7 +436,7 @@ def main() -> int:
print("[Agent] Usage: #agent <agent_spec>") print("[Agent] Usage: #agent <agent_spec>")
return return
try: try:
temp_agent = build_agent(spec, allowed_tools=AVAILABLE_TOOL_NAMES) temp_agent = build_agent(spec, allowed_tools=TOOL_DESCRIPTIONS)
except RuntimeError as exc: except RuntimeError as exc:
print(f"[Agent] Failed to configure '{spec}': {exc}", file=sys.stderr) print(f"[Agent] Failed to configure '{spec}': {exc}", file=sys.stderr)
return return
@ -486,22 +515,3 @@ def main() -> int:
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(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()))

View file

@ -32,7 +32,7 @@ class IntelligentAgent(Agent):
max_output_tokens: int = 200 max_output_tokens: int = 200
instruction: str = "" instruction: str = ""
turn_delay: float = 0.0 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) history: Deque[dict[str, str]] = field(default_factory=deque, init=False)
def observe(self, message: str) -> None: def observe(self, message: str) -> None:
@ -60,13 +60,15 @@ class IntelligentAgent(Agent):
messages.append({"role": "system", "content": self.instruction}) messages.append({"role": "system", "content": self.instruction})
messages.extend(self.history) messages.extend(self.history)
if self.allowed_tools: 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( messages.append(
{ {
"role": "system", "role": "system",
"content": ( "content": (
"Available tools: " "Available tools (only use these names when type=tool):\n"
f"{tool_list}. Use only these tool names when type=tool; do not invent new tools." f"{tool_list}"
), ),
} }
) )
@ -129,14 +131,20 @@ class IntelligentAgent(Agent):
print("[Agent] LLM returned empty action", file=sys.stderr) print("[Agent] LLM returned empty action", file=sys.stderr)
return 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": if action_type == "tool":
lower = value.lower() lower = value.lower()
if allowed_map and lower not in allowed_map: 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 return
canonical = allowed_map.get(lower, value) canonical, _ = allowed_map.get(lower, (value, ""))
success = invoke_tool(canonical) success = invoke_tool(canonical)
print(f"[Agent] Executed tool: {canonical} (success={success})") print(f"[Agent] Executed tool: {canonical} (success={success})")
self.history.append({"role": "assistant", "content": f"TOOL {canonical}"}) self.history.append({"role": "assistant", "content": f"TOOL {canonical}"})
@ -148,25 +156,33 @@ class IntelligentAgent(Agent):
print(f"[Agent] Sent command: {value}") print(f"[Agent] Sent command: {value}")
self.history.append({"role": "assistant", "content": f"COMMAND {value}"}) self.history.append({"role": "assistant", "content": f"COMMAND {value}"})
self._trim_history() self._trim_history()
elif action_type == "end":
print("[Agent] LLM requested to end the session.")
self.history.append({"role": "assistant", "content": "END"})
self._trim_history()
stop_event.set()
break
else: else:
print(f"[Agent] Unknown action type '{action_type}'", file=sys.stderr) print(f"[Agent] Unknown action type '{action_type}'", file=sys.stderr)
return return
if self.turn_delay > 0 and stop_event.wait(self.turn_delay): if self.turn_delay > 0 and stop_event.wait(self.turn_delay):
return break
messages = [{"role": "system", "content": self.system_prompt}] messages = [{"role": "system", "content": self.system_prompt}]
if self.instruction: if self.instruction:
messages.append({"role": "system", "content": self.instruction}) messages.append({"role": "system", "content": self.instruction})
messages.extend(self.history) messages.extend(self.history)
if self.allowed_tools: 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( messages.append(
{ {
"role": "system", "role": "system",
"content": ( "content": (
"Available tools: " "Available tools (only use these names when type=tool):\n"
f"{tool_list}. Use only these tool names when type=tool; do not invent new tools." f"{tool_list}"
), ),
} }
) )
@ -175,7 +191,7 @@ class IntelligentAgent(Agent):
"role": "system", "role": "system",
"content": ( "content": (
"Respond with JSON only. Schema: {\n" "Respond with JSON only. Schema: {\n"
" \"type\": \"tool\" or \"command\",\n" " \"type\": \"tool\" or \"command\" or \"end\",\n"
" \"value\": string,\n" " \"value\": string,\n"
" \"notes\": optional string\n}""" " \"notes\": optional string\n}"""
), ),

View file

@ -21,18 +21,24 @@ class Tool(ABC):
@dataclass @dataclass
class SimpleTool(Tool): class LookTool(Tool):
"""Minimal tool that always returns the same command.""" """Tool that sends a look command to refresh the room description."""
default_command: str = "schau" default_command: str = "schau"
last_output: str = field(default="", init=False) last_output: str = field(default="", init=False)
needs_look: bool = field(default=True, init=False)
def observe(self, output: str) -> None: def observe(self, output: str) -> None:
if output: if output:
self.last_output = output self.last_output = output
# Once we received output, we consider the look complete.
self.needs_look = False
def decide(self) -> Optional[str]: def decide(self) -> Optional[str]:
if self.needs_look:
self.needs_look = False
return self.default_command return self.default_command
return None
@dataclass @dataclass