from coreskill import LLMClient, SkillManager, EvoEngine, IntentEnginefrom cores.v1.llm_client import LLMClient
llm = LLMClient(
api_key="sk-or-v1-...", # Opcjonalnie
model="openrouter/meta-llama/llama-3.3-70b-instruct:free"
)Wykonuje chat completion z automatycznym fallbackiem.
result = llm.chat(
messages=[
{"role": "system", "content": "You are helpful"},
{"role": "user", "content": "Hello"}
],
temperature=0.7,
max_tokens=4096
)Returns: str - odpowiedź modelu
Zwraca dostępne modele.
# Wszystkie modele
all_models = llm.get_available_models()
# Tylko free tier
free_models = llm.get_available_models(tier="free")
# Tylko local
local_models = llm.get_available_models(tier="local")Ustawia aktywny model.
llm.set_model("ollama/llama3.2:3b")from cores.v1.skill_manager import SkillManager
sm = SkillManager(skills_dir="./skills")Wykonuje skill z walidacją preflight.
result = sm.exec_skill(
name="tts",
action="execute",
params={"text": "Hello world"}
)
# Result format:
# {
# "success": True,
# "result": {...},
# "duration_ms": 123
# }Ewoluuje skill używając LLM.
sm.smart_evolve(
name="moj_skill",
goal="Dodaj obsługę formatowania markdown",
llm_client=llm
)Sprawdza health skilla.
health = sm.check_health("tts")
# Returns: {"status": "ok", "errors": [], "warnings": []}Lista wszystkich skillów.
skills = sm.list_skills()
# Returns: ["tts", "stt", "shell", ...]from cores.v1.evo_engine import EvoEngine
evo = EvoEngine(
skill_manager=sm,
llm_client=llm,
intent_engine=intent
)Obsługuje request z pipeline'em.
result = evo.handle_request(
goal="Przeczytaj mi artykuł o Pythonie",
context={"voice_mode": True}
)Ewoluuje skill iteracyjnie.
evo.evolve_skill(
name="web_search",
goal="Popraw wyszukiwanie DuckDuckGo",
iterations=3
)from cores.v1.intent_engine import IntentEngine
intent = IntentEngine(skills={"tts": {...}, "stt": {...}})Analizuje intencję tekstu.
analysis = intent.analyze(
text="Przeczytaj mi to na głos",
skills={"tts": {...}, "stt": {...}},
conversation=[]
)
# Returns:
# {
# "action": "use",
# "skill": "tts",
# "confidence": 0.85,
# "goal": "Przeczytaj tekst na głos"
# }Rejestruje poprawkę do uczenia.
intent.record_correction(
text="Przeczytaj mi to",
correct_skill="tts" # Poprzednio wykryło "stt"
)from cores.v1.user_memory import UserMemory
from cores.v1.config import load_state
state = load_state()
memory = UserMemory(state)Dodaje dyrektywę.
memory.add("Zawsze rozmawiaj po polsku", priority="high")Usuwa dyrektywę.
memory.remove(1)Buduje kontekst dla LLM.
context = memory.build_system_context()
# Returns: "WAŻNE — trwałe preferencje użytkownika: ..."Wyświetla wszystkie dyrektywy.
memory.display()from cores.v1.config import load_state, save_state
# Load state
state = load_state()
# Modify
state["api_key"] = "new-key"
# Save (merges with existing)
save_state(state)from cores.v1.config import (
ROOT, # Project root
SKILLS_DIR, # skills/ folder
STATE_FILE, # .evo_state.json
LOGS_DIR # logs/ folder
)Każdy skill musi implementować:
def get_info() -> dict:
"""Zwraca metadane skilla."""
return {
"name": "skill_name",
"version": "v1",
"description": "Opis co skill robi"
}
def execute(params: dict) -> dict:
"""Wykonuje akcję skilla."""
# ... logika ...
return {
"success": True, # bool - czy się udało
"result": {...}, # dane wynikowe
"error": None # string jeśli błąd
}class MySkill:
def __init__(self):
self.name = "my_skill"
def get_info(self):
return {"name": self.name, "version": "v1"}
def execute(self, params):
return {"success": True, "result": "done"}
def get_info():
return MySkill().get_info()
def execute(params):
return MySkill().execute(params)from cores.v1.skill_logger import get_skill_logger
logger = get_skill_logger("my_skill")
@logger.log_call
def my_function():
passfrom cores.v1.skill_logger import (
query_skill_errors,
query_slow_calls,
skill_health_summary
)
# Get recent errors
errors = query_skill_errors("tts", last_n=10)
# Get slow calls
slow = query_slow_calls(threshold_ms=1000, last_n=5)
# Health summary
health = skill_health_summary("tts")
# Returns: {"total_calls": 100, "errors": 2, "error_rate": 0.02, ...}{
"success": False,
"error": "Opis błędu",
"error_type": "import_error", # lub: syntax_error, runtime_error, etc.
"suggestion": "Jak naprawić"
}from cores.v1.preflight import EvolutionGuard
guard = EvolutionGuard()
guard.record_error("tts", error_fingerprint)
strategy = guard.suggest_strategy(error_fingerprint)
# Returns: "auto_fix_imports" | "install_deps" | "rewrite" | "evolve"from cli import cmd_logs_reset, cmd_cache_reset
import argparse
# Reset logs
args = argparse.Namespace()
cmd_logs_reset(args)
# Reset cache
args = argparse.Namespace(full=False)
cmd_cache_reset(args)# Detect intent
analysis = intent.analyze(user_input, skills)
# Execute skill
if analysis["action"] == "use":
result = sm.exec_skill(
analysis["skill"],
params={"text": analysis["goal"]}
)
# Handle result
if result["success"]:
print(result["result"])
else:
# Auto-fix
evo.evolve_skill(analysis["skill"], result["error"])# Build context
memory_context = memory.build_system_context()
# Create LLM messages
messages = [
{"role": "system", "content": SYSTEM_PROMPT + "\n" + memory_context},
{"role": "user", "content": user_input}
]
# Call LLM
response = llm.chat(messages)# Auto-create skill
from cores.v1.evo_engine import EvoEngine
evo = EvoEngine(sm, llm, intent)
code = evo.generate_skill(
name="calculator",
description="Simple calculator with + - * /"
)
# Save
sm.save_skill("calculator", code)from cores.v1.skill_forge import SkillForge
from cores.v1.smart_intent import EmbeddingEngine
# Z embedding engine (dla semantic search)
embedder = EmbeddingEngine()
forge = SkillForge(embedding_engine=embedder)
# Bez embedding (keyword fallback)
forge = SkillForge()Buduje indeks embeddingów dla istniejących skillów.
skills = sm.list_skills() # {name: [versions]}
forge.index_skills(skills)Decyduje czy tworzyć nowy skill.
should_create, reason = forge.should_create(
"policz 2+2",
sm.list_skills()
)
# reason values:
# - "reuse:kalkulator" -> użyj istniejącego skillu
# - "chat" -> konwersacja, nie twórz skillu
# - "budget_exceeded" -> za dużo błędów (max 10/h)
# - "new_skill_needed" -> stwórz nowy skill
if reason.startswith("reuse:"):
skill_name = reason.split(":")[1]
result = sm.exec_skill(skill_name, params={"text": "2+2"})Wyszukiwanie semantyczne skillów.
matches = forge.search("obliczanie matematyczne", top_k=3)
for match in matches:
print(f"{match.name}: {match.similarity:.2f}")from cores.v1.base_skill import BaseSkill, _make_module_functions
class CalculatorSkill(BaseSkill):
name = "calculator"
version = "v1"
description = "Simple calculator supporting + - * /"
def execute(self, params: dict) -> dict:
# Get input with fallback to text parsing
expression = params.get("expression",
params.get("text", ""))
# Business logic
try:
result = eval(expression) # Safe: only math
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
# Generate module-level functions
execute, get_info, health_check = _make_module_functions(CalculatorSkill)
if __name__ == "__main__":
import json
print(json.dumps(execute({"text": "2+2"}), indent=2))# get_info() - zwraca metadane z atrybutów klasy
info = skill.get_info()
# {"name": "calculator", "version": "v1", "description": "..."}
# health_check() - domyślnie zwraca {"status": "ok"}
health = skill.health_check()
# safe_execute() - wrapper z obsługą błędów
result = skill.safe_execute(params) # Zawsze zwraca dict z successfrom cores.v1.base_skill import SkillManifest, InputField
manifest = SkillManifest(
name="camera_scanner",
version="v1",
description="Scans network for IP cameras",
inputs=[
InputField(name="network", type="string",
default="192.168.1.0/24",
description="Network CIDR to scan"),
InputField(name="timeout", type="integer",
default=30, required=False),
],
requires_commands=["nmap"],
tags=["network", "security"]
)Ładuje manifest z YAML/JSON.
manifest = SkillManifest.from_file(Path("skills/my_skill/manifest.yaml"))Waliduje parametry wejściowe.
errors = manifest.validate_input({"network": 123}) # type error
# ["network: expected string, got int"]Generuje kod scaffold dla LLM.
from cores.v1.base_skill import generate_scaffold
code = generate_scaffold(manifest)
# Returns complete Python skill templateTIER_FREE = "free"
TIER_LOCAL = "local"
TIER_PAID = "paid"
COOLDOWN_RATE_LIMIT = 60 # seconds
MAX_EVO_ITERATIONS = 5
INTENT_MODEL_MAX_PARAMS = 3.0 # max params for local intent classificationfrom cores.v1.quality_gate import SkillQualityGate, QualityReport
qg = SkillQualityGate(preflight_checker)5-etapowa ocena jakości skillu.
report = qg.evaluate(
skill_path=Path("skills/tts/providers/piper/v1"),
name="tts"
)
# QualityReport fields:
# - score: 0.0-1.0 (weighted sum of checks)
# - passed: list of passed check names
# - failed: list of failed check names
# - warnings: list of warnings
# - details: dict with per-check results
print(f"Quality score: {report.score:.2f}")
print(f"Passed: {report.passed}")
print(f"Failed: {report.failed}")Wagi oceny:
| Check | Weight |
|---|---|
| preflight | 0.30 |
| health_check | 0.15 |
| test_exec | 0.30 |
| output_valid | 0.15 |
| code_quality | 0.10 |
from cores.v1.repair_journal import RepairJournal
journal = RepairJournal()Rejestruje próbę naprawy i uczy się z wyniku.
journal.record_attempt(
skill="tts",
error="ModuleNotFoundError: No module named 'piper'",
fix_type="pip_install",
fix_command="pip install piper-tts",
success=True
)Znajduje najlepszą znaną naprawę dla błędu.
fix = journal.get_known_fix("ImportError: No module named 'requests'")
if fix:
print(f"Known fix: {fix.fix_type} (confidence: {fix.confidence:.2f})")
# Returns fix with highest success/total ratioPełny cykl: zapytaj LLM → spróbuj naprawy → zapisz wynik.
success = journal.ask_llm_and_try("tts", "SyntaxError: invalid syntax")from cores.v1.stable_snapshot import StableSnapshot
snapshot = StableSnapshot(skills_dir=Path("skills"))Promuje wersję latest do stable (archiwizuje starą stable).
snapshot.save_as_stable("tts", provider="piper")
# skills/tts/providers/piper/latest/ → skills/tts/providers/piper/stable/Tworzy branch od stable (bugfix_YYYYMMDD_HHMMSS lub feature_...).
snapshot.create_branch("tts", "bugfix", provider="piper")
# Creates: skills/tts/providers/piper/branches/bugfix_20260304_161500/Przywraca stable jako latest (rollback).
snapshot.restore_stable("tts", provider="piper")from cores.v1.self_reflection import SelfReflection
reflection = SelfReflection(llm_client, skill_manager, repair_journal)Rejestruje wynik i triggeruje diagnostykę po 3 porażkach.
reflection.record_skill_outcome(
skill="tts",
success=False,
partial=False,
error="Connection timeout"
)
# Auto-triggers run_diagnostic() + attempt_auto_fix() after 3 failuresPełna diagnostyka systemu (7 checks + LLM analysis).
report = reflection.run_diagnostic("tts", "Audio playback failed")
# DiagnosisReport fields:
# - findings: list of detected issues
# - auto_fixable: bool - czy można auto-naprawić
# - requires_user: bool - czy wymaga interwencji użytkownika
# - recommendations: list of suggested actionsfrom cores.v1.bandit_selector import UCB1BanditSelector
bandit = UCB1BanditSelector()Wybiera providera używając algorytmu UCB1.
selected = bandit.select(
capability="tts",
providers=["piper", "pyttsx3"],
base_scores={"piper": 0.9, "pyttsx3": 0.5}
)
# UCB1 = mean_reward * 0.6 + exploration * 0.3 + base_score * 0.1Rejestruje wynik dla uczenia bandita.
bandit.record("tts", "piper", reward=1.0, success=True)from cores.v1.adaptive_monitor import AdaptiveResourceMonitor
monitor = AdaptiveResourceMonitor()Uruchamia/zatrzymuje monitoring w tle.
monitor.start(interval_s=5.0) # próbkowanie co 5s
# ... działa w tle ...
monitor.stop()Zwraca obciążenie systemu 0.0-1.0.
score = monitor.pressure_score()
# 0.0 = idle, 1.0 = critical
# Weighted: CPU 0.3 + RAM 0.5 + disk 0.2Wykrywa trend: "rising" | "falling" | "stable".
trend = monitor.trend() # "rising" = rosnące obciążeniefrom cores.v1.proactive_scheduler import ProactiveScheduler, setup_default_tasks
scheduler = ProactiveScheduler()
setup_default_tasks(scheduler, monitor, gc, sm)Rejestruje zadanie periodyczne.
def my_task():
print("Running periodic task")
scheduler.register("my_task", my_task, interval_s=60)Uruchamia/zatrzymuje scheduler (tick loop 1s).
scheduler.start() # startuje w tle
# ...
scheduler.stop()from cores.v1.utils import clean_code, clean_json
# Remove markdown fences
code = clean_code("```python\nprint('hello')\n```")
# Returns: "print('hello')"
# Parse JSON
json_obj = clean_json('{"key": "value"}')from cores.v1.config import C
print(f"{C.GREEN}Success!{C.RESET}")
print(f"{C.RED}Error!{C.RESET}")
print(f"{C.DIM}Muted text{C.RESET}")