1from openai import OpenAI
2from datetime import datetime
3from dataclasses import dataclass, field
4from typing import Any
5import json
6
7client = OpenAI()
8
9@dataclass
10class Episode:
11 episode_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 metadata: dict = field(default_factory=dict)
18
19 def add_step(self, action: str, result: Any, tool: str | None = None):
20 self.steps.append({"action": action, "result": str(result), "tool": tool, "ts": datetime.now().isoformat()})
21
22 def summarise(self) -> str:
23 return f"Episode '{self.task}' ({'SUCCESS' if self.success else 'FAIL'}): " + " → ".join(s["action"] for s in self.steps)
24
25episodes: list[Episode] = []
26
27def start_episode(task: str) -> Episode:
28 ep = Episode(episode_id=f"ep_{len(episodes)+1}", task=task)
29 episodes.append(ep)
30 return ep
31
32def find_similar_episodes(task: str, n: int = 3) -> list[Episode]:
33 response = client.chat.completions.create(
34 model="gpt-4o-mini",
35 messages=[{
36 "role": "user",
37 "content": f"Task: {task}
38
39Episodes:
40" + "
41".join(f"{i}: {e.summarise()}" for i, e in enumerate(episodes)) + f"
42
43Return the indices of the top {n} most relevant episodes as a JSON list."
44 }],
45 response_format={"type": "json_object"}
46 )
47 indices = json.loads(response.choices[0].message.content).get("indices", [])
48 return [episodes[i] for i in indices if i < len(episodes)]
49
50def run_with_episodic_memory(task: str) -> str:
51 ep = start_episode(task)
52 similar = find_similar_episodes(task)
53 prior_experience = "
54".join(f"- Previous: {s.summarise()}" for s in similar)
55 messages = [
56 {"role": "system", "content": f"You are an agent. Prior relevant experience:
57{prior_experience}"},
58 {"role": "user", "content": task}
59 ]
60 response = client.chat.completions.create(model="gpt-4o", messages=messages)
61 answer = response.choices[0].message.content
62 ep.add_step("generate_answer", answer)
63 ep.outcome = answer
64 ep.success = True
65 return answer
