이 튜토리얼에서는 Letta 를 사용하여 상태 기반 엔지니어링 코파일럿을 구축하고, 에이전트 메모리가 기존의 프롬프트 기반 대화를 넘어 어떻게 확장되는지 보여줍니다. 먼저 에이전트가 자율적으로 업데이트할 수 있는 검사 가능한 코어 메모리 블록을 만들고, 그 다음 프로그래밍 방식의 상태 동기화와 여러 에이전트 간의 공유 접근을 위해 해당 블록을 노출합니다. 또한 시맨틱 검색을 위한 아카이벌 벡터 메모리, 결정론적 위험 추정을 위한 커스텀 서버 측 도구, 에이전트 실행을 관찰하기 위한 스트리밍, 영구 대화를 위한 서버 측 메시지 기록, 그리고 이식성을 위한 .af 내보내기로 아키텍처를 확장합니다. 이러한 구성 요소들은 영구 상태, 검색, 도구, 멀티 에이전트 협업, 그리고 실행 추적이 하나의 통합된 에이전트 시스템으로 작동하는 실용적인 메모리 아키텍처를 보여줍니다.
코드 복사import subprocess, sys
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", "letta-client"], check=True)
import os, json, time, textwrap
from getpass import getpass
from letta_client import Letta
def flex(*attempts, label=""):
err = None
for fn in attempts:
try:
return fn()
except Exception as e:
err = e
raise RuntimeError(f"All call variants failed for {label or 'call'}: {err}") from err
API_KEY = os.environ.get("LETTA_API_KEY") or getpass("Letta API key (platform.letta.com/api-keys): ")
BASE_URL = os.environ.get("LETTA_BASE_URL", "").strip()
client = flex(
lambda: Letta(base_url=BASE_URL, api_key=API_KEY) if BASE_URL else Letta(api_key=API_KEY),
lambda: Letta(base_url=BASE_URL, token=API_KEY) if BASE_URL else Letta(token=API_KEY),
label="client init",
)
def hr(title):
print("\n" + "=" * 92 + f"\n {title}\n" + "=" * 92)
def as_text(content):
if isinstance(content, str):
return content
if isinstance(content, list):
return "".join(getattr(p, "text", "") or "" for p in content)
return str(content or "")
def show(response, hide_reasoning=False):
for m in getattr(response, "messages", response) or []:
kind = getattr(m, "message_type", None) or getattr(m, "type", "?")
if kind == "reasoning_message" and not hide_reasoning:
print(" reasoning :", textwrap.shorten(getattr(m, "reasoning", "") or "", 220))
elif kind == "tool_call_message":
tc = getattr(m, "tool_call", None)
print(" tool call :", getattr(tc, "name", "?"), textwrap.shorten(str(getattr(tc, "arguments", "")), 200))
elif kind == "tool_return_message":
print(" tool ret :", textwrap.shorten(str(getattr(m, "tool_return", "")), 260))
elif kind == "assistant_message":
print(" assistant :", as_text(getattr(m, "content", "")))
elif kind == "approval_request_message":
print(" approval : agent is waiting for you to approve a tool call")
u = getattr(response, "usage", None)
if u:
print(f" ↳ steps={getattr(u, 'step_count', '?')} tokens={getattr(u, 'total_tokens', '?')}")
def chat(agent_id, text, max_steps=12, quiet=False):
r = flex(
lambda: client.agents.messages.create(agent_id=agent_id, input=text, max_steps=max_steps),
lambda: client.agents.messages.create(agent_id=agent_id, input=text),
lambda: client.agents.messages.create(agent_id=agent_id,
messages=[{"role": "user", "content": text}]),
label="messages.create",
)
if not quiet:
print(f"\n user : {text}")
show(r)
return r
PREFERRED = ["openai/gpt-4.1", "openai/gpt-4o-mini", "anthropic/claude-sonnet-4-20250514"]
try:
handles = [getattr(m, "handle", None) for m in client.models.list()]
MODEL = next((h for h in PREFERRED if h in handles), None) or handles[0]
except Exception:
MODEL = PREFERRED[0]
print(f" connected · model handle = {MODEL}")
hr("1. Create a stateful agent with custom memory blocks")
def make_agent(**kw):
return flex(
lambda: client.agents.create(model=MODEL, embedding="openai/text-embedding-3-small", **kw),
lambda: client.agents.create(model=MODEL, **kw),
label="agents.create",
)
copilot = make_agent(
name=f"copilot-{int(time.time())}",
description="Engineering copilot with inspectable long-term memory",
tags=["tutorial", "copilot"],
memory_blocks=[
{"label": "persona",
"value": "I am Mnemo, an engineering copilot. I am terse, concrete, and I never "
"invent facts — if I don't know something I say so and then go look it up.",
"limit": 3000},
{"label": "human",
"description": "Durable facts about the engineer I work with: name, role, stack, "
"preferences, working style. Update this whenever I learn something stable.",
"value": "Unknown so far.",
"limit": 3000},
{"label": "project_state",
"description": "Live state of the project we are shipping: milestones, blockers, "
"decisions and their rationale. Keep it dense and current; delete stale lines.",
"value": "(empty)",
"limit": 5000},
{"label": "lessons",
"description": "Hard-won lessons and mistakes to never repeat. Append-only, one line each.",
"value": "(empty)",
"limit": 2000},
],
)
print("agent id:", copilot.id)
def get_block(agent_id, label):
return flex(
lambda: client.agents.blocks.retrieve(agent_id=agent_id, block_label=label),
lambda: client.agents.blocks.retrieve(agent_id, label),
label="agents.blocks.retrieve",
)
def set_block(agent_id, label, value):
return flex(
lambda: client.agents.blocks.update(agent_id=agent_id, block_label=label, value=value),
lambda: client.agents.blocks.modify(agent_id=agent_id, block_label=label, value=value),
lambda: client.agents.blocks.update(agent_id, label, value=value),
label="agents.blocks.update",
)
def dump_memory(agent_id, title="core memory"):
blocks = flex(lambda: client.agents.blocks.list(agent_id=agent_id),
lambda: client.agents.blocks.list(agent_id), label="agents.blocks.list")
print(f"\n──── {title} ────")
for b in blocks:
ro = " [read-only]" if getattr(b, "read_only", False) else ""
print(f"▸ {b.label}{ro} ({len(b.value or '')}/{getattr(b, 'limit', '?')} chars)")
print(textwrap.indent(textwrap.fill(b.value or "", 100), " "))
return blocks
dump_memory(copilot.id, "core memory at birth")
Letta 클라이언트를 설정하고, SDK를 설치하며, Letta 버전 간의 API 차이를 처리하는 호환성 헬퍼를 추가합니다. 사용 가능한 모델을 동적으로 선택하고 구조화된 코어 메모리 블록을 갖춘 상태 기반 엔지니어링 코파일럿을 생성합니다. 그런 다음 초기 메모리 상태를 검사하여 에이전트의 영구 컨텍스트를 확립합니다.
코드 복사hr("2. Self-editing memory — talk once, then read the block back")
chat(copilot.id,
"Hi, I'm Priya, staff engineer at Northwind. We're shipping 'Halo', a Rust ingestion "
"service, on Nov 14. I hate long explanations — give me bullets or code, never essays. "
"Current blocker: the Kafka consumer rebalances every ~4 minutes under load.")
chat(copilot.id,
"Update: we traced the rebalance to a 5s max.poll.interval.ms with slow batch commits. "
"Decision: move commits off the poll loop. Also remember we already tried bumping "
"session.timeout.ms and it did nothing — don't suggest that again.")
dump_memory(copilot.id, "core memory after 2 turns (note what moved where)")
hr("3. Your code writes into the context window")
live = get_block(copilot.id, "project_state")
set_block(copilot.id, "project_state",
(live.value or "") + "\n[CI 09:41Z] main is RED: integration/kafka_rebalance_test failing (flaky x3).")
chat(copilot.id, "Anything I should know before I start the day?")
hr("4. Shared memory between two agents")
policies = client.blocks.create(
label="policies",
description="Non-negotiable engineering policies. Read-only: obey, never edit.",
value="1. No direct pushes to main.\n2. Every fix ships with a regression test.\n"
"3. Customer data never leaves EU regions.",
limit=2000,
read_only=True,
)
scratchpad = client.blocks.create(
label="team_scratchpad",
description="Shared working notes between the copilot and the reviewer agent. "
"Both agents read and write here; keep entries prefixed with your name.",
value="(empty)",
limit=4000,
)
for bid in (policies.id, scratchpad.id):
flex(lambda: client.agents.blocks.attach(agent_id=copilot.id, block_id=bid),
lambda: client.agents.blocks.attach(copilot.id, bid),
label="agents.blocks.attach")
reviewer = make_agent(
name=f"reviewer-{int(time.time())}",
description="Strict code reviewer sharing memory with the copilot",
tags=["tutorial", "reviewer"],
block_ids=[policies.id, scratchpad.id],
memory_blocks=[{"label": "persona",
"value": "I am Argus, a paranoid staff reviewer. I look for the failure mode "
"everyone else skipped, and I cite the team policies by number."}],
)
print("reviewer id:", reviewer.id)
chat(copilot.id, "Write your proposed fix for the rebalance issue into the team_scratchpad "
"block so the reviewer can see it. Keep it to 4 lines.")
print("\n what the REVIEWER sees in the shared block (written by the other agent):")
print(textwrap.indent(client.blocks.retrieve(scratchpad.id).value, " "))
chat(reviewer.id, "Read team_scratchpad and review the proposed fix. Cite any policy it violates, "
"then append your verdict to the same block.")
print("\n shared block after the reviewer replied:")
print(textwrap.indent(client.blocks.retrieve(scratchpad.id).value, " "))
print("\nagents attached to this block:",
[a.name for a in flex(lambda: client.blocks.agents.list(block_id=scratchpad.id),
lambda: client.blocks.agents.list(scratchpad.id),
label="blocks.agents.list")])
엔지니어, 프로젝트, 결정 사항, 알려진 장애 양상에 대한 영구 정보를 코파일럿에 제공하여 셀프 편집 메모리를 시연합니다. 또한 애플리케이션 코드에서 직접 프로젝트 상태를 업데이트하고, 멀티 에이전트 협업을 위한 공유 읽기 전용 및 쓰기 가능 블록을 생성합니다. 그런 다음 리뷰어 에이전트를 공유 메모리에 연결하고 두 에이전트가 영구 작업 컨텍스트를 어떻게 교환하는지 관찰합니다.
코드 복사hr("5. Archival memory: bulk-load knowledge, then let the agent retrieve it")
KNOWLEDGE = [
("Postmortem 2024-08-02: Halo dropped 1.2M events when the consumer group rebalanced during a "
"deploy. Root cause: commits inside the poll loop blocked past max.poll.interval.ms.",
["postmortem", "kafka", "halo"]),
("ADR-017: Halo uses at-least-once delivery with idempotent writes keyed on (tenant_id, event_id). "
"Exactly-once via Kafka transactions was rejected for throughput reasons.",
["adr", "architecture", "halo"]),
("Runbook: to drain a Halo node, set DRAIN=1, wait for lag<1000, then SIGTERM. Never SIGKILL — "
"in-flight batches are only checkpointed on graceful shutdown.",
["runbook", "ops", "halo"]),
("Benchmark 2025-01: rdkafka 2.3 with 8 partitions sustained 240k events/s at p99 32ms on "
"c6i.4xlarge; the bottleneck was JSON parsing, not the network.",
["benchmark", "performance"]),
("Priya's preference on file layout: one module per bounded context, integration tests live "
"next to the module, no tests/ mega-directory.", ["preference", "style"]),
("Incident 2025-03-19: an EU tenant's payload was mirrored to us-east-1 for 40 minutes. "
"Fixed by region-pinning the DLQ. This is why policy 3 exists.", ["incident", "compliance"]),
]
def archive(agent_id, content, tags):
return flex(
lambda: client.agents.passages.insert(agent_id=agent_id, content=content, tags=tags),
lambda: client.agents.passages.create(agent_id=agent_id, text=content, tags=tags),
lambda: client.agents.passages.create(agent_id=agent_id, text=content),
label="agents.passages.insert",
)
for content, tags in KNOWLEDGE:
archive(copilot.id, content, tags)
print(f"inserted {len(KNOWLEDGE)} passages")
hits = flex(
lambda: client.agents.passages.search(agent_id=copilot.id, query="losing messages during a deploy"),
lambda: client.agents.passages.list(agent_id=copilot.id, search="losing messages during a deploy"),
label="agents.passages.search",
)
print("\n programmatic semantic search for 'losing messages during a deploy':")
for h in list(hits)[:3]:
print(" •", textwrap.shorten(getattr(h, "content", None) or getattr(h, "text", ""), 150))
chat(copilot.id, "Before I restart a node tonight: is there anything in our history I should "
"worry about, and what's the safe procedure?")
hr("6. Give the agent a custom tool")
def estimate_release_risk(open_bugs: int, days_to_deadline: int, test_coverage: float) -> str:
"""Estimate the risk of shipping a release on time.
Args:
open_bugs (int): Number of open P0/P1 bugs.
days_to_deadline (int): Calendar days left before the ship date.
test_coverage (float): Line coverage as a fraction between 0.0 and 1.0.
Returns:
str: A risk score from 0-100 and a verdict.
"""
import math
pressure = open_bugs / max(days_to_deadline, 1)
score = max(0.0, min(100.0, 100 * (1 - math.exp(-1.4 * pressure)) * (1.3 - test_coverage)))
verdict = "SHIP" if score < 35 else ("SLIP ONE WEEK" if score < 70 else "DO NOT SHIP")
return f"risk_score={score:.1f} verdict={verdict}"
tool = flex(
lambda: client.tools.upsert_from_function(func=estimate_release_risk),
lambda: client.tools.create_from_function(func=estimate_release_risk),
label="tools.upsert_from_function",
)
flex(lambda: client.agents.tools.attach(agent_id=copilot.id, tool_id=tool.id),
lambda: client.agents.tools.attach(copilot.id, tool.id),
label="agents.tools.attach")
print("attached tool:", tool.name, "| agent tools:",
[t.name for t in flex(lambda: client.agents.tools.list(agent_id=copilot.id),
lambda: client.agents.tools.list(copilot.id), label="agents.tools.list")])
chat(copilot.id, "We have 9 open P0/P1 bugs, 12 days to the Nov 14 date, coverage is 0.61. "
"Score the risk with the tool, then record the verdict in project_state.")
태그가 지정된 엔지니어링 지식으로 아카이벌 메모리를 채우고, 시맨틱 검색을 사용하여 정확한 키워드 일치에 의존하지 않고 관련 정보를 검색합니다. 그런 다음 운영 관련 질문에 답할 때 에이전트가 이 아카이벌 지식을 자율적으로 쿼리하도록 허용합니다. 마지막으로 커스텀 서버 측 릴리스 위험 도구를 생성하고 연결하여 에이전트가 결정론적 계산을 수행하고 그 결과로 얻은 결정을 메모리에 기록하도록 합니다.
코드 복사hr("7. Streaming the agent's steps")
def stream_chat(agent_id, text):
print(f" user : {text}")
s = flex(
lambda: client.agents.messages.stream(agent_id=agent_id, input=text),
lambda: client.agents.messages.stream(agent_id=agent_id,
messages=[{"role": "user", "content": text}]),
lambda: client.agents.messages.create_stream(agent_id=agent_id,
messages=[{"role": "user", "content": text}]),
label="messages.stream",
)
for chunk in s:
kind = getattr(chunk, "message_type", None) or getattr(chunk, "type", "")
if kind == "reasoning_message":
print("", textwrap.shorten(getattr(chunk, "reasoning", "") or "", 160))
elif kind == "tool_call_message":
print("", getattr(getattr(chunk, "tool_call", None), "name", "?"))
elif kind == "assistant_message":
print("", as_text(getattr(chunk, "content", "")))
try:
stream_chat(copilot.id, "One-line status of Halo, from memory only. No tools.")
except Exception as e:
print("streaming unavailable on this SDK version:", e)
hr("8. Server-side conversation history")
hist = flex(lambda: client.agents.messages.list(agent_id=copilot.id, limit=8),
lambda: client.agents.messages.list(copilot.id, limit=8),
label="agents.messages.list")
for m in hist:
kind = getattr(m, "message_type", "?")
body = as_text(getattr(m, "content", "")) or getattr(m, "reasoning", "") or ""
print(f" [{kind:<24}] {textwrap.shorten(str(body), 110)}")
print("\n reconnect from a fresh process with just the id:")
print(" client.agents.messages.create(agent_id='%s', input='...')" % copilot.id)
Letta의 스트리밍 인터페이스를 사용하여 에이전트가 작업을 실행할 때 추론, 도구 호출, 어시스턴트 응답을 관찰합니다. 그런 다음 서버 측 대화 기록을 검사하여 상호작용 상태가 로컬 Python 프로세스와 독립적으로 지속되는지 확인합니다. 또한 영구 식별자만을 사용하여 새 프로세스에서 에이전트를 다시 연결하는 방법도 시연합니다.
코드 복사hr("9. Export / import the agent")
try:
dump = flex(lambda: client.agents.export_file(agent_id=copilot.id),
lambda: client.agents.export_file(copilot.id), label="agents.export_file")
blob = dump if isinstance(dump, (str, bytes)) else json.dumps(dump, default=str)
path = "/content/copilot.af" if os.path.isdir("/content") else "copilot.af"
with open(path, "wb" if isinstance(blob, bytes) else "w") as f:
f.write(blob)
print(f"exported → {path} ({len(blob)} bytes) — memory, tools and history travel with it.")
print("re-import later with: client.agents.import_file(file=open(path,'rb'))")
except Exception as e:
print("export not available on this SDK/deployment:", e)
CLEANUP = False
if CLEANUP:
for aid in (copilot.id, reviewer.id):
try: client.agents.delete(aid)
except Exception as e: print("agent delete:", e)
for bid in (policies.id, scratchpad.id):
try: client.blocks.delete(bid)
except Exception as e: print("block delete:", e)
print("deleted.")
else:
hr("Done")
print(f"copilot : {copilot.id}\nreviewer : {reviewer.id}")
print("Inspect them visually in the ADE at https://app.letta.com")
전체 에이전트 상태를 .af 파일로 내보내어 에이전트의 메모리, 도구, 대화 상태를 이식 가능한 형태로 제공합니다. 그런 다음 튜토리얼에서 생성된 에이전트와 공유 블록을 제거하는 선택적 정리 로직을 포함합니다. 마지막으로 영구 에이전트 식별자와 추가 검사를 위한 Letta ADE 위치를 표시하며 마무리합니다.
결론적으로, 이제 우리는 Letta가 모든 상호작용을 고립된 프롬프트로 취급하는 것이 아니라 영구적이고 검사 가능한 에이전트 워크플로우를 어떻게 지원하는지에 대한 전체적인 모습을 갖게 되었습니다. 우리는 코어 메모리가 컨텍스트 내에서 고가치 상태를 유지하고, 아카이벌 메모리가 확장 가능한 검색을 제공하며, 공유 및 읽기 전용 블록이 여러 에이전트를 조율하고, 애플리케이션 코드가 외부 상태를 직접 동기화할 수 있는 방법을 보았습니다. 또한 커스텀 도구와 스트리밍을 사용하여 에이전트 동작을 확장하고 관찰하는 한편, 서버 측 기록과 에이전트 내보내기가 영속성과 이식성을 제공합니다. 이 워크플로우를 통해 우리는 컨텍스트를 지속적으로 유지하고, 축적된 지식을 기반으로 추론하며, 에이전트 간에 협업하고, 명시적이고 제어 가능한 메모리로 작동하는 엔지니어링 코파일럿을 구축하기 위한 실용적인 기반을 마련했습니다.
다음을 확인해 보세요 전체 코드는 여기`. 이 프로젝트의 연구자에게 모든 공이 돌아갑니다. 또한, 언제든지 저희를 팔로우해 주세요:` 트위터 그리고 저희의 다음 항목에 참여하는 것을 잊지 마세요 ML 서브레딧 150k+ 그리고 구독하세요: 우리의 뉴스레터. 잠깐! 텔레그램을 사용하고 계신가요? 이제 텔레그램에서도 저희와 함께하실 수 있습니다.
GitHub Repo OR Hugging Face Page OR Product Release OR Webinar 등을 홍보하기 위해 저희와 파트너십이 필요하신가요? 저희에게 연락하세요
다음 글 Letta를 활용한 상태 기반 에이전틱 엔지니어링 코파일럿: 코어 메모리, 공유 상태, 시맨틱 검색, 사용자 정의 도구, 스트리밍 및 이식 가능한 에이전트 워크플로 는 최초 게시된 글입니다: MarkTechPost.