Pythonepisodic.py
Copy
1
from openai import OpenAI
2
from dataclasses import dataclass, field, asdict
3
from datetime import datetime
4
import json, os
5
 
6
client = OpenAI()
7
EPISODES_FILE = "episodes.json"
8
 
9
@dataclass
10
class Episode:
11
    id: str
12
    task: str
13
    steps: list[dict] = field(default_factory=list)
14
    outcome: str = ""
15
    success: bool = False
16
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
17
 
18
def load_episodes() -> list[Episode]:
19
    if not os.path.exists(EPISODES_FILE):
20
        return []
21
    with open(EPISODES_FILE) as f:
22
        return [Episode(**e) for e in json.load(f)]
23
 
24
def save_episodes(episodes: list[Episode]):
25
    with open(EPISODES_FILE, "w") as f:
26
        json.dump([asdict(e) for e in episodes], f, indent=2)
27
 
28
def log_episode(task: str, steps: list[dict], outcome: str, success: bool) -> Episode:
29
    episodes = load_episodes()
30
    ep = Episode(id=f"ep_{len(episodes)+1}", task=task, steps=steps, outcome=outcome, success=success)
31
    episodes.append(ep)
32
    save_episodes(episodes)
33
    return ep
34
 
35
def find_relevant(task: str, n: int = 2) -> list[Episode]:
36
    episodes = load_episodes()
37
    if not episodes:
38
        return []
39
    summaries = [f"{i}: {'✓' if e.success else '✗'} Task: {e.task}" for i, e in enumerate(episodes)]
40
    resp = client.chat.completions.create(
41
        model="gpt-4o-mini",
42
        messages=[{"role": "user", "content": f"New task: {task}
43
Episodes:
44
" + "
45
".join(summaries) + f"
46
Return JSON with indices of {n} most relevant: {{"indices": []}}"}],
47
        response_format={"type": "json_object"}
48
    )
49
    indices = json.loads(resp.choices[0].message.content).get("indices", [])
50
    return [episodes[i] for i in indices if i < len(episodes)]
51
 
52
# Demo: log 3 episodes then retrieve relevant one
53
log_episode("Research Python web frameworks", [{"action": "web_search", "result": "FastAPI, Django, Flask"}], "FastAPI is best for APIs", success=True)
54
log_episode("Write REST API endpoint", [{"action": "write_code", "result": "POST /items endpoint created"}], "Endpoint created and tested", success=True)
55
log_episode("Deploy to AWS Lambda", [{"action": "zip_package"}, {"action": "aws deploy"}], "Timeout error on large payload", success=False)
56
 
57
relevant = find_relevant("Build a FastAPI microservice")
58
for ep in relevant:
59
    print(f"Relevant: [{ep.id}] {ep.task} → {'✓' if ep.success else '✗'}")
60