JuicyTalk
async def run_warm_start_task():
print("="*60)
print("🔥 WARM START: Reusing previously evolved skills")
print("="*60)
task = (
"Create a Python script that analyzes a CSV file containing "
"inventory data with columns: date, item, quantity, cost. "
"The script should compute monthly expenditures, identify the top "
"5 most purchased items, and output a formatted summary report."
)
print(f"\n📝 Task: {task[:100]}...")
print(" (Similar to cold start task — skills should be reused)\n")
start_time = time.time()
try:
from openspace import OpenSpace
async with OpenSpace() as cs:
result = await cs.execute(task)
elapsed = time.time() - start_time
print(f"\n⏱️ Execution time: {elapsed:.1f}s")
response_text = result.get("response", str(result))
print(f"\n📄 Response (first 500 chars):")
print("-" * 40)
print(response_text[:500])
evolved = result.get("evolved_skills", [])
reused = result.get("reused_skills", [])
if reused:
print(f"\n♻️ Skills Reused: {len(reused)}")
for skill in reused:
print(f" • {skill.get('name', 'unnamed')}")
if evolved:
print(f"\n🧬 New Skills Evolved: {len(evolved)}")
for skill in evolved:
print(f" • {skill.get('name', 'unnamed')} ({skill.get('origin', '')})")
return result
except Exception as e:
print(f"\n⚠️ Execution error: {type(e).__name__}: {e}")
print("We'll simulate the comparison below.")
return None
warm_start_result = await run_warm_start_task()
async def demo_skill_search():
print("="*60)
print("🔎 SKILL SEARCH & DISCOVERY")
print("="*60)
try:
from openspace import OpenSpace
async with OpenSpace() as cs:
queries = [
"CSV data analysis with pandas",
"PDF report generation",
"web scraping with error handling",
]
for query in queries:
print(f"\n🔍 Query: '{query}'")
if hasattr(cs, 'skill_engine') and cs.skill_engine:
results = await cs.skill_engine.search(query)
if results:
for r in results[:3]:
print(f" 📋 {r.get('name', 'unnamed')} "
f"(score: {r.get('score', 'N/A')})")
else:
print(" (no matching skills found)")
else:
print(" (skill engine not initialized — "
"skills accumulate after task executions)")
except Exception as e:
print(f"\n⚠️ Search demo: {e}")
print("\n💡 Skill search becomes available after skills are evolved.")
print(" In production, run several tasks first to build up the skill database.")
await demo_skill_search()
def create_custom_skill(skill_name, description, instructions, triggers):
skill_dir = SKILLS_DIR / skill_name
skill_dir.mkdir(parents=True, exist_ok=True)
skill_md = f"""---
name: {skill_name}
description: {description}
version: 1.0.0
origin: manual
triggers: {json.dumps(triggers)}
---
# {skill_name}
{description}
## Instructions
{instructions}
## Quality Metrics
- Applied Rate: 0% (new skill)
- Completion Rate: N/A
- Effective Rate: N/A
"""
skill_path = skill_dir / "SKILL.md"
skill_path.write_text(skill_md)
print(f"✅ Created skill: {skill_name}")
print(f" Path: {skill_path}")
return skill_path
create_custom_skill(
skill_name="data-validation-csv",
description="Validate CSV files for common issues before processing: check encoding, detect delimiter, handle missing values, verify column types.",
instructions="""When working with CSV data:
1. **Encoding Detection**: Try UTF-8 first, then fall back to latin-1, cp1252
2. **Delimiter Detection**: Use csv.Sniffer() to auto-detect delimiter
3. **Missing Values**: Count NaN/null per column, report percentage
4. **Type Inference**: Check if numeric columns are actually numeric
5. **Duplicate Check**: Identify duplicate rows
```python
import pandas as pd
import csv
import chardet
def validate_csv(filepath):
with open(filepath, 'rb') as f:
result = chardet.detect(f.read(10000))
encoding = result['encoding']
df = pd.read_csv(filepath, encoding=encoding)
report = {
'rows': len(df),
'columns': list(df.columns),
'missing': 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 reports with multiple fallback strategies: try reportlab PDF first, fall back to HTML, then plain text.",
instructions="""When generating reports:
1. **Try reportlab PDF** first for professional output
2. **Fall back to HTML** if reportlab fails (common in sandboxed envs)
3. **Final fallback: plain text** with formatted tables
Always verify the output file exists and has non-zero size after generation.
```python
def generate_report(data, output_path):
try:
from reportlab.lib.pagesizes import letter
from reportlab.platypus import SimpleDocTemplate
return output_path
except ImportError:
pass
try:
html_path = output_path.replace('.pdf', '.html')
return html_path
except Exception:
pass
txt_path = output_path.replace('.pdf', '.txt')
return txt_path
```""",
triggers=["report", "PDF", "document generation", "reportlab"]
)
print()
create_custom_skill(
skill_name="execution-recovery",
description="Multi-layer execution recovery: handle sandbox failures, shell errors, and file write issues with progressive fallbacks.",
instructions="""When code execution fails:
1. **Capture the full error** including traceback
2. **Identify the failure type**: ImportError, PermissionError, TimeoutError, etc.
3. **Apply targeted fix**:
- ImportError → pip install the missing package
- PermissionError → change output directory to /tmp
- TimeoutError → reduce data size or add chunking
- MemoryError → process in batches
4. **Retry with fix applied**
5. **Log the fix** for future skill evolution
This skill was captured from 28 real execution failures in the GDPVal benchmark.""",
triggers=["error", "failure", "recovery", "fallback", "retry"]
)
print("\n" + "="*60)
print("📋 All registered skills:")
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 = skill_md.read_text()
for content line.split('\n'):
if line.startswith('name:'):
name = line.split(':', 1)[1].strip()
print(f" 🧩 {name}")
break
