This guide explains how schemas are extracted from commands and used to generate commands based on user prompts.
# src/nlp2cmd/schema_extraction/__init__.py
class DynamicSchemaRegistry:
"""Main registry for managing command schemas"""
def register_shell_help(self, command: str) -> ExtractedSchema
def register_appspec(self, appspec_path: str) -> ExtractedSchema
def register_dynamic_export(self, file_path: str) -> List[ExtractedSchema]
# src/nlp2cmd/storage/per_command_store.py
class PerCommandSchemaStore:
"""Stores each command schema in individual JSON files"""
def store_schema(self, schema: ExtractedSchema) -> bool
def load_schema(self, command: str) -> Optional[ExtractedSchema]
def list_commands(self) -> List[str]
# src/nlp2cmd/schema_based/generator.py
class SchemaBasedGenerator:
"""Generates commands using learned schemas"""
def learn_from_schema(self, schema: ExtractedSchema)
def generate_command(self, command: str, context: Dict) -> str
from nlp2cmd.schema_extraction import DynamicSchemaRegistry
# Initialize registry
registry = DynamicSchemaRegistry(
use_per_command_storage=True,
storage_dir="./command_schemas"
)
# Extract schema from command help
schema = registry.register_shell_help("docker")
print(f"Extracted schema for {schema.commands[0].name}")
print(f"Template: {schema.commands[0].template}")
{
"command": "docker",
"version": "1.0",
"description": "Docker container management",
"category": "container",
"parameters": [
{
"name": "subcommand",
"type": "string",
"description": "Docker subcommand",
"required": true,
"choices": ["ps", "run", "stop", "rm"]
}
],
"examples": [
"docker ps",
"docker run nginx",
"docker stop container_id"
],
"template": "docker {subcommand} {options}"
}
command_schemas/
├── commands/
│ ├── docker.json # Individual command schema
│ ├── kubectl.json
│ └── git.json
├── categories/
│ ├── container.json # Category index
│ └── version_control.json
└── index.json # Master index
from nlp2cmd.schema_based.adapter import SchemaDrivenAppSpecAdapter
from nlp2cmd import NLP2CMD
# Initialize with schema-driven adapter
adapter = SchemaDrivenAppSpecAdapter(schema_registry=registry)
nlp = NLP2CMD(adapter=adapter)
# Transform user prompt
result = nlp.transform("list all running containers")
print(result.command) # Output: docker ps
from nlp2cmd.schema_based.generator import SchemaBasedGenerator
# Load schema
schema = registry.get_command_by_name("docker")
# Create generator
generator = SchemaBasedGenerator()
generator.learn_from_schema(schema)
# Generate command
context = {"action": "list", "resource": "containers"}
command = generator.generate_command("docker", context)
print(command) # Output: docker ps
from nlp2cmd.intelligent.version_aware_generator import VersionAwareCommandGenerator
# Initialize with version detection
generator = VersionAwareCommandGenerator(schema_store=registry)
# Generate with automatic version detection
command, metadata = generator.generate_command("list containers")
print(f"Command: {command}")
print(f"Detected version: {metadata['detected_version']}")
print(f"Schema used: v{metadata['schema_version']}")
user_prompt = "show all running docker containers"
from nlp2cmd.intelligent.command_detector import CommandDetector
detector = CommandDetector()
detected_command = detector.detect_command(user_prompt)
# Returns: "docker"
schema = registry.get_command_by_name(detected_command)
# Loads: ./command_schemas/commands/docker.json
context = {
"action": "show",
"state": "running",
"resource": "containers"
}
generator = SchemaBasedGenerator()
generator.learn_from_schema(schema)
command = generator.generate_command(detected_command, context)
# Returns: "docker ps"
# examples/docker/basic_docker.py
from nlp2cmd import NLP2CMD
from nlp2cmd.adapters.dynamic import DynamicAdapter
# Initialize
adapter = DynamicAdapter(schema_registry=registry)
nlp = NLP2CMD(adapter=adapter)
# Test queries
queries = [
"list containers",
"run nginx on port 80",
"stop all containers",
"show container logs"
]
for query in queries:
result = nlp.transform(query)
print(f"Query: {query}")
print(f"Command: {result.command}")
print()
# examples/kubernetes/basic_kubernetes.py
from nlp2cmd.schema_based.adapter import SchemaDrivenAppSpecAdapter
adapter = SchemaDrivenAppSpecAdapter(schema_registry=registry)
nlp = NLP2CMD(adapter=adapter)
# Kubernetes queries
queries = [
"list all pods",
"get services",
"scale deployment to 3 replicas",
"check pod logs"
]
for query in queries:
result = nlp.transform(query)
print(f"{query} -> {result.command}")
# examples/shell/basic_shell.py
from nlp2cmd.intelligent import IntelligentNLP2CMD
# Initialize with intelligent generation
nlp = IntelligentNLP2CMD(storage_dir="./command_schemas")
# File operations
result = nlp.transform("find all python files")
# Output: find . -name "*.py"
result = nlp.transform("compress logs directory")
# Output: tar -czf logs.tar.gz logs/
# Generate schemas for all commands in cmd.csv
python3 tools/schema/update_schemas.py --force
# Generate schemas from prompts
python3 tools/generation/generate_cmd_simple.py
# List all stored commands
commands = registry.list_all_commands()
print(f"Available commands: {commands}")
# Get specific schema
schema = registry.get_command_by_name("docker")
print(f"Docker schema: {schema.commands[0].template}")
# Force update schema
schema = registry.register_shell_help("docker", force_update=True)
# Save to persistent storage
registry._auto_save()
src/nlp2cmd/schema_extraction/ - Schema extraction logicsrc/nlp2cmd/schema_based/ - Schema-based generationsrc/nlp2cmd/storage/ - Persistent storagesrc/nlp2cmd/intelligent/ - Intelligent generationexamples/docker/basic_docker.py - Docker examplesexamples/kubernetes/basic_kubernetes.py - Kubernetes examplesexamples/shell/basic_shell.py - Shell command examplescommand_schemas/commands/ - Individual command schemascommand_schemas/index.json - Master indexdocs/SCHEMA_SYSTEMS.md - Complete system documentationdocs/VERSIONED_SCHEMAS.md - Version management guide#!/usr/bin/env python3
"""Quick start example for NLP2CMD schemas"""
from nlp2cmd import NLP2CMD
from nlp2cmd.schema_extraction import DynamicSchemaRegistry
from nlp2cmd.adapters.dynamic import DynamicAdapter
def main():
# Initialize registry with storage
registry = DynamicSchemaRegistry(
use_per_command_storage=True,
storage_dir="./command_schemas"
)
# Ensure schemas are loaded
if not registry.schemas:
print("No schemas found. Generating...")
# Generate schemas for common commands
commands = ["docker", "kubectl", "git", "find", "grep"]
for cmd in commands:
registry.register_shell_help(cmd)
# Initialize NLP2CMD
adapter = DynamicAdapter(schema_registry=registry)
nlp = NLP2CMD(adapter=adapter)
# Test queries
test_queries = [
"list docker containers",
"show git status",
"find python files",
"grep for pattern in file"
]
print("NLP2CMD Schema-Based Generation:")
print("=" * 50)
for query in test_queries:
try:
result = nlp.transform(query)
print(f"Query: {query}")
print(f"Command: {result.command}")
print()
except Exception as e:
print(f"Error: {e}")
print(f"Total schemas loaded: {len(registry.schemas)}")
if __name__ == "__main__":
main()
# Check if schema exists
if command not in registry.schemas:
# Generate it
registry.register_shell_help(command)
# Enable debug mode
import logging
logging.basicConfig(level=logging.DEBUG)
# Check schema structure
schema = registry.get_command_by_name(command)
print(f"Schema template: {schema.commands[0].template}")
# Check storage directory
import os
if not os.path.exists("./command_schemas"):
os.makedirs("./command_schemas")