Практычныя нататкі: агентныя архітэктуры, артыкул 12 — ці гатовыя мы?
Практычны разбор тэмы «агентныя архітэктуры і ці гатовыя мы да наступнага кроку»: кантракты, праверкі і гатовыя фрагменты кода для каманд, якія ўкараняюць гэты падыход.
Крыніцы: Agentic Architectures — Article 12: Are We Ready For An Agent-Native Memory System?
Наступныя прыміткі паказваюць практычны падход да разумэння тэмы «Агентныя архітэктуры — Артыкул 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: Associative Retrieval
Для этапы асоцыяўнага выкарыстання даных 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 Cross-Agent Memory неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і крэтыры завершэння пры перадзеі коду. Аперацыйныя працавнікі должны магчымае запускаць крок з вядомай точкі контролю без неабяжнага вычыслення захаваных станоў. Запісваць час выканання і кост токеноў або запытаў разам з функцыйнаімі рэзультатамі. Відразлівае паказанне костаў запобегае неспадзяваным рахункам, калі працэс пераходзіць з дэмовай асяроддзя ў спадзеленую. Пры выкананні дзеянь, якія коштаюць грошы або зменяюць даны ў працэсе, неабяжна ўключыць людзкую апраўдку. Конфігурацыя ў часе компілявання не ёсць падставай для стверджэння, што продукт ўсё завершаны. Для этапы Gap 4 Cross-Agent Memory неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і крэтыры завершэння пры перадзеі коду. Аперацыйныя працавнікі должны магчымае запускаць крок з вядомай точкі контролю без неабяжнага вычыслення захаваных станоў. Адначасова задокументаваць шлях успеху і шлях вярнення ў нормальны стан. Перапрыбуткі, людзкія контрольныя пункты і обработка некорэктных запытаў ёсць часткай продукту, а не елементамі, якія дадаюцца пазней.
# 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: Забыванне як прынцыпавая процедура
Калі працюеце над этапам Забыванне з Этапу 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, калі аператар праканае пазнейшы вузел.
Што гэта значыць для вашай сучаснай працы
Калі працуеце над этапам «Што гэта значыць», спачатку запісайце умовы кантракту: неабяжлівыя даныя, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі спіс дапамагае заліцвачыць пазнейшыя змены ў кодзе. Запісвайце час выканання і кост токена або запыту праза функцыйнальныя рэзултаты. Відразлівасць коста з самага пачатку запобегае неспакоўным рахункам, калі працэс пераходзіць з дэмаверсіі ў спяльныя асяроддзя. Зробіце контрольную пазнаку пасля дорогіх крокаў. Сістэма вярнення не павінна знову ставіць плату за той самы вызов 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: не трэба кантрацеўваць ключы прадаўцаў у рэпазітарыі, задаць максімальную кантэйнернасць токена на кожную сесію, а таксама зберагчы транскрыпціі праза фіксатуры адлічэння, каб пазнейшыя замены мадэляў заставаліся порównанымі.
Для запіскі параграфу 0 пра зміцнэнне: перад зменай коду неабходна адзначыць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння. Аперацыёныя працавнікі должны магчымае перадзваначыць крок з вядомага пункта контролю без неабясненняя схованага стану. Неабходна фіксавацыя часу выкарыстоўвання, а таксама косту токена чы супылкі праза функцыйнае рэзультаты. Відкрытая інформацыя пра косты запобегае неспадзеваным рачункам, калі працэс пераходзіць з дэмавайнага сераўера ў спадзеленыя сераўеры.
Дакладнасць параграфу 0/837 пра зміцнэнне: неабходна замерваць час выкарыстоўвання, класы каштоўкаў і кількасць токенаў для гэтай запіскі, а праза тое вырашваць, чы трэба застаўляць змяну на адной пазначанай сэткі пытанняў, а не на адной лячбе.
Калі працюеце над першым этапам зміцнення, спачатку запісайце умовы кантракта: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі спіс контроля дапамагае заліцвачыць пазнейшыя змены коду.
Документавайце як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцця і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не пазнейшым дапрацоўкам.
Дзеянні зміцнення 1/837: вымерайце час выканання, класію памылак і колькасць выкорыстоўваных токенав для гэтага пункту, а потым выберайце, чы робіць змену на аднойчынных крытэрыях, а не на аснове індывідуальных спостарожэнняў.