feat: have in-depth tool descriptios

This commit is contained in:
Daniel Eder 2025-09-29 07:00:07 +02:00
parent c807d8a6b3
commit f7bdf34329
3 changed files with 115 additions and 89 deletions

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

102
app.py
View file

@ -11,32 +11,58 @@ 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")},
),
TOOL_REGISTRY = {
"simple": {
"module": "tools",
"class": "SimpleTool",
"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
@ -190,8 +216,11 @@ def build_tool(spec: str) -> Tool:
if key == "simple":
return SimpleTool()
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}"})
@ -149,56 +157,59 @@ class IntelligentAgent(Agent):
self.history.append({"role": "assistant", "content": f"COMMAND {value}"})
self._trim_history()
elif action_type == "end":
print(f"[Agent] Execution ended by LLM.")
self.history.append({"role": "assistant", "content": f"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\" or \"end\",\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.")