Практичні нотатки: Агентні архітектури — Стаття 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 Temporal Awareness працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Зберігайте стан графа у випрямленому та типованому форматі. Вкладені блоки приховують інформацію про те, який вузол заповнив яке поле, та ускладнюють відновлення після перерв. Етап Gap 1 Temporal Awareness працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу. Документуйте як успішний, так і відновлювальний шляхи роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
+----------------------------------+------------------------------------------+
| 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: Забування як процес першого класу
Під час роботи над етапом «Забування» записуйте спочатку контракт: необхідні вхідні дані, сигнал успіху та те, що відбувається при частковій невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність операцій. Робіть перевірки після дорогих кроків. Функція відновлення не повинна знову оплачувати один і той самий виклик LLM, коли оператор намагається знову виконати пізнішу операцію.
# 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 []
Що створює екосистема
Під час роботи над етапом «Що таке екосистема» спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Дайте назви елементам, визначте критерії успіху та не допускайте безповідомного часткового виконання. Робіть перевірки після дорогих кроків. Система відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор намагається знову виконати пізнішу операцію.
Що це означає для вашого сучасного розробництва
Під час роботи над етапом «Що це означає», спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із результатами функціонування. Відображення витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-версії до спільних середовищ. Робіть контрольні пункти після дорогих кроків. Система повторного запуску не повинна знову стягувати плату за один і той самий виклик ШІ, коли оператор перепробовує пізніший етап. Під час роботи над етапом «Що це означає», спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте оптимальний та альтернативний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Перевірка реальності у продакшені
Етап перевірки реальності виробництва працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій. Зберігайте стан графа у простому та типованому вигляді. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, і ускладнюють продовження роботи після перерв.
Архітектура для посилання
Етап проєктування архітектури працює найкраще, коли його розглядають як вимірювану поверхню. Запишіть один ідеальний зразок роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Зберігайте структуру графа простою та типованою. Вкладені елементи приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.
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: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього заходу, а потім вирішіть, чи зберегти зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.