Copilotos de ingeniería agénticos con estado usando Letta: memoria central, estado compartido, recuperación semántica, herramientas…

En este tutorial, construimos un copiloto de ingeniería con estado usando Letta que demuestra cómo la memoria agéntica va más allá de las conversaciones convencionales basadas en prompts. Comenzamos creando bloques de memoria central inspeccionables que el agente puede actualizar de forma autónoma, y luego exponemos esos bloques para sincronización de estado programática y acceso compartido entre múltiples agentes. Extendemos aún más la arquitectura con memoria vectorial de archivo para recuperación semántica, una herramienta personalizada del lado del servidor para estimación determinista de riesgos, streaming para observar la ejecución del agente, historial de mensajes del lado del servidor para conversaciones persistentes y exportación .af para portabilidad. Juntos, estos componentes demuestran una arquitectura de memoria práctica en la que el estado persistente, la recuperación, las herramientas, la colaboración multiagente y los rastros de ejecución funcionan como un sistema agéntico integrado.

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")

Configuramos el cliente de Letta, instalamos el SDK y añadimos funciones auxiliares de compatibilidad que manejan variaciones de la API entre versiones de Letta. Seleccionamos dinámicamente un modelo disponible y creamos un copiloto de ingeniería con estado con bloques de memoria central estructurados. Luego inspeccionamos el estado inicial de la memoria para establecer el contexto persistente del agente.

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")])

Demostramos la autoedición de la memoria proporcionando al copiloto información durable sobre el ingeniero, el proyecto, las decisiones y los modos de fallo conocidos. También actualizamos el estado del proyecto directamente desde el código de la aplicación y creamos bloques compartidos de solo lectura y de escritura para la colaboración multiagente. Luego conectamos un agente revisor a la memoria compartida y observamos cómo ambos agentes intercambian contexto de trabajo persistente.

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.")

Poblamos la memoria de archivo con conocimiento de ingeniería etiquetado y usamos búsqueda semántica para recuperar información relevante sin depender de coincidencias exactas de palabras clave. Luego permitimos que el agente consulte este conocimiento de archivo de forma autónoma al responder una pregunta operativa. Finalmente, creamos y adjuntamos una herramienta personalizada del lado del servidor para el riesgo de lanzamiento, de modo que el agente pueda realizar cálculos deterministas y registrar la decisión resultante en la memoria.

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)

Usamos la interfaz de streaming de Letta para observar el razonamiento, las llamadas a herramientas y las respuestas del asistente mientras el agente ejecuta una tarea. Luego inspeccionamos el historial de conversación del lado del servidor para verificar que el estado de la interacción persiste independientemente del proceso local de Python. También demostramos cómo el agente puede reconectarse desde un proceso nuevo usando únicamente su identificador persistente.

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")

Exportamos el estado completo del agente a un archivo .af, proporcionando una representación portátil de la memoria, las herramientas y el estado de conversación del agente. Luego incluimos lógica opcional de limpieza que elimina los agentes y bloques compartidos creados durante el tutorial. Terminamos mostrando los identificadores persistentes de los agentes y la ubicación del Letta ADE para una posterior inspección.

En conclusión, ahora tenemos una visión completa de cómo Letta admite flujos de trabajo de agentes persistentes e inspeccionables en lugar de tratar cada interacción como un prompt aislado. Vimos cómo la memoria central mantiene en contexto el estado de alto valor, la memoria de archivo proporciona una recuperación escalable, los bloques compartidos y de solo lectura coordinan múltiples agentes, y el código de la aplicación puede sincronizar directamente el estado externo. También usamos herramientas personalizadas y streaming para extender y observar el comportamiento del agente, mientras que el historial del lado del servidor y la exportación del agente proporcionan persistencia y portabilidad. A través de este flujo de trabajo, establecimos una base práctica para construir copilotos de ingeniería que retienen continuamente el contexto, razonan sobre el conocimiento acumulado, colaboran entre agentes y operan con memoria explícita y controlable.

Consulta los CÓDIGOS COMPLETOS aquí. Todo el mérito es del investigador de este proyecto. También, no dudes en seguirnos en Twitter y no olvides unirte a nuestro SubReddit de ML con más de 150k y suscribirte a nuestro boletín. ¡Espera! ¿estás en telegram? ahora también puedes unirte a nosotros en telegram.

¿Necesitas asociarte con nosotros para promocionar tu Repositorio de GitHub O Página de Hugging Face O Lanzamiento de Producto O Webinar, etc.? Conéctate con nosotros

La entrada Copilotos de ingeniería agentica con estado con Letta: memoria central, estado compartido, recuperación semántica, herramientas personalizadas, streaming y flujos de trabajo portátiles de agentes apareció primero en MarkTechPost.