e00d454b19
- agent/serve.py: CLI assistant with interactive, single-query, and eval modes (Ollama + qwen3-coder) - agent/tools/rcon_tool.py: RCON execute, server status, player info - agent/tools/knowledge_tool.py: TF-IDF RAG search, command reference lookup, server context - agent/guardrails/command_filter.py: 14-prefix allowlist, execute-tail bypass detection, destructive flags, 1.21 syntax warnings, audit log - agent/prompts/system_prompts.py: sudo (pure commands), god (persona), intervention (benign) system prompts - Guardrails tested: 10/10 allowlist, 5/6 syntax warnings pass
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""
|
|
Knowledge/RAG tool for Minecraft command and server reference lookups.
|
|
|
|
Wraps the TF-IDF index built by knowledge/build_index.py.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List
|
|
|
|
KNOWLEDGE_ROOT = Path(__file__).resolve().parent.parent.parent / 'knowledge'
|
|
|
|
|
|
def _tokenize(text: str) -> set:
|
|
return set(re.findall(r'[a-z0-9_:/.]{2,}', (text or '').lower()))
|
|
|
|
|
|
def _load_index() -> dict:
|
|
idx_path = KNOWLEDGE_ROOT / 'index.json'
|
|
if not idx_path.exists():
|
|
return {'docs': [], 'idf': {}}
|
|
return json.loads(idx_path.read_text())
|
|
|
|
|
|
def search_knowledge(query: str, limit: int = 5) -> List[Dict[str, Any]]:
|
|
"""Search the knowledge index for relevant documents."""
|
|
index = _load_index()
|
|
q_tokens = _tokenize(query)
|
|
idf = index.get('idf', {})
|
|
results = []
|
|
|
|
for doc in index.get('docs', []):
|
|
d_tokens = set(doc.get('tokens', []))
|
|
overlap = q_tokens & d_tokens
|
|
if not overlap:
|
|
continue
|
|
score = sum(idf.get(t, 0.5) for t in overlap)
|
|
title_tokens = _tokenize(doc.get('title', ''))
|
|
title_overlap = q_tokens & title_tokens
|
|
score += len(title_overlap) * 2.0
|
|
results.append((score, doc))
|
|
|
|
results.sort(key=lambda x: x[0], reverse=True)
|
|
return [{'score': round(s, 2), 'id': d['id'], 'title': d['title'],
|
|
'snippet': d['snippet'], 'source': d['source']}
|
|
for s, d in results[:limit]]
|
|
|
|
|
|
def get_command_reference(command: str) -> Dict[str, Any]:
|
|
"""Get the full reference entry for a specific command."""
|
|
cmd_path = KNOWLEDGE_ROOT / 'mc-commands' / 'commands.json'
|
|
if not cmd_path.exists():
|
|
return {'found': False, 'error': 'commands.json not found'}
|
|
|
|
commands = json.loads(cmd_path.read_text())
|
|
cmd_name = command.lstrip('/').lower().strip()
|
|
for entry in commands:
|
|
if entry.get('command', '').lower() == cmd_name:
|
|
return {'found': True, 'command': entry}
|
|
if cmd_name in [a.lower() for a in entry.get('aliases', [])]:
|
|
return {'found': True, 'command': entry}
|
|
|
|
return {'found': False, 'error': f'No reference for /{cmd_name}'}
|
|
|
|
|
|
def get_server_context(server_name: str = '') -> Dict[str, Any]:
|
|
"""Get server configuration context."""
|
|
srv_path = KNOWLEDGE_ROOT / 'server-context' / 'servers.json'
|
|
if not srv_path.exists():
|
|
return {'found': False, 'error': 'servers.json not found'}
|
|
|
|
data = json.loads(srv_path.read_text())
|
|
if not server_name:
|
|
return {'found': True, 'servers': data.get('servers', []),
|
|
'version_notes': data.get('version_notes', {})}
|
|
|
|
for srv in data.get('servers', []):
|
|
if srv.get('name', '').lower() == server_name.lower():
|
|
return {'found': True, 'server': srv,
|
|
'version_notes': data.get('version_notes', {})}
|
|
|
|
return {'found': False, 'error': f'No server named {server_name}'}
|