async def run_warm_start_task():
print(“=”*60)
print(“🔥 WARM START: Reusing beforehand developed expertise”)
print(“=”*60)
activity = (
“Create a Python script that analyzes a CSV file containing ”
“stock knowledge 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: {activity[:100]}…”)
print(” (Just like chilly begin activity — expertise must be reused)n”)
start_time = time.time()
strive:
from openspace import OpenSpace
async with OpenSpace() as cs:
end result = await cs.execute(activity)
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])
developed = 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(‘identify’, ‘unnamed’)}”)
if developed:
print(f”n🧬 New Expertise Developed: {len(developed)}”)
for talent in developed:
print(f” • {talent.get(‘identify’, ‘unnamed’)} ({talent.get(‘origin’, ”)})”)
return end result
besides Exception as e:
print(f”n⚠️ Execution error: {sort(e).__name__}: {e}”)
print(“We’ll simulate the comparability under.”)
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(‘identify’, ‘unnamed’)} ”
f”(rating: {r.get(‘rating’, ‘N/A’)})”)
else:
print(” (no matching expertise discovered)”)
else:
print(” (talent engine not initialized — ”
“expertise accumulate after activity executions)”)
besides Exception as e:
print(f”n⚠️ Search demo: {e}”)
print(“n💡 Talent search turns into accessible after expertise are developed.”)
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”””—
identify: {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: verify encoding, detect delimiter, deal with lacking values, confirm column varieties.”,
directions=”””When working with CSV knowledge:
1. **Encoding Detection**: Attempt UTF-8 first, then fall again to latin-1, cp1252
2. **Delimiter Detection**: Use csv.Sniffer() to auto-detect delimiter
3. **Lacking Values**: Rely NaN/null per column, report share
4. **Kind Inference**: Examine if numeric columns are literally numeric
5. **Duplicate Examine**: Establish 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’: checklist(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. **Attempt 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(knowledge, 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.exchange(‘.pdf’, ‘.html’)
return html_path
besides Exception:
go
txt_path = output_path.exchange(‘.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 total error** together with traceback
2. **Establish the failure sort**: ImportError, PermissionError, TimeoutError, and so on.
3. **Apply focused repair**:
– ImportError → pip set up the lacking package deal
– PermissionError → change output listing to /tmp
– TimeoutError → cut back knowledge 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 expertise:”)
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.cut up(‘n’):
if line.startswith(‘identify:’):
identify = line.cut up(‘:’, 1)[1].strip()
print(f” 🧩 {identify}”)
break
