Add baseline assistant with tools, guardrails, and system prompts (Phase 1.4)
- 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
This commit is contained in:
@@ -0,0 +1,83 @@
|
||||
"""
|
||||
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}'}
|
||||
Reference in New Issue
Block a user