Budowanie agenta AI, który nigdy nie zapomina (Część 2): Teleportacja
Część 2 agenta AI, który nigdy nie zapomina: teleportacja pamięci między sesjami, umiejętności, proces degradacji pamięci oraz pamięć ograniczona do konkretnych gałęzi dla Claude Code i podobnych agentów.
Poniższe notatki przedstawiają praktyczny plan działania dotyczący tematu „Stworzenie agenta AI, który nigdy nie zapomina (Część 2): Teleportacja, umiejętności, degradacja i pamięć skalowana według gałęzi”. Nacisk kładziony jest na umowy, sprawdzania oraz miejsca na kod do wstawienia, a nie na motywacyjne aspekty. Podczas przechodzenia przez etap przeglądu najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się od demonstracji do środowisk współdzielonych.
Gdzie przerwaliśmy
Scenariusz „Where We Left Off” funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres pracy. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny poufnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione bloki danych ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwę w kontynuacji pracy po zakłóceniach.
Krok 5: Teleportacja pamięci
Etap 5 – teleportacja pamięci – działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden przykład udanego działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres.
Proces pracy
Faza procesu pracy funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakaś etap zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwę w kontynuacji pracy po zakłóceniach.
Agent A: snapshot("for_review.vdb")
| send file to Agent B
v
Agent B: open("for_review.vdb") -> work -> add new memories
| send file back
v
Agent A: merge("for_review.vdb")
-> new memories imported
-> conflicts detected and resolved
Faza procesu pracy funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z wersji demonstracyjnej do środowisk współdzielonych.
Zdjęcie obrazowe
W fazie Snapshot należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać oddzielnie od kodu aplikacji. Pliki środowiskowe, magazyny tajemnic oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Wprowadź ludzką aprobatę dla operacji, które wiążą się z wydawaniem pieniędzy lub modyfikacją danych produkcyjnych. Połączenia ustalone w czasie kompilacji nie równają się pełności funkcjonalności biznesowej.
class MemoryTeleport:
def __init__(self, db: vectlite.Database):
self.db = db
def snapshot(self, path: str) -> str:
"""Create a portable copy of the entire database."""
self.db.snapshot(path)
return path
Diff — przegląd bez modyfikacji
Aby uzyskać wstępny przegląd różnic bez modyfikacji etapu, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa nieudanych transakcji stanowią część produktu, a nie elementy dodawane później. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności funkcjonalności biznesowej.
def diff(self, remote_path: str,
namespaces: list[str] | None = None) -> MergeResult:
"""Preview what a merge would do, without writing anything."""
return self._merge_internal(remote_path, namespaces, dry_run=True)
Fuzja — najciekawsza część
W fazie „Merge the interesting part” należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki nad rozbudowanymi skryptami. Gdy jakiś krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności biznesowej. W fazie „Merge the interesting part” należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z wersji demonstracyjnej do środowiska współdzielonego.
Środowiska.
def merge(self, remote_path: str,
namespaces: list[str] | None = None,
strategy: ConflictStrategy = ConflictStrategy.NEWER_WINS
) -> MergeResult:
"""Merge memories from a remote .vdb into the local database."""
return self._merge_internal(
remote_path, namespaces,
dry_run=False, strategy=strategy,
)
Strategie rozwiązywania konfliktów
Podczas prace nad etapem strategii rozwiązywania konfliktów najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. System powrotu nie powinien ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator ponawia próbę z późniejszym węzłem.
class ConflictStrategy(str, Enum):
LOCAL_WINS = "local_wins" # keep local version
REMOTE_WINS = "remote_wins" # keep remote version
NEWER_WINS = "newer_wins" # keep whichever has a more recent updated_at
KEEP_BOTH = "keep_both" # import remote as a new record
Motor łączenia
Gdy przechodzisz przez etap silnika łączenia, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponowne, kontrola przez ludzi oraz obsługa wiadomości błędowych są częścią produktu, a nie elementem późniejszej dopracowywania. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
def _merge_internal(self, remote_path, namespaces=None,
dry_run=False, strategy=ConflictStrategy.NEWER_WINS):
result = MergeResult()
remote_db = vectlite.open(remote_path, dimension=DIMENSION)
try:
ns_list = namespaces or self._discover_namespaces(remote_db)
local_index = self._build_index(self.db, ns_list) for ns in ns_list:
remote_records = remote_db.list(namespace=ns, limit=10_000) for record in remote_records:
rec_id = record["id"]
remote_meta = record.get("metadata", {})
remote_hash = remote_meta.get("content_hash", "") if rec_id in local_index:
local_info = local_index[rec_id] if local_info["content_hash"] == remote_hash:
result.skipped_duplicates += 1 # same content
continue # Conflict -- same ID, different content
conflict = MergeConflict(
record_id=rec_id,
namespace=ns,
local_content=local_info.get("content", ""),
remote_content=remote_meta.get("content", ""),
local_updated=local_info.get("updated_at", ""),
remote_updated=remote_meta.get("updated_at", ""),
)
if not dry_run:
self._resolve_conflict(conflict, record, strategy)
result.conflict_resolutions += 1
result.conflicts.append(conflict)
else:
# New record -- import
if not dry_run:
self._import_record(record, ns)
result.imported += 1
finally:
remote_db.close() return result
Zobaczenie, jak to działa
Gdy przechodzisz przez etap „Widzenie, jak to działa”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki od rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustalaj punkty kontrolne po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą funkcję LLM, gdy operator próbuje ponownie uruchomić późniejszy element.
from agentmem.store.memory_store import MemoryStore
from agentmem.services.teleport import MemoryTeleport
# Agent A: has been working for weeks
store_a = MemoryStore("agent_a.vdb")
tp = MemoryTeleport(store_a.db)# Create a snapshot for Agent B
tp.snapshot("for_agent_b.vdb")# ... send for_agent_b.vdb to Agent B's machine ...# Agent B works with the snapshot, adds new memories
store_b = MemoryStore("for_agent_b.vdb")
store_b.store(Memory(
id="mem_new_finding",
content="The payment API requires idempotency keys for all POST requests",
memory_type=MemoryType.REFERENCE,
importance=0.8,
))
store_b.save()
store_b.close()# ... send for_agent_b.vdb back to Agent A ...# Agent A merges the new knowledge
result = tp.merge("for_agent_b.vdb")
print(result.summary())
# -> Imported: 1 new memories
# Skipped: 47 duplicates
# Conflicts: 0
Gdy przechodzisz przez etap „Widzenie, jak to działa”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do wspólnych środowisk.
Dlaczego format transportowy .vdb działa skutecznie
Najlepiej sprawdza się, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę o cofnięciu działań, zanim rozszerzysz zakres. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny poufnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury. Utrzymuj stan struktury w prostym formacie i z określonym typem danych. Wplecione bloki danych ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwę w kontynuacji działania po zakłóceniach.
Krok 6: Silnik umiejętności
Faza Engine Umiejętności z kroku 6 działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zdokumentuj zarówno ścieżkę pomyślnego przebiegu, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane w danym polu, i powodują przerwanie kontynuacji po przerwach.
---
description: Docker best practices
glob_patterns: ["Dockerfile", "docker-compose*.yml"]
trigger_keywords: ["docker", "container", "compose"]
---
## Docker Guidelines
- Always use multi-stage builds
- Pin base image versions
- Never run as root in production
Trzy sygnały aktywacji
Faza trzech sygnałów aktywacji funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwę w kontynuacji działania po zakłóceniach.
User message: "deploy the containers to staging"
|
v
+-- SkillEngine.activate() ----------------------------+
| |
| 1. Hybrid search (semantic + keyword via VectLite) |
| -> matches "docker" skill (score 0.91) |
| -> matches "devops" skill (score 0.78) |
| |
| 2. Glob match (if file path detected in message) |
| -> "Dockerfile" mentioned? -> boost "docker" |
| |
| 3. Keyword boost |
| -> "containers" matches docker triggers |
| |
| 4. Weighted combination -> top-k activated skills |
+------------------------------------------------------+
|
v
Injected into system prompt as ## Active skills
Faza trzech sygnałów aktywacji funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych.
Model danych umiejętności
W fazie modelu danych The Skill należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. W przypadku, gdy następnym krokiem jest kod lub wywołanie narzędzia, lepiej używać ustrukturyzowanych wyników z walidacją schematu niż tekstu w formie swobodnej.
@dataclass
class Skill:
name: str # "docker", "python", "security"
content: str # Full Markdown instructions
glob_patterns: list[str] # ["Dockerfile", "*.py"]
trigger_keywords: list[str] # ["container", "docker"]
tools_allowed: list[str] # ["docker_build", "docker_push"]
importance: float = 0.7 # Base weight
tags: list[str] = field(default_factory=list)
description: str = ""
Aby sklasyfikować umiejętności indeksowania na danej fazie rejestracji, należy zdefiniować dane wejściowe, osobę odpowiedzialną za daną czynność oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić tę czynność od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżki naprawcze. Próby ponownych działań, kontrola przez ludzi oraz obsługa nieudanych przesyłek stanowią część produktu, a nie elementy dodawane później. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności funkcjonalności biznesowej.
NAMESPACE = "__skills__"
class SkillEngine:
def __init__(self, db: vectlite.Database,
max_active: int = 3, min_score: float = 0.30):
self.db = db
self.max_active = max_active
self.min_score = min_score def register(self, skill: Skill) -> str:
skill_id = f"skill_{skill.name}_{uuid.uuid4().hex[:6]}" # Rich embedding text: name + description + keywords + content
embed_text = (
f"{skill.name}: {skill.description or ''} "
f"{' '.join(skill.trigger_keywords)} "
f"{skill.content[:500]}"
)
vector = embed(embed_text)
sparse = vectlite.sparse_terms(embed_text) self.db.upsert(
skill_id, vector, skill.to_metadata(),
sparse=sparse, namespace=NAMESPACE,
)
return skill_id
Aktywacja — podstawowa logika dopasowywania
W fazie podstawowego dopasowywania w procesie aktywacji należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie gwarantują kompletności rozwiązania biznesowego. W fazie podstawowego dopasowywania w procesie aktywacji należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z wersji demonstracyjnej do wersji wspólnej.
środowisk.def activate(self, message: str, file_path: str | None = None,
k: int | None = None) -> list[ActivatedSkill]:
k = k or self.max_active
# 1. Hybrid search via VectLite
vector = embed(message)
sparse = vectlite.sparse_terms(message)
results = self.db.search(
vector,
k=min(k * 3, self.count()), # over-fetch for re-ranking
sparse=sparse, fusion="rrf",
namespace=NAMESPACE,
) # Build candidates with normalized semantic scores
candidates = {}
max_score = max((r.get("score", 0) for r in results), default=1) or 1
for r in results:
meta = r.get("metadata", {})
name = meta.get("name", "")
if name:
candidates[name] = _Candidate(
skill=self._meta_to_skill(meta),
semantic_score=r.get("score", 0) / max_score,
) # 2. Keyword boost
msg_lower = message.lower()
for name, cand in candidates.items():
hits = sum(1 for kw in cand.skill.trigger_keywords
if kw.lower() in msg_lower)
if hits > 0:
cand.keyword_score = min(0.5, hits * 0.2) # 3. Glob match
detected_paths = self._detect_paths(message)
if file_path:
detected_paths.append(file_path)
for name, cand in candidates.items():
for pattern in cand.skill.glob_patterns:
for path in detected_paths:
if fnmatch.fnmatch(os.path.basename(path), pattern):
cand.glob_score = 0.35
break # 4. Weighted combination
activated = []
for name, cand in candidates.items():
final_score = (
cand.semantic_score * 0.55
+ cand.keyword_score * 0.25
+ cand.glob_score * 0.15
+ cand.skill.importance * 0.05
)
if final_score >= self.min_score:
activated.append(ActivatedSkill(
skill=cand.skill,
score=final_score,
activated_by=..., # "semantic", "keyword", "glob", "combined"
)) activated.sort(key=lambda a: a.score, reverse=True) # Gap filter: drop skills scoring < 75% of top skill
if activated:
top = activated[0].score
activated = [a for a in activated if a.score >= top * 0.75] return activated[:k]
Ładowanie umiejętności z plików Markdown
Podczas przechodzenia przez etap ładowania umiejętności z plików Markdown najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. System powrotu nie powinien ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
def load_from_directory(self, dir_path: str) -> list[str]:
"""Load skills from .md files with optional YAML-like frontmatter."""
ids = []
for md_file in sorted(Path(dir_path).glob("*.md")):
skill = self._parse_skill_file(md_file)
if skill:
ids.append(self.register(skill))
return ids
---
description: Python best practices
glob_patterns: ["*.py", "*.pyx", "pyproject.toml"]
trigger_keywords: ["python", "pip", "venv", "pytest"]
tools_allowed: ["python_run", "pytest"]
importance: 0.8
tags: ["language", "backend"]
---
## Python Guidelines
- Use type hints for all function signatures
- Prefer pathlib over os.path
- Use pytest with fixtures, not unittest
- Format with black, lint with ruff
Iniekcja promptu
Gdy przechodzisz przez etap iniekcji promptu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponowne, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Zachowuj w pamięci podręcznej stabilne instrukcje systemu oraz schematy narzędzi. Ponowne wysyłanie identycznego prefiksu jest częstą przyczyną problemów.
def format_for_prompt(self, activated: list[ActivatedSkill]) -> str:
if not activated:
return ""
lines = ["\n\n## Active skills\n"]
for act in activated:
lines.append(f"### {act.skill.name}")
if act.skill.tools_allowed:
lines.append(f"**Allowed tools:** {', '.join(act.skill.tools_allowed)}")
lines.append(act.skill.content)
lines.append("") return "\n".join(lines)
Zobaczenie, jak to działa
Gdy przechodzisz przez etap „Widzenie, jak to działa”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki od rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustalaj punkty kontrolne po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą funkcję LLM, gdy operator próbuje ponownie uruchomić późniejszy element.
engine = SkillEngine(db)
engine.register(Skill(
name="docker",
content="## Docker\n- Always use multi-stage builds\n- Pin versions",
glob_patterns=["Dockerfile", "docker-compose*.yml"],
trigger_keywords=["docker", "container", "image", "compose"],
))engine.register(Skill(
name="python",
content="## Python\n- Use type hints\n- Prefer pathlib\n- Use pytest",
glob_patterns=["*.py", "pyproject.toml"],
trigger_keywords=["python", "pip", "venv"],
))# Semantic + keyword match
activated = engine.activate("deploy the containers to staging")
# -> [ActivatedSkill(skill=<docker>, score=0.87, activated_by="combined")]# Glob match
activated = engine.activate("review the Dockerfile")
# -> [ActivatedSkill(skill=<docker>, score=0.91, activated_by="combined")]# Semantic match only (no exact keywords)
activated = engine.activate("optimize the CI/CD pipeline for faster builds")
# -> [ActivatedSkill(skill=<docker>, score=0.52, activated_by="semantic")]
Gdy przechodzisz przez etap „Widzenie, jak to działa”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co się dzieje w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.
Krok 7: Degradacja pamięci i rozwiązywanie konfliktów
Etap degradacji pamięci w Kroku 7 działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres analizy. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione dane ukrywają informację o tym, który węzeł zapisał dany pole, co utrudnia kontynuację pracy po przerwach.
Wzór na degradację
Formuła degradacji funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia. Zapisz jeden udany przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres badania. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później w celu udoskonalenia. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji działania po zakłóceniach.
new_importance = current_importance * 2^(-elapsed_hours / half_life)
The MemoryDecayEngine
Etap MemoryDecayEngine funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co utrudnia kontynuację pracy po przerwach.
class MemoryDecayEngine:
def __init__(self, store: MemoryStore,
half_life_hours: float = 168.0, # 1 week
tombstone_threshold: float = 0.05,
conflict_threshold: float = 0.80):
self.store = store
self.half_life_hours = half_life_hours
self.tombstone_threshold = tombstone_threshold
self.conflict_threshold = conflict_threshold
Zastosowanie efektu degradacji
Faza stosowania degradacji działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden „złoty” przekaz, jeden przypadek awarii oraz notatkę o cofnięciu działań, zanim rozszerzysz zakres. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej typowości. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach.
def apply_decay(self, now: datetime | None = None) -> DecayReport:
"""Run exponential decay over all memories."""
now = now or datetime.now(timezone.utc)
report = DecayReport()
for mtype in MemoryType:
memories = self.store.recall_by_type(mtype, k=500)
for mem in memories:
if mem.superseded_by:
continue # already replaced report.total_scanned += 1
last_accessed = self._parse_time(
mem.last_accessed or mem.created_at
)
elapsed_hours = max(
(now - last_accessed).total_seconds() / 3600.0, 0.0
) # The core formula
decay_factor = math.pow(
2.0, -elapsed_hours / self.half_life_hours
)
new_importance = mem.importance * decay_factor if new_importance < self.tombstone_threshold:
# Below threshold -> auto-delete
self.store.update_importance(mem.id, 0.0)
self.store.forget(mem.id)
self.store.compact()
report.pruned += 1
elif abs(new_importance - mem.importance) > 0.001:
# Importance decayed -> update
self.store.update_importance(mem.id, new_importance)
report.decayed += 1
else:
report.untouched += 1 return report
Detekcja konfliktów
Etap wykrywania konfliktów działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres badania. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane w danym polu, i powodują przerwanie kontynuacji po przerwach.
def check_conflicts(self, new_memory: Memory) -> list[ConflictResult]:
"""Find existing memories that contradict the new one."""
# Get candidates via VectLite search
similar = self.store.find_similar(
new_memory.content,
memory_type=new_memory.memory_type,
k=5, min_score=0.0,
) new_hash = self.store._hash(new_memory.content)
new_vector = embed(new_memory.content) conflicts = []
for existing, _ in similar:
if existing.id == new_memory.id:
continue
if existing.superseded_by:
continue # Skip exact duplicates (same hash = not a conflict)
if self.store._hash(existing.content) == new_hash:
continue # Compute direct cosine similarity
existing_vector = embed(existing.content)
cosine_sim = self._cosine_similarity(new_vector, existing_vector) if cosine_sim >= self.conflict_threshold:
conflicts.append(ConflictResult(
existing=existing,
new_content=new_memory.content,
similarity=cosine_sim,
resolution=self.default_resolution,
)) return conflicts
Miejsce przechowywania z rozwiązywaniem konfliktów
Sklep z etapem rozwiązywania konfliktów funkcjonuje najlepiej, gdy traktowany jest jako mierzalna struktura. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione bloki danych ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwanie kontynuacji działania po zakłóceniach.
def store_with_conflict_check(self, memory: Memory,
resolution=None) -> tuple[str, list]:
resolution = resolution or self.default_resolution
conflicts = self.check_conflicts(memory)
if not conflicts:
return self.store.store(memory), [] for conflict in conflicts:
conflict.resolution = resolution if resolution == ConflictResolution.SUPERSEDE:
mem_id = self.store.store(memory)
self.store.mark_superseded(conflict.existing.id, mem_id) elif resolution == ConflictResolution.KEEP_BOTH:
mem_id = self.store.store(memory) elif resolution == ConflictResolution.KEEP_EXISTING:
mem_id = "" # don't store the new one return mem_id, conflicts
Zobacz, jak to działa
Etap obserwacji działania funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden przykład udanego działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zdokumentuj zarówno ścieżkę udanego działania, jak i ścieżkę przywracania do stanu poprzedniego. Próby ponownych działań, mechanizmy kontroli ludzkiej oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później. Utrzymuj stan grafu w formie prostych, spójnych struktur typowych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwanie kontynuacji działania po zakłóceniach.
store = MemoryStore("decay_demo.vdb")
decay = MemoryDecayEngine(store)
# Store two memories
store.store(Memory(id="m1", content="The database is PostgreSQL 14",
memory_type=MemoryType.PROJECT, importance=0.7))# Later, the project migrates
new_mem = Memory(id="m2", content="The database is PostgreSQL 16",
memory_type=MemoryType.PROJECT, importance=0.8)mem_id, conflicts = decay.store_with_conflict_check(new_mem)
print(conflicts[0])
# -> Conflict: [project] sim=0.953 -> supersede
# m1 is now marked as superseded by m2# Weeks later, run decay
report = decay.apply_decay()
print(report)
# -> Decay: 15 scanned, 8 decayed, 2 pruned, 5 untouched
Krok 8: Pamięć świadoma struktury pracy
Faza pamięci świadomej struktury roboczej z kroku 8 działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy któryś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję działań. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwę w kontynuacji działania po przerwach. Faza pamięci świadomej struktury roboczej z kroku 8 działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.
Strategia przestrzeni nazw
W fazie strategii przestrzeni nazw należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zatwierdzenie przez człowieka powinno być wymagane dla operacji, które wiążą się z wydawaniem pieniędzy lub modyfikacją danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie równają się pełności funkcjonalności biznesowej.
# Global -- shared across all branches
GLOBAL_TYPES = {MemoryType.USER, MemoryType.FEEDBACK}
# Scoped -- isolated per branch
SCOPED_TYPES = {MemoryType.PROJECT, MemoryType.REFERENCE}
Memory Type Branch VectLite Namespace
-----------------------------------------------------
user (any) "user"
feedback (any) "feedback"
project main "project:main"
project feature/auth "project:feature/auth"
reference main "reference:main"
reference feature/auth "reference:feature/auth"
The WorktreeManager
Dla etapu The WorktreeManager należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno standardową ścieżkę działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później. Konieczne jest uzyskanie zatwierdzenia człowieka w przypadku operacji, które wiążą się z wydatkami lub zmianą danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.
BRANCH_SEP = ":"
class WorktreeManager:
def __init__(self, db: vectlite.Database,
branch: str | None = None):
self.db = db
self._branch = branch or self._detect_git_branch() def resolve_namespace(self, memory_type: MemoryType) -> str:
"""Map a MemoryType to its VectLite namespace."""
if memory_type in GLOBAL_TYPES:
return memory_type.value # "user", "feedback"
return f"{memory_type.value}{BRANCH_SEP}{self._branch}"
# -> "project:main", "reference:feature/auth" def recall_namespaces(self) -> list[str]:
"""All namespaces to search for the current branch."""
namespaces = []
for mt in MemoryType:
if mt in GLOBAL_TYPES:
namespaces.append(mt.value)
else:
namespaces.append(f"{mt.value}{BRANCH_SEP}{self._branch}")
return namespaces
# -> ["user", "feedback", "project:main", "reference:main"]
Autodetekcja gałęzi git
Aby automatycznie wykrywać etap gałęzi git, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną przyczynę, a nie na skomplikowany łańcuch operacji. Zatwierdzenie przez człowieka powinno być wymagane w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.
@staticmethod
def _detect_git_branch() -> str:
"""Auto-detect the current git branch."""
try:
result = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True, text=True, timeout=5,
)
if result.returncode == 0:
branch = result.stdout.strip()
if branch and branch != "HEAD":
return branch
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return "main" # fallback
Aby automatycznie wykrywać etap gałęzi git, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.
Zmiana gałęzi
Gdy przechodzisz przez etap przekształcania gałęzi, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
def switch_branch(self, branch: str) -> str:
"""Switch to a different branch context. Returns previous branch."""
prev = self._branch
self._branch = branch
return prev
Łączenie gałęzi (jak git merge dla pamięci)
Gdy pracujesz nad łączeniem gałęzi, tak jak w narzędziu git stage, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponowne, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Ustalaj punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
def merge_branch(self, source: str,
into: str | None = None) -> BranchMergeResult:
"""Merge branch-scoped memories from source into target."""
target = into or self._branch
result = BranchMergeResult(source=source, target=target)
for mt in SCOPED_TYPES: # only project + reference
src_ns = f"{mt.value}{BRANCH_SEP}{source}"
tgt_ns = f"{mt.value}{BRANCH_SEP}{target}" src_records = self.db.list(namespace=src_ns, limit=500) # Build dedup set from target
tgt_records = self.db.list(namespace=tgt_ns, limit=500)
tgt_hashes = {
r.get("metadata", {}).get("content_hash", "")
for r in tgt_records
} for record in src_records:
content_hash = record.get("metadata", {}).get("content_hash", "")
if content_hash in tgt_hashes:
result.skipped_duplicates += 1
continue # Import with new ID into target namespace
new_id = f"mem_{uuid.uuid4().hex[:12]}"
new_meta = dict(record.get("metadata", {}))
new_meta["merged_from"] = source self.db.upsert(
new_id, record["vector"], new_meta,
sparse=record.get("sparse", {}),
namespace=tgt_ns,
)
result.imported += 1 return result
Różnice i usuwanie
Gdy przechodzisz przez etap Diff i usuwania, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustalaj punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłaty za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy przechodzisz przez etap Diff i usuwania, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do wspólnych środowisk.
def diff_branch(self, source: str,
target: str | None = None) -> BranchDiffResult:
"""Like git diff -- preview without modifying anything."""
# Same logic as merge_branch but without writing
...
def delete_branch(self, branch: str) -> int:
"""Delete all branch-scoped memories. Global types untouched."""
total_deleted = 0
for mt in SCOPED_TYPES:
ns = f"{mt.value}{BRANCH_SEP}{branch}"
records = self.db.list(namespace=ns, limit=500)
for record in records:
self.db.delete(record["id"])
total_deleted += 1
if total_deleted > 0:
self.db.compact()
return total_deleted
Zobaczenie całego cyklu życia
Etap „Zobaczenie całego cyklu życia” funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia do analizy. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres projektu. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą audytować bez konieczności czytania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione dane ukrywają informację o tym, który węzeł zapisał dany pole, co utrudnia kontynuację pracy po przerwach.
wt = WorktreeManager(db, branch="main")
# Store memories on main
store.store(Memory(content="FastAPI backend", memory_type=MemoryType.PROJECT))
# -> stored in namespace "project:main"store.store(Memory(content="I prefer type hints", memory_type=MemoryType.USER))
# -> stored in namespace "user" (global)# Switch to feature branch
wt.switch_branch("feature/auth")store.store(Memory(content="OAuth2 with PKCE for mobile clients",
memory_type=MemoryType.PROJECT))
# -> stored in namespace "project:feature/auth"# Recall on feature/auth -- sees global + branch-scoped
namespaces = wt.recall_namespaces()
# -> ["user", "feedback", "project:feature/auth", "reference:feature/auth"]
# "I prefer type hints" is visible (global)
# "FastAPI backend" is NOT visible (it's in project:main)
# "OAuth2 with PKCE" is visible (feature branch)# Merge feature into main
result = wt.merge_branch("feature/auth", into="main")
# -> Imported: 1, Skipped: 0# Now on main, "OAuth2 with PKCE" is also visible
wt.switch_branch("main")
# -> recall sees: "FastAPI backend" + "OAuth2 with PKCE" + all globals# Clean up
wt.delete_branch("feature/auth")
# -> project:feature/auth and reference:feature/auth emptied
Ostateczna struktura projektu
Najlepiej funkcjonuje końcowa faza struktury projektu, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres projektu. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę przywracania do stanu poprzedniego. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później w celu udoskonalenia. Utrzymuj stan grafu w formie prostych, spójnych struktur typowych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwanie kontynuacji działania po przerwach.
agentmem/
core/
memory_types.py # MemoryType + Memory dataclass
embedder.py # embed() / embed_batch()
store/
memory_store.py # MemoryStore -- VectLite persistence
services/
extractor.py # Background memory extraction
session.py # Session summary
compaction.py # 4-level cascading compaction
swarm.py # Multi-agent swarm
teleport.py # Snapshot / merge / diff
skills.py # Skill engine
decay.py # Exponential decay + conflict resolution
worktree.py # Branch-scoped namespaces
agent.py # Agent -- the full loop
mcp_server.py # MCP server (bonus)
Pełny obraz
Faza The Complete Picture funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia. Zapisz jeden „złoty” zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwania w kontynuacji pracy po przerwach. Faza The Complete Picture funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia. Zapisz jeden „złoty” zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych.
User message
|
v
+-- Agent.chat() ----------------------------------------+
| |
| RECALL: hybrid search (HNSW + BM25 + RRF) [P1] |
| SKILLS: semantic + keyword + glob activation [P2] |
| BUILD PROMPT: memories + skills + session summary |
| COMPACT: snip -> micro -> collapse -> auto [P1] |
| LLM RESPONSE |
| EXTRACT: background sub-agent learns facts [P1] |
| SESSION: update summary if threshold crossed [P1] |
| |
| TELEPORT: snapshot / merge / diff [P2] |
| DECAY: exponential forgetting + conflict detect [P2] |
| WORKTREE: branch-scoped namespaces [P2] |
| SWARM: multi-agent via shared namespaces [P1] |
| |
| Storage: single .vdb file (VectLite) |
+--------------------------------------------------------+
Czym umożliwia VectLite
Aby określić etap, który umożliwia What VectLite, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny poufnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Wprowadź procedurę zatwierdzenia przez człowieka dla operacji, które wiążą się z wydatkami lub zmianami w danych produkcyjnych. Połączenia ustalone w czasie kompilacji nie równają się pełności funkcjonalności biznesowej.
Jak to porównuje się z Claude Code
Aby określić, jak to funkcjonuje w porównaniu z etapem realizacji, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanej punktacji kontrolnej, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżki naprawcze. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później. Konieczne jest uzyskanie zatwierdzenia człowieka w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie równają się pełnej kompletności funkcjonalności biznesowej.
Spróbuj
W fazie „Spróbuj” należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Lepiej używać małych, łatwych do przetestowania jednostek niż rozbudowanych skryptów. Gdy jakiś krok zawiedzie, powinno to wskazywać na konkretną przyczynę, a nie na skomplikowaną strukturę procesu. Konieczna jest ludzka akceptacja w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.
pip install vectlite sentence-transformers
git clone https://github.com/mcsedition-hub/agentmem-vectlite.git
cd agentmem-vectlite
python demo.py
W fazie „Spróbuj” należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Obok wyników funkcjonalnych należy rejestrować czas wykonywania oraz koszt tokenów lub zapytań. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do wspólnych środowisk.
List kontrolny operacyjny
Na etapie listy kontrolnej operacyjnej należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie wykonać dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.
Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania.
Zapewnij ludzką aprobatę w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie równają się pełności funkcjonowania systemu biznesowego.
Napisz krótki przewodnik: jak rotować klucze, jak opróżnić kolej z zadań, jak cofnąć ostatnie operacje importu.
Literatura pokrewna
- Praktyczne notatki: Które agent SDK należy stworzyć? — Szczegółowy przewodnik po Praktycznych notatkach: Które agent SDK należy stworzyć?: kontrakty, sprawdzenia oraz miejsca na kod do wstawienia dla zespołów wdrażających ten wzorzec.
- Praktyczne notatki: Używanie narzędzia agenta i wywoływanie funkcji w produkcji — Kiedy — Szczegółowy przewodnik po Praktycznych notatkach: Używanie narzędzia agenta i wywoływanie funkcji w produkcji — Kiedy: kontrakty, sprawdzenia oraz miejsca na kod do wstawienia dla zespołów wdrażających ten wzorzec.