Практические заметки: Агентные архитектуры — Статья 12: Готовы ли мы к...
Пошаговое руководство по практическим заметкам: агентные архитектуры — Статья 12: Готовы ли мы к использованию контрактов, проверок и готовых блоков кода для команд, применяющих эту паттерн-архитектуру.
В следующих заметках описывается практический подход к решению вопросов, связанных с темой «Агентные архитектуры — Статья 12: Готовы ли мы к системе памяти, ориентированной на агентов?». Основное внимание уделяется контрактам, проверкам и местам для вставки кода, а не мотивационным аспектам. На этапе обзора сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Документируйте как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.
Что вы найдете здесь
Этап «Что вы найдете» работает наилучшим образом, если рассматривать его как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, и приводят к нарушению последовательности выполнения после перерывов.
Проблема памяти агента, сформулированная точно
Этап «Проблема памяти агента» работает наилучшим образом, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма задачи. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия всем элементам, определите критерии успеха и не соглашайтесь на молчаливое частичное выполнение задачи. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению возобновления работы после перерывов.
+------------------------+-----------------------------+-------------------------+
| Human Memory Type | Article 7 Implementation | How It Is Accessed |
+------------------------+-----------------------------+-------------------------+
| Working memory | LangGraph state | Always present in ctx |
| (active context) | MemorySaver checkpoints | |
+------------------------+-----------------------------+-------------------------+
| Episodic memory | DynamoDB + embeddings | Semantic similarity |
| (what happened before) | TTL 90 days | query at run start |
+------------------------+-----------------------------+-------------------------+
| Semantic memory | Bedrock Knowledge Base | Vector search query |
| (domain knowledge) | S3-backed JSONL | at run start |
+------------------------+-----------------------------+-------------------------+
| Procedural memory | DynamoDB validated table | Task type lookup |
| (how to do things) | Success rate tracking | at run start |
+------------------------+-----------------------------+-------------------------+
Пробел 1: Временное восприятие
Этап осознания времени Gap 1 работает наилучшим образом, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счёты при переходе от демо-среды к общедоступным средам. Сохраняйте состояние графа простым и типизированным. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и могут нарушить возобновление работы после прерываний. Этап осознания времени Gap 1 работает наилучшим образом, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно успешный путь выполнения и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.
+----------------------------------+------------------------------------------+
| Temporal Property | What It Enables |
+----------------------------------+------------------------------------------+
| Creation timestamp | Basic recency weighting in retrieval |
+----------------------------------+------------------------------------------+
| Last confirmed timestamp | Distinguish stale from fresh knowledge |
+----------------------------------+------------------------------------------+
| Confidence decay function | Facts become less certain over time |
| | without confirmation |
+----------------------------------+------------------------------------------+
| Version history | Track how understanding of a topic |
| | has evolved across runs |
+----------------------------------+------------------------------------------+
| Temporal context at retrieval | "What did I know about X on this date?" |
| | not just "What do I know about X now?" |
+----------------------------------+------------------------------------------+
# harness/memory/temporal.py
import time
from typing import List
def apply_temporal_weighting(
retrieved_facts: List[dict],
recency_half_life_days: float = 30.0,
) -> List[dict]:
"""
Weights retrieved facts by recency using exponential decay.
Facts confirmed recently score higher than stale ones with
the same semantic similarity.
"""
now = time.time()
half_life_seconds = recency_half_life_days * 86400
for fact in retrieved_facts:
base_score = fact.get("similarity_score", 0.8)
last_confirmed = fact.get("last_confirmed_at", fact.get("created_at", now))
age_seconds = now - last_confirmed
# Exponential decay: score halves every half_life_days
import math
decay_factor = math.exp(-0.693 * age_seconds / half_life_seconds)
fact["temporal_weighted_score"] = base_score * decay_factor
return sorted(retrieved_facts, key=lambda f: f["temporal_weighted_score"], reverse=True)
Gap 2: Ассоциативный поиск
Для этапа ассоциативного поиска Gap 2 необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии системы. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной функцией, а не с запутанной структурой обработки данных. Указывайте те части текста, на которых основан ответ. Без цитат операторы не смогут отличить галлюцинации от проблем с индексацией.
# harness/memory/associative.py
import boto3
from typing import List, Optional
class AssociativeMemoryIndex:
"""
Maintains an association graph between memory entities.
Complements vector search with relationship-based retrieval.
This is a simplified implementation. Production would use
Amazon Neptune or a graph database for complex traversals.
"""
def __init__(
self,
table_name: str = "agent-memory-associations",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def record_association(
self,
entity_a: str,
entity_b: str,
relationship: str,
strength: float = 1.0,
run_id: str = None,
):
"""
Records that two memory entities are related.
entity_a, entity_b: fact_ids, episode_ids, or concept labels
relationship: "co-occurred", "caused", "contradicts", "supports"
strength: 0.0 to 1.0, increases with repeated co-occurrence
"""
import time
self.table.update_item(
Key={"entity_a": entity_a, "entity_b": entity_b},
UpdateExpression=(
"SET relationship = :r, "
"strength = if_not_exists(strength, :z) + :s, "
"occurrence_count = if_not_exists(occurrence_count, :z) + :one, "
"last_seen = :now"
),
ExpressionAttributeValues={
":r": relationship,
":z": 0,
":s": strength,
":one": 1,
":now": int(time.time()),
}
)
def get_associated_entities(
self,
entity: str,
min_strength: float = 0.5,
max_results: int = 10,
) -> List[dict]:
"""Retrieves entities associated with the given entity."""
response = self.table.query(
KeyConditionExpression="entity_a = :e",
FilterExpression="strength >= :s",
ExpressionAttributeValues={":e": entity, ":s": min_strength},
)
items = sorted(
response.get("Items", []),
key=lambda x: x.get("strength", 0),
reverse=True
)
return items[:max_results]
Gap 3: Интеллектуальный подход к пути записи
На этапе Gap 3 Write-Path Intelligence необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Укажите названия элементов, определите критерии успеха и не допускайте молчаливого частичного завершения работы. Внедрите человеческое утверждение для операций, связанных с расходованием средств или изменением производственных данных. Компиляционная настройка не заменяет полноту выполнения бизнес-задач.
# harness/memory/annotation.py
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_aws import ChatBedrock
import boto3
import json
MEMORY_ANNOTATION_PROMPT = """
You have just completed a reasoning step. Before continuing, consider:
1. Did you discover something that would be useful in future runs on similar tasks?
2. Did you encounter a pattern you had not seen before?
3. Did something fail that you want to remember to avoid next time?
If yes to any of these, describe what you want to remember in one or two sentences.
If no, respond with null.
Respond with JSON:
{"worth_remembering": true | false, "annotation": "description or null"}
"""
class InlineMemoryAnnotator:
"""
Runs between agent reasoning steps and asks the agent to flag
anything worth remembering before the run ends.
This is experimental. The risk is that the agent annotates
incorrect conclusions. Pair with the validation gate from Article 7.
"""
def __init__(self, region: str = "us-east-1"):
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.model = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-4-5",
model_kwargs={"temperature": 0, "max_tokens": 256},
)
self._annotations: list = []
def maybe_annotate(self, last_reasoning_step: str) -> bool:
"""
Called after each significant reasoning step.
Returns True if an annotation was recorded.
"""
response = self.model.invoke([
SystemMessage(content=MEMORY_ANNOTATION_PROMPT),
HumanMessage(content=f"Recent reasoning:\n{last_reasoning_step[:1000]}")
])
try:
result = json.loads(response.content)
if result.get("worth_remembering") and result.get("annotation"):
self._annotations.append(result["annotation"])
return True
except json.JSONDecodeError:
pass
return False
def get_annotations(self) -> list:
return list(self._annotations)
Gap 4: Память между агентами с изоляцией
На этапе памяти межагентного взаимодействия Gap 4 необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Регистрируйте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Внедряйте человеческое утверждение для операций, связанных с тратой средств или изменением производственных данных. Настройки во время компиляции не гарантируют полноты функционала продукта. На этапе памяти межагентного взаимодействия Gap 4 необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Документируйте одновременно успешный и восстановительный сценарии работы. Повторные попытки, человеческое утверждение и обработка некорректных сообщений являются частью продукта, а не элементами последующей доработки.
# harness/memory/shared_memory.py
import boto3
from typing import List, Optional
class SharedMemoryPolicy:
"""
Controls which agent types can read and write which memory namespaces.
Sharing is opt-in. Default is isolated per agent_id.
"""
def __init__(self, policies: dict):
"""
policies example:
{
"security_findings": {
"readers": ["supervisor", "security_reviewer", "code_analyst"],
"writers": ["security_reviewer"],
},
"code_patterns": {
"readers": ["supervisor", "code_analyst"],
"writers": ["code_analyst"],
}
}
"""
self.policies = policies
def can_read(self, agent_id: str, namespace: str) -> bool:
policy = self.policies.get(namespace, {})
return agent_id in policy.get("readers", [])
def can_write(self, agent_id: str, namespace: str) -> bool:
policy = self.policies.get(namespace, {})
return agent_id in policy.get("writers", [])
def filter_retrievable(
self,
agent_id: str,
records: List[dict],
) -> List[dict]:
"""
Filters a list of memory records to only those the agent can read.
"""
return [
r for r in records
if self.can_read(agent_id, r.get("namespace", "private"))
or r.get("agent_id") == agent_id # always read own memories
]
Этап 5: Забывание как процесс первого класса
При работе над этапом «Забывание» сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Выполняйте контрольные точки после дорогостоящих шагов. Механизм возобновления работы не должен повторно оплачивать один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий узел.
# harness/memory/intelligent_forgetting.py
import boto3
import time
from typing import List
class IntelligentForgettingManager:
"""
Manages memory removal based on content-aware signals,
not just age. Complements the TTL-based decay from Article 7.
"""
def __init__(
self,
episodic_table: str = "agent-episodic-memory",
semantic_kb_id: str = None,
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.episodic_table = dynamodb.Table(episodic_table)
self.kb_id = semantic_kb_id
self.region = region
def mark_contradicted(
self,
fact_id: str,
contradicting_run_id: str,
contradiction_description: str,
):
"""
Marks a semantic fact as contradicted by newer evidence.
Does not delete immediately: flags for review first.
"""
self.episodic_table.update_item(
Key={"fact_id": fact_id},
UpdateExpression=(
"SET contradicted = :t, "
"contradicted_by = :run, "
"contradiction_note = :note, "
"needs_review = :t"
),
ExpressionAttributeValues={
":t": True,
":run": contradicting_run_id,
":note": contradiction_description,
}
)
def detect_contradictions(
self,
new_fact_content: str,
existing_facts: List[dict],
detector_model,
) -> List[str]:
"""
Checks whether a new fact contradicts existing ones.
Returns list of fact_ids that are contradicted.
"""
if not existing_facts:
return []
from langchain_core.messages import SystemMessage, HumanMessage
import json
facts_text = "\n".join([
f"[{f.get('fact_id', 'unknown')}]: {f.get('content', '')}"
for f in existing_facts
])
response = detector_model.invoke([
SystemMessage(content="""
You are checking for contradictions between a new fact and existing facts.
A contradiction means the new fact and an existing fact cannot both be true.
Return JSON:
{"contradicted_ids": ["fact_id_1", ...]}
Return empty list if no contradictions found.
"""),
HumanMessage(content=f"""
New fact: {new_fact_content}
Existing facts:
{facts_text}
""")
])
try:
result = json.loads(response.content)
return result.get("contradicted_ids", [])
except json.JSONDecodeError:
return []
Что создаёт экосистема
При работе над этапом «Что такое экосистема» сначала запишите контракт: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Создавайте контрольные точки после дорогостоящих шагов. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий этап.
Что это означает для вашей современной разработки
При работе над этапом «Что это означает», сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Рядом с результатами работы запишите время выполнения и стоимость токенов или запросов. Отображение затрат с самого начала предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Устанавливайте контрольные точки после дорогостоящих шагов. Система возобновления работы не должна снова взимать плату за один и тот же вызов большой языковой модели при повторной попытке обработки последующего элемента. При работе над этапом «Что это означает», сначала запишите условия использования: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Документируйте одновременно успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.
Проверка реальности в производственных условиях
Этап проверки реальности производства работает наилучшим образом, когда рассматривается как измеримая основа. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы вместо обширных скриптов. Когда какой-то шаг сбивается, причина сбоя должна указывать на конкретный ответственный элемент, а не на запутанную цепочку операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
Архитектура для справки
Этап разработки эталонной архитектуры работает наилучшим образом, когда его рассматривают как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия создаваемым элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения задач. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.
Current State (Article 7) Future Agent-Native System
----------------------- --------------------------
Run starts Run starts
| |
Retrieve episodes <-- embedding Query temporal memory
Retrieve KB facts <-- vector Traverse association graph
Retrieve procedures <-- exact key Agent-selected retrieval
| |
Agent runs Agent runs
| |
End-of-run extractor Inline annotation (agent)
stores artifacts + end-of-run consolidation
| |
DynamoDB episodes Temporal-aware store
KB facts (S3/vector) Association index
DynamoDB procedures Policy-gated sharing
Contradiction detection
Intelligent forgetting
Чек-лист операционной работы
На этапе составления чек-листа операционной работы определите входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние системы.
Храните конфигурацию вне кода приложения. Файлы среды, хранилища конфиденциальных данных и флаги функций должны находиться в одном месте, которое операторы могут проверять, не читая весь граф структур.
Необходимо получать одобрение человека для операций, связанных с тратой денег или изменением производственных данных. Настройка во время компиляции не гарантирует полноты функционала продукта.
Напишите краткий руководство: как обновлять ключи, как опустошать очередь, как откатывать последнюю операцию ввода данных.
Документируйте как успешный, так и аварийный сценарии работы. Повторные попытки, проверки человеком и обработка некорректных сообщений являются частью продукта, а не дополнительными улучшениями.
Необходимо получать одобрение человека для операций, связанных с тратой денег или изменением производственных данных. Настройка во время компиляции не гарантирует полноты функционала продукта.
Перед внедрением всей стек-технологии заморозьте версии, сохраните эталонный протокол действий для критически важных сценариев и убедитесь, что известны шаги отката. В совместных средах необходимы ограничения на частоту запросов, проверки принадлежности ресурсов и четко определенный ответственный за обновление секретов. Лучше предпочесть простую надежность сложным одноразовым демонстрациям.
Примечание к пакету обновлений для 8969a47a89be: не включать ключи поставщиков в репозиторий, установить лимит токенов на одну сессию и хранить транскрипции рядом с фиксами для оценки, чтобы последующие замены моделей оставались сопоставимыми.
Для этапа 0 примечания по усилению безопасности необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо записывать время выполнения, стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе с демо-среды в общедоступные среды.
Подробности усиления безопасности 0/837: измерять время выполнения, класс ошибок и расход токенов для этого примечания, а затем решать, следует ли сохранять изменения на основе фиксированного набора вопросов, а не на основе устных замечаний.
При работе над первым этапом записки по усилению безопасности сначала запишите условия работы: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода.
Документируйте одновременно «идеальный» сценарий работы и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.
Подробности усиления безопасности 1/837: измерьте время выполнения, класс ошибки и количество использованных токенов для данной записки, затем решите, следует ли сохранять изменение, опираясь на заранее определенный набор критериев, а не на устные оценки.