Copilots d'ingénierie agentiques avec état et Letta : mémoire principale, état partagé, récupération sémantique, outils personnalisés…

Dans ce tutoriel, nous construisons un copilote d'ingénierie avec état à l'aide de Letta qui démontre comment la mémoire agentique s'étend au-delà des conversations conventionnelles basées sur des prompts. Nous commençons par créer des blocs de mémoire principale inspectables que l'agent peut mettre à jour de manière autonome, puis nous exposons ces blocs pour une synchronisation programmatique de l'état et un accès partagé entre plusieurs agents. Nous étendons ensuite l'architecture avec une mémoire d'archive vectorielle pour la récupération sémantique, un outil personnalisé côté serveur pour une estimation déterministe des risques, le streaming pour observer l'exécution de l'agent, l'historique des messages côté serveur pour des conversations persistantes, et l'export .af pour la portabilité. Ensemble, ces composants démontrent une architecture mémoire pratique dans laquelle l'état persistant, la récupération, les outils, la collaboration multi-agents et les traces d'exécution fonctionnent comme un système agentique intégré.

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

Nous configurons le client Letta, installons le SDK et ajoutons des fonctions utilitaires de compatibilité qui gèrent les variations d'API entre les versions de Letta. Nous sélectionnons dynamiquement un modèle disponible et créons un copilote d'ingénierie avec état avec des blocs de mémoire principale structurés. Nous examinons ensuite l'état initial de la mémoire pour établir le contexte persistant de l'agent.

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

Nous démontrons la mémoire auto-éditable en fournissant au copilote des informations durables sur l'ingénieur, le projet, les décisions et les modes de défaillance connus. Nous mettons également à jour l'état du projet directement depuis le code applicatif et créons des blocs partagés en lecture seule et en écriture pour la collaboration multi-agents. Nous connectons ensuite un agent relecteur à la mémoire partagée et observons comment les deux agents échangent un contexte de travail persistant.

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

Nous alimentons la mémoire d'archive avec des connaissances d'ingénierie étiquetées et utilisons la recherche sémantique pour récupérer les informations pertinentes sans dépendre de correspondances exactes de mots-clés. Nous permettons ensuite à l'agent d'interroger ces connaissances d'archive de manière autonome pour répondre à une question opérationnelle. Enfin, nous créons et attachons un outil personnalisé côté serveur d'évaluation des risques de mise en production afin que l'agent puisse effectuer des calculs déterministes et enregistrer la décision résultante en mémoire.

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)

Nous utilisons l'interface de streaming de Letta pour observer le raisonnement, les appels d'outils et les réponses de l'assistant pendant que l'agent exécute une tâche. Nous examinons ensuite l'historique des conversations côté serveur pour vérifier que l'état de l'interaction persiste indépendamment du processus Python local. Nous montrons également comment l'agent peut être reconnecté depuis un processus frais en utilisant uniquement son identifiant persistant.

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

Nous exportons l'état complet de l'agent vers un fichier .af, fournissant une représentation portable de la mémoire de l'agent, de ses outils et de son état de conversation. Nous incluons ensuite une logique de nettoyage facultative qui supprime les agents et les blocs partagés créés pendant le tutoriel. Nous terminons en affichant les identifiants persistants des agents et l'emplacement du Letta ADE pour une inspection plus approfondie.

En conclusion, nous avons maintenant une vue complète de la façon dont Letta prend en charge des workflows d'agents persistants et inspectables plutôt que de traiter chaque interaction comme un prompt isolé. Nous avons vu comment la mémoire principale maintient un état à forte valeur en contexte, la mémoire d'archive fournit une récupération évolutive, les blocs partagés et en lecture seule coordonnent plusieurs agents, et le code applicatif peut synchroniser directement un état externe. Nous avons également utilisé des outils personnalisés et le streaming pour étendre et observer le comportement de l'agent, tandis que l'historique côté serveur et l'export d'agent offrent persistance et portabilité. Grâce à ce workflow, nous avons établi une base pratique pour construire des copilotes d'ingénierie qui conservent continuellement le contexte, raisonnent sur les connaissances accumulées, collaborent entre agents et opèrent avec une mémoire explicite et contrôlable.

Découvrez les CODES COMPLETS ici. Tout le mérite revient au chercheur de ce projet. N'hésitez pas également à nous suivre sur Twitter et n'oubliez pas de rejoindre notre SubReddit ML de plus de 150k et de vous abonner à notre Newsletter. Attention ! êtes-vous sur Telegram ? vous pouvez maintenant nous rejoindre également sur Telegram.

Besoin de vous associer à nous pour promouvoir votre dépôt GitHub OU votre page Hugging Face OU le lancement d'un produit OU un webinar, etc. ? Connectez-vous avec nous

L'article Copilots d'ingénierie agentiques avec état grâce à Letta : mémoire centrale, état partagé, recherche sémantique, outils personnalisés, streaming et workflows d'agents portables est apparu en premier sur MarkTechPost.