lycis/mistle
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: tools run now from intelligent agent

Browse source
This commit is contained in:
Daniel Eder 2025-09-28 21:57:11 +02:00
parent 278cb7a8ec
commit f95b824a94
5 changed files with 124 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -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`.
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.
## Environment Variables

5
agent_runtime.py
View file

@ -8,7 +8,7 @@ from agents import Agent
from tools import Tool
ToolBuilder = Callable[[str], Tool]
ToolRunner = Callable[[Tool], None]
ToolRunner = Callable[[Tool], bool]
CommandRunner = Callable[[str], None]
@ -33,8 +33,7 @@ def run_agent(
print(f"[Agent] Failed to load tool '{name}': {exc}", file=sys.stderr)
return False
print(f"[Agent] Running tool '{name}'")
run_tool(tool)
return True
return run_tool(tool)
try:
agent.run(

7
agents.py
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
from typing import Callable, Optional, Sequence
ToolInvoker = Callable[[str], bool]
@ -83,7 +83,7 @@ class LoopAgent(Agent):
break
def build_agent(spec: str) -> Agent:
def build_agent(spec: str, *, allowed_tools: Optional[Sequence[str]] = None) -> Agent:
normalized = spec.strip()
if not normalized:
raise RuntimeError("Agent specification must not be empty")
@ -123,7 +123,8 @@ def build_agent(spec: str) -> Agent:
else:
instruction = segment
agent = IntelligentAgent(instruction=instruction)
whitelist = tuple(sorted({tool.lower() for tool in (allowed_tools or [])}))
agent = IntelligentAgent(instruction=instruction, allowed_tools=whitelist)
if model:
agent.model = model
agent.turn_delay = turn_delay if turn_delay > 0 else 1.0

72
app.py
View file

@ -10,6 +10,33 @@ from typing import Callable, Optional, Type
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")},
),
}
AVAILABLE_TOOL_NAMES = [
"move",
"movement",
"explore",
"communication",
"intelligentcommunication",
"simple",
]
from telnetclient import TelnetClient
@ -163,25 +190,8 @@ def build_tool(spec: str) -> Tool:
if key == "simple":
return SimpleTool()
builtin_tools = {
"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")},
),
}
if key in builtin_tools:
module_name, class_name, kwargs = builtin_tools[key]
if key in BUILTIN_TOOLS:
module_name, class_name, kwargs = BUILTIN_TOOLS[key]
try:
module = import_module(module_name)
tool_cls = getattr(module, class_name)
@ -397,7 +407,7 @@ def main() -> int:
print("[Agent] Usage: #agent <agent_spec>")
return
try:
temp_agent = build_agent(spec)
temp_agent = build_agent(spec, allowed_tools=AVAILABLE_TOOL_NAMES)
except RuntimeError as exc:
print(f"[Agent] Failed to configure '{spec}': {exc}", file=sys.stderr)
return
@ -411,7 +421,7 @@ def main() -> int:
except Exception as exc: # pragma: no cover - defensive
print(f"[Agent] observe failed: {exc}", file=sys.stderr)
def run_tool_instance(tool: Tool) -> None:
def run_tool_instance(tool: Tool) -> bool:
run_tool_loop(
client,
state,
@ -429,6 +439,7 @@ def main() -> int:
observe(output_after)
except Exception as exc: # pragma: no cover - defensive
print(f"[Agent] observe failed: {exc}", file=sys.stderr)
return True
thread = Thread(
target=run_agent,
@ -475,3 +486,22 @@ 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()))

93
intelligent_agent.py
View file

@ -32,6 +32,7 @@ class IntelligentAgent(Agent):
max_output_tokens: int = 200
instruction: str = ""
turn_delay: float = 0.0
allowed_tools: tuple[str, ...] = ()
history: Deque[dict[str, str]] = field(default_factory=deque, init=False)
def observe(self, message: str) -> None:
@ -58,6 +59,17 @@ class IntelligentAgent(Agent):
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.append(
{
"role": "system",
"content": (
"Available tools: "
f"{tool_list}. Use only these tool names when type=tool; do not invent new tools."
),
}
)
messages.append(
{
"role": "system",
@ -100,11 +112,8 @@ class IntelligentAgent(Agent):
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)
json_payload = self._extract_json(content)
payload = json.loads(json_payload)
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)
@ -120,10 +129,17 @@ class IntelligentAgent(Agent):
print("[Agent] LLM returned empty action", file=sys.stderr)
return
allowed_map = {tool.lower(): tool for tool in self.allowed_tools}
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}"})
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)
return
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}"})
self._trim_history()
if not success:
return
@ -136,40 +152,63 @@ class IntelligentAgent(Agent):
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
if self.turn_delay > 0 and stop_event.wait(self.turn_delay):
return
messages = [{"role": "system", "content": self.system_prompt}]
if self.instruction:
messages.append({"role": "system", "content": self.instruction})
messages.extend(self.history)
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.append(
{
"role": "system",
"content": (
"Respond with JSON only. Schema: {\n"
" \"type\": \"tool\" or \"command\",\n"
" \"value\": string,\n"
" \"notes\": optional string\n}"""
"Available tools: "
f"{tool_list}. Use only these tool names when type=tool; do not invent new tools."
),
}
)
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": "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": "user",
"content": "What is the next action you will take?",
"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()
def _extract_json(self, content: str) -> str:
text = content.strip()
if text.startswith("```"):
text = text.strip("` ")
if "\n" in text:
text = text.split("\n", 1)[1]
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end < start:
raise json.JSONDecodeError("No JSON object found", text, 0)
return text[start : end + 1]