File size: 6,350 Bytes
30d572f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | #!/usr/bin/env python3
"""
Extract training data from RTMP tools for Stack 2.9
Creates synthetic tool-use examples from the RTMP codebase
"""
import os
import json
from pathlib import Path
RTMP_DIR = "/Users/walidsobhi/.openclaw/workspace/RTMP"
OUTPUT_DIR = "/Users/walidsobhi/.openclaw/workspace/stack-2.9/data/rtmp-tools"
def get_tool_description(tool_name: str) -> str:
"""Get tool descriptions from tool names"""
descriptions = {
"BashTool": "Execute shell commands in a sandboxed environment",
"FileReadTool": "Read file contents from the filesystem",
"FileWriteTool": "Write content to files",
"FileEditTool": "Edit files using sed-style replacements",
"GlobTool": "Find files matching glob patterns",
"GrepTool": "Search for patterns in files",
"TaskCreateTool": "Create tasks in the task list",
"TaskListTool": "List all tasks in the task list",
"TaskUpdateTool": "Update task status and details",
"TaskGetTool": "Get details of a specific task",
"WebSearchTool": "Search the web for information",
"WebFetchTool": "Fetch and analyze web pages",
"SkillTool": "Execute user-invocable skills",
"McpTool": "Call MCP (Model Context Protocol) tools",
"AgentTool": "Delegate tasks to sub-agents",
}
return descriptions.get(tool_name.replace("Tool", ""), f"Tool: {tool_name}")
def extract_tool_examples():
"""Extract tool patterns and create training examples"""
examples = []
tools_dir = Path(RTMP_DIR) / "tools"
if not tools_dir.exists():
print(f"❌ Tools directory not found: {tools_dir}")
return examples
# Process each tool
tool_dirs = [d for d in tools_dir.iterdir() if d.is_dir() and d.name.endswith("Tool")]
for tool_dir in tool_dirs:
tool_name = tool_dir.name
# Get main tool file
main_files = list(tool_dir.glob("*.tsx")) + list(tool_dir.glob("*.ts"))
# Create example for this tool
desc = get_tool_description(tool_name)
# Generate prompt-response pairs for tool usage
example = {
"messages": [
{
"role": "user",
"content": f"How do I use the {tool_name}? What does it do?"
},
{
"role": "assistant",
"content": f"The {tool_name} allows you to {desc.lower()}.\n\n" +
f"This is one of the core tools in RTMP (Real-Time AI Assistant).\n\n" +
f"Tools like {tool_name} enable the assistant to interact with the filesystem, " +
f"execute commands, manage tasks, and more."
}
]
}
examples.append(example)
# Add more specific examples for common tools
if "Bash" in tool_name:
example = {
"messages": [
{"role": "user", "content": "List all files in the current directory"},
{"role": "assistant", "tool_calls": [
{"name": "Bash", "arguments": {"command": "ls -la"}}
]}
]
}
examples.append(example)
elif "FileRead" in tool_name:
example = {
"messages": [
{"role": "user", "content": "Read the contents of config.json"},
{"role": "assistant", "tool_calls": [
{"name": "FileRead", "arguments": {"file_path": "config.json"}}
]}
]
}
examples.append(example)
elif "Glob" in tool_name:
example = {
"messages": [
{"role": "user", "content": "Find all TypeScript files in the project"},
{"role": "assistant", "tool_calls": [
{"name": "Glob", "arguments": {"pattern": "**/*.ts"}}
]}
]
}
examples.append(example)
elif "Grep" in tool_name:
example = {
"messages": [
{"role": "user", "content": "Find all occurrences of 'TODO' in the code"},
{"role": "assistant", "tool_calls": [
{"name": "Grep", "arguments": {"pattern": "TODO", "path": "."}}
]}
]
}
examples.append(example)
elif "TaskCreate" in tool_name:
example = {
"messages": [
{"role": "user", "content": "Create a task to fix the login bug"},
{"role": "assistant", "tool_calls": [
{"name": "TaskCreate", "arguments": {
"subject": "Fix login bug",
"description": "Investigate and fix the login issue"
}}
]}
]
}
examples.append(example)
elif "WebSearch" in tool_name:
example = {
"messages": [
{"role": "user", "content": "Search for latest Python 3.14 features"},
{"role": "assistant", "tool_calls": [
{"name": "WebSearch", "arguments": {"query": "Python 3.14 new features"}}
]}
]
}
examples.append(example)
return examples
def main():
print("=" * 60)
print("Extracting RTMP Tool Patterns for Training")
print("=" * 60)
# Create output directory
os.makedirs(OUTPUT_DIR, exist_ok=True)
# Extract examples
examples = extract_tool_examples()
print(f"\n✅ Extracted {len(examples)} tool usage examples")
# Save to JSONL
output_file = os.path.join(OUTPUT_DIR, "tool_patterns.jsonl")
with open(output_file, 'w') as f:
for ex in examples:
f.write(json.dumps(ex) + '\n')
print(f"✅ Saved to: {output_file}")
# Also show some examples
print("\n📋 Sample examples:")
for i, ex in enumerate(examples[:3]):
user_msg = ex["messages"][0]["content"]
print(f" {i+1}. User: {user_msg[:60]}...")
if __name__ == "__main__":
main() |