Создание ИИ-агента, который никогда не забывает (Часть 2): Телепортация
Часть 2 агента ИИ, который никогда не забывает: телепортация памяти между сессиями, навыки, ухудшение качества памяти и память с ограничением по веткам для Claude Code и подобных агентов.
В следующих заметках описывается практический подход к реализации проекта «Создание ИИ-агента, который никогда не забывает (Часть 2): телепортация, навыки, ухудшение качества и память в пределах веток». Основное внимание уделяется контрактам, проверкам и шаблонам кода, которые можно легко вставить, а не мотивирующим аспектам. На этапе обзора сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость в токенах или запросах. Отображение затрат с самого начала предотвращает неожиданные расходы при переходе от демо-версии к общедоступным средам.
Где мы остановились
Механизм «Там, где мы остановились» работает наилучшим образом, когда рассматривается как измеримая структура. Сохраняйте один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф. Сохраняйте состояние графа простым и типизированным. Вложенные структуры данных маскируют информацию о том, какой узел записал тот или иной поле, и приводят к нарушению возобновления работы после перерывов.
Шаг 5: Телепортация памяти
Этап телепортации памяти на шаге 5 работает наилучшим образом, если рассматривать его как измеримую поверхность. Соберите один пример успешного выполнения, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работы. Документируйте одновременно путь успешного выполнения и путь восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. Сохраняйте структуру графа простой и типизированной; вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
Рабочий процесс
Этап рабочего процесса работает наилучшим образом, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретный элемент ответственности, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
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
Этап рабочего процесса работает наилучшим образом, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Записывайте временные показатели, а также стоимость обработки токенов или запросов вместе с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.
Снимок состояния
На этапе создания снимка необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Конфигурацию следует хранить отдельно от кода приложения. Файлы среды, хранилища конфиденциальных данных и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Для операций, связанных с тратой денег или изменением производственных данных, необходимо предусмотреть утверждение человеком. Подключения, созданные во время компиляции, не гарантируют полноты охвата бизнес-логики.
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 — превью без изменений
Для предварительного просмотра Diff без изменения этапа необходимо заранее определить входные данные, ответственного за шаг и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо одновременно задокументировать успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Внедрять утверждение человеком для операций, связанных с тратой денег или изменением производственных данных. Настройка на этапе компиляции не гарантирует полноты функционала продукта.
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)
Слияние — самая интересная часть
На этапе «Интересная часть слияния» необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на запутанную цепочку операций. Внедрять человеческое утверждение для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты бизнес-логики. На этапе «Интересная часть слияния» необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-режима к общей среде.
Стратегии разрешения конфликтов.
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,
)
Стратегии разрешения конфликтов
На этапе разработки стратегий разрешения конфликтов сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Создавайте контрольные точки после дорогостоящих операций. Механизм возобновления работы не должен повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг.
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
Механизм слияния
При работе над этапом движка слияния сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Создавайте контрольные точки после дорогостоящих операций. Механизм возобновления работы не должен повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий этап.
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
Проверка работы
На этапе проверки корректности работы сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые модули большим скриптам. Если какой-то шаг не сработает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Выполняйте проверки после дорогостоящих шагов. Система должна не повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий этап.
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
На этапе проверки корректности работы сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам.
Почему формат передачи .vdb эффективен
Формат vdb является наилучшим для передачи данных, если рассматривать его как измеримую поверхность. Сначала соберите один эталонный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию, прежде чем расширять объем обработки. Храните конфигурацию вне кода приложения: файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф. Состояние графа должно быть простым и типизированным; вложенные структуры данных маскируют информацию о том, какой узел записал тот или иной поле, что приводит к нарушению возобновления работы после перерывов.
Шаг 6: Двигатель навыков
Этап Skill Engine шага 6 работает наилучшим образом, если рассматривать его как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Задокументируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не этапом последующей доработки. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
---
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
Три сигнала активации
Этап трех сигналов активации работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
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
Этап трех сигналов активации работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.
Модель данных навыков
На этапе модели данных The Skill необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Предпочитайте структурированные выходные данные с проверкой схемы вместо свободного текста, когда следующим шагом является выполнение кода или вызов инструмента.
@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 = ""
Регистрация — индексация навыков в VectLite
Для индексации навыков регистрации на данном этапе необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Внедрять утверждение человеком для операций, связанных с тратой денег или изменением производственных данных. Подключение компонентов во время компиляции не гарантирует полноты функционала продукта.
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
Активация — основная логика сопоставления
Для активации основного этапа сопоставления необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на запутанную цепочку операций. Внедрять человеческое утверждение для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты бизнес-логики. Для активации основного этапа сопоставления необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счета при переходе от демо-режима к общему использованию.
среды.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]
Загрузка навыков из файлов Markdown
При работе над этапом загрузки навыков из файлов Markdown сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет избежать ошибок при последующих изменениях кода. Храните конфигурацию вне кода приложения. Файлы с настройками среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Создавайте контрольные точки после дорогостоящих операций. Система возобновления работы не должна повторно оплачивать один и тот же вызов LLM, когда оператор пытается выполнить следующий шаг.
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
Внедрение промптов
При работе над этапом внедрения команды сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и последствия частичной неудачи. Такой список поможет сохранять честность при последующих изменениях кода. Задокументируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Храните в кэше стабильные инструкции системы и схемы инструментов. Пересылка одинакового вступительного текста — частая причина избыточных расходов.
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)
Проверка работы
На этапе «Проверка работы» сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые модули большим скриптам. Если какой-то шаг не сработает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Выполняйте проверки после дорогостоящих шагов. Система должна не повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий этап.
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")]
На этапе «Проверка работы» сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам.
Шаг 7: Ухудшение памяти и разрешение конфликтов
Этап ухудшения памяти в рамках Шага 7 работает наилучшим образом, если рассматриваться как измеримая структура. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема обработки. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф структур. Сохраняйте состояние графа простым и типизированным. Вложенные структуры данных скрывают информацию о том, какой узел записал тот или иной поле, и мешают возобновлению работы после прерываний.
Формула ухудшения
Формула разложения работает наилучшим образом, когда её рассматривают как измеримую поверхность. Соберите один пример успешного выполнения, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работы. Документируйте одновременно путь успешного выполнения и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не этапом последующей доработки. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
new_importance = current_importance * 2^(-elapsed_hours / half_life)
The MemoryDecayEngine
Механизм MemoryDecayEngine работает наилучшим образом, если рассматривать его как измеримую поверхность. Сначала зафиксируйте один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию, прежде чем расширять объем работы. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое какого-либо шага причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению возобновления работы после перерывов.
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
Применение эффекта ухудшения качества
Этап применения процесса разложения работает наилучшим образом, если рассматривать его как измеримую поверхность. Зафиксируйте один «золотой» пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия создаваемым объектам, определите критерии успешного выполнения и не соглашайтесь на молчаливое частичное завершение работы. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных скрывают информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению возобновления работы после перерывов.
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
Обнаружение конфликтов
Этап обнаружения конфликтов работает наилучшим образом, когда его рассматривают как измеримую структуру. Сначала соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию, прежде чем расширять объем работы. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на ранних этапах предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Сохраняйте структуру графа простой и типизированной; вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и могут нарушить возобновление работы после прерываний.
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
Хранилище с механизмом разрешения конфликтов
Магазин с этапом разрешения конфликтов работает наилучшим образом, когда рассматривается как измеримая структура. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работы. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая всю структуру. Сохраняйте состояние структуры простым и типизированным. Вложенные объекты скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
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
Как это работает
Этап проверки корректной работы функционала наиболее эффективен, если рассматривать его как измеримую поверхность. Соберите один пример успешной работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно путь успешной работы и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не этапом последующей доработки. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
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
Шаг 8: Память, учитывающая структуру работы
Этап памяти, учитывающей структуру рабочего дерева из шага 8, работает наилучшим образом, если рассматриваться как измеримая среда. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы вместо обширных скриптов. При сбое какого-либо шага причина должна указывать на конкретный ответственный элемент, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению продолжения работы после перерывов. Этап памяти, учитывающей структуру рабочего дерева из шага 8, работает наилучшим образом, если рассматриваться как измеримая среда. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Записывайте временные показатели, а также стоимость обработки токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные расходы при переходе с демо-среды в общедоступные среды.
Стратегия пространства имён
На этапе стратегии пространства имён необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Вводите утверждение человека для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты охвата бизнес-логики.
# 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
Для этапа The WorktreeManager необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии системы. Необходимо документировать как успешный, так и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Внедрять утверждение человеком для операций, связанных с тратой денег или изменением производственных данных. Настройка на этапе компиляции не гарантирует полноты функционала продукта.
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"]
Автоматическое обнаружение ветки git
Для автоматического определения этапа разветвления git необходимо заранее определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на сложную структуру всей цепочки обработки. Внедрять человеческое утверждение для операций, связанных с расходами или изменением производственных данных. Настройки, выполняемые на этапе компиляции, не гарантируют полноты бизнес-процесса.
@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
Для автоматического определения этапа ветки git необходимо заранее определить параметры ввода, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Регистрируйте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее помогает избежать неожиданных счетов при переходе с демо-среды в общедоступные среды.
Переключение веток
При работе над этапом переключения веток сначала запишите контракт: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Храните конфигурацию отдельно от кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Создавайте контрольные точки после дорогостоящих операций. Функция возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий узел.
def switch_branch(self, branch: str) -> str:
"""Switch to a different branch context. Returns previous branch."""
prev = self._branch
self._branch = branch
return prev
Слияние веток (как git merge для «памяти»)
При работе с функцией объединения веток, такой как git stage, сначала запишите описание процесса: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист поможет сохранять честность последующих изменений в коде. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, проверки со стороны оператора и обработка некорректных сообщений являются частью продукта, а не элементами последующей доработки. Создавайте контрольные точки после дорогостоящих операций. Функция возобновления работы не должна повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг.
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
Различия и удаление
При работе с этапами Diff и удаления сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность последующих изменений в коде. Предпочитайте небольшие, тестируемые модули большим скриптам. Если какой-то шаг не сработает, причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Выполняйте контрольные проверки после дорогостоящих шагов. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели при повторной попытке обработки последующего элемента. При работе с этапами Diff и удаления сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность последующих изменений в коде. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам.
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
Полный жизненный цикл
Этап «Полный жизненный цикл» работает наилучшим образом, если рассматривать его как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая всю структуру. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
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
Итоговая структура проекта
Наилучшим образом этап формирования итоговой структуры проекта функционирует, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объёма работ. Документируйте одновременно успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не этапом последующей доработки. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и нарушают возможность продолжения работы после прерываний.
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)
Полная картина
Этап «The Complete Picture» работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению работоспособности после перерывов. Этап «The Complete Picture» работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные расходы при переходе от демо-среды к общедоступным средам.
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) |
+--------------------------------------------------------+
Что позволяет VectLite
Для этапа, который позволяет использовать What VectLite, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Вводите ручное утверждение для операций, связанных с тратой денег или изменением производственных данных. Подключение на этапе компиляции не гарантирует полноты охвата бизнес-логики.
Как это сравнивается с Claude Code
Чтобы понять, как это сравнивается со стадией разработки, необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо одновременно задокументировать успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки. Внедрять утверждение человеком для тех случаев, когда происходит трата средств или изменение данных в продакшене. Подключение компонентов на этапе компиляции не гарантирует полноты функционала продукта.
Попробуйте сами
На этапе «Попробуйте» необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на сложную цепочку операций. Внедрять человеческое утверждение для операций, связанных с тратой денег или изменением производственных данных. Компиляционная настройка не гарантирует полноты бизнес-процессов.
pip install vectlite sentence-transformers
git clone https://github.com/mcsedition-hub/agentmem-vectlite.git
cd agentmem-vectlite
python demo.py
На этапе «Попробуйте» необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Регистрировать время выполнения, а также стоимость токенов или запросов вместе с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам.
Чек-лист операционной работы
На этапе чек-листа операционной работы необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность повторно выполнить шаг, исходя из известной точки контроля, без необходимости угадывания скрытого состояния.
Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Укажите названия элементов, определите критерии успеха и не допускайте молчаливого частичного завершения работы.
Обеспечьте человеческое утверждение для операций, связанных с тратой денег или изменением производственных данных. Компиляционная настройка не заменяет полноты обработки в бизнес-процессе.
Напишите краткий руководство: как обновлять ключи, как опустошать очередь, как откатить последнюю операцию ввода данных.