async def run_warm_start_task():
print("="*60)
print("🔥 WARM START: Reusing beforehand advanced abilities")
print("="*60)
job = (
"Create a Python script that analyzes a CSV file containing "
"stock information with columns: date, merchandise, amount, value. "
"The script ought to compute month-to-month expenditures, establish the highest "
"5 most bought objects, and output a formatted abstract report."
)
print(f"n📝 Activity: {job[:100]}...")
print(" (Just like chilly begin job — abilities must be reused)n")
start_time = time.time()
strive:
from openspace import OpenSpace
async with OpenSpace() as cs:
end result = await cs.execute(job)
elapsed = time.time() - start_time
print(f"n⏱️ Execution time: {elapsed:.1f}s")
response_text = end result.get("response", str(end result))
print(f"n📄 Response (first 500 chars):")
print("-" * 40)
print(response_text[:500])
advanced = end result.get("evolved_skills", [])
reused = end result.get("reused_skills", [])
if reused:
print(f"n♻️ Expertise Reused: {len(reused)}")
for talent in reused:
print(f" • {talent.get('title', 'unnamed')}")
if advanced:
print(f"n🧬 New Expertise Developed: {len(advanced)}")
for talent in advanced:
print(f" • {talent.get('title', 'unnamed')} ({talent.get('origin', '')})")
return end result
besides Exception as e:
print(f"n⚠️ Execution error: {kind(e).__name__}: {e}")
print("We'll simulate the comparability beneath.")
return None
warm_start_result = await run_warm_start_task()
async def demo_skill_search():
print("="*60)
print("🔎 SKILL SEARCH & DISCOVERY")
print("="*60)
strive:
from openspace import OpenSpace
async with OpenSpace() as cs:
queries = [
"CSV data analysis with pandas",
"PDF report generation",
"web scraping with error handling",
]
for question in queries:
print(f"n🔍 Question: '{question}'")
if hasattr(cs, 'skill_engine') and cs.skill_engine:
outcomes = await cs.skill_engine.search(question)
if outcomes:
for r in outcomes[:3]:
print(f" 📋 {r.get('title', 'unnamed')} "
f"(rating: {r.get('rating', 'N/A')})")
else:
print(" (no matching abilities discovered)")
else:
print(" (talent engine not initialized — "
"abilities accumulate after job executions)")
besides Exception as e:
print(f"n⚠️ Search demo: {e}")
print("n💡 Ability search turns into accessible after abilities are advanced.")
print(" In manufacturing, run a number of duties first to construct up the talent database.")
await demo_skill_search()
def create_custom_skill(skill_name, description, directions, triggers):
skill_dir = SKILLS_DIR / skill_name
skill_dir.mkdir(mother and father=True, exist_ok=True)
skill_md = f"""---
title: {skill_name}
description: {description}
model: 1.0.0
origin: handbook
triggers: {json.dumps(triggers)}
---
# {skill_name}
{description}
## Directions
{directions}
## High quality Metrics
- Utilized Price: 0% (new talent)
- Completion Price: N/A
- Efficient Price: N/A
"""
skill_path = skill_dir / "SKILL.md"
skill_path.write_text(skill_md)
print(f"✅ Created talent: {skill_name}")
print(f" Path: {skill_path}")
return skill_path
create_custom_skill(
skill_name="data-validation-csv",
description="Validate CSV information for frequent points earlier than processing: test encoding, detect delimiter, deal with lacking values, confirm column varieties.",
directions="""When working with CSV information:
1. **Encoding Detection**: Strive UTF-8 first, then fall again to latin-1, cp1252
2. **Delimiter Detection**: Use csv.Sniffer() to auto-detect delimiter
3. **Lacking Values**: Depend NaN/null per column, report proportion
4. **Sort Inference**: Test if numeric columns are literally numeric
5. **Duplicate Test**: Determine duplicate rows
```python
import pandas as pd
import csv
import chardet
def validate_csv(filepath):
with open(filepath, 'rb') as f:
end result = chardet.detect(f.learn(10000))
encoding = end result['encoding']
df = pd.read_csv(filepath, encoding=encoding)
report = {
'rows': len(df),
'columns': record(df.columns),
'lacking': df.isnull().sum().to_dict(),
'duplicates': df.duplicated().sum(),
'dtypes': df.dtypes.astype(str).to_dict()
}
return report
```""",
triggers=["csv", "data validation", "data quality", "pandas"]
)
print()
create_custom_skill(
skill_name="report-gen-fallback",
description="Generate stories with a number of fallback methods: strive reportlab PDF first, fall again to HTML, then plain textual content.",
directions="""When producing stories:
1. **Strive reportlab PDF** first for skilled output
2. **Fall again to HTML** if reportlab fails (frequent in sandboxed envs)
3. **Closing fallback: plain textual content** with formatted tables
All the time confirm the output file exists and has non-zero dimension after era.
```python
def generate_report(information, output_path):
strive:
from reportlab.lib.pagesizes import letter
from reportlab.platypus import SimpleDocTemplate
return output_path
besides ImportError:
go
strive:
html_path = output_path.substitute('.pdf', '.html')
return html_path
besides Exception:
go
txt_path = output_path.substitute('.pdf', '.txt')
return txt_path
```""",
triggers=["report", "PDF", "document generation", "reportlab"]
)
print()
create_custom_skill(
skill_name="execution-recovery",
description="Multi-layer execution restoration: deal with sandbox failures, shell errors, and file write points with progressive fallbacks.",
directions="""When code execution fails:
1. **Seize the complete error** together with traceback
2. **Determine the failure kind**: ImportError, PermissionError, TimeoutError, and many others.
3. **Apply focused repair**:
- ImportError → pip set up the lacking bundle
- PermissionError → change output listing to /tmp
- TimeoutError → cut back information dimension or add chunking
- MemoryError → course of in batches
4. **Retry with repair utilized**
5. **Log the repair** for future talent evolution
This talent was captured from 28 actual execution failures within the GDPVal benchmark.""",
triggers=["error", "failure", "recovery", "fallback", "retry"]
)
print("n" + "="*60)
print("📋 All registered abilities:")
print("="*60)
for skill_dir in sorted(SKILLS_DIR.iterdir()):
if skill_dir.is_dir():
skill_md = skill_dir / "SKILL.md"
if skill_md.exists():
content material = skill_md.read_text()
for content material line.break up('n'):
if line.startswith('title:'):
title = line.break up(':', 1)[1].strip()
print(f" 🧩 {title}")
break
