Практычныя прытамулі: Агентныя архітектуры — Стаття 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: 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: вымерайце час выканання, класію памылак і колькасць выкорыстоўваных токенав для гэтага пункту, а потым выберайце, чы робіць змену на аднойчынных критэрыях, а не на аснове індывідуальных спостарожэнняў.