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.
- 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>`.
- 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
@ -64,7 +64,7 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows,
#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
@ -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_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` | ❌ | 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`). |
| `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
- 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.
## Danger Zone

View file

@ -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

110
app.py
View file

@ -7,36 +7,62 @@ from pathlib import Path
from threading import Event, Lock, Thread
from typing import Callable, Optional, Type
from tools import Tool, SimpleTool
from tools import Tool, LookTool
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 = {
"look": {
"module": "tools",
"class": "LookTool",
"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 <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 = [
"move",
"movement",
"explore",
"communication",
"intelligentcommunication",
"simple",
]
TOOL_DESCRIPTIONS = {
name: meta["description"] for name, meta in TOOL_REGISTRY.items()
}
from telnetclient import TelnetClient
@ -184,14 +210,17 @@ def build_tool(spec: str) -> Tool:
"""Instantiate a tool based on configuration."""
normalized = spec.strip()
if not normalized:
return SimpleTool()
return LookTool()
key = normalized.lower()
if key == "simple":
return SimpleTool()
if key in {"simple", "look"}:
return LookTool()
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 <agent_spec>")
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()))

View file

@ -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}"})
@ -148,52 +156,60 @@ class IntelligentAgent(Agent):
print(f"[Agent] Sent command: {value}")
self.history.append({"role": "assistant", "content": f"COMMAND {value}"})
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:
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\",\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.")

View file

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