Letta 实战教程：构建有状态工程副驾驶

We use cookies on our website to give you the most relevant experience by remembering your preferences and repeat visits. By clicking “Accept”, you consent to the use of ALL the cookies. .

Cookie settings REJECT ACCEPT

Necessary Always Active

Necessary cookies are absolutely essential for the website to function properly. This category only includes cookies that ensures basic functionalities and security features of the website. These cookies do not store any personal information.

Cookie __cf_bm
Duration 1 hour
Description This cookie, set by Cloudflare, is used to support Cloudflare Bot Management.
Cookie _pxvid
Duration 1 year
Description PerimeterX sets this cookie to detect fraud and bot activity.
Cookie _px3
Duration 6 minutes
Description This cookie is set by the Bloomberg to protect the site from BOT attacks.
Cookie CookieLawInfoConsent
Duration 1 year
Description CookieYes sets this cookie to record the default button state of the corresponding category and the status of CCPA. It works only in coordination with the primary cookie.
Cookie cookielawinfo-checkbox-necessary
Duration 11 months
Description This cookie is set by GDPR Cookie Consent plugin. The cookies is used to store the user consent for the cookies in the category "Necessary".
Cookie cookielawinfo-checkbox-others
Duration 1 year
Description Set by the GDPR Cookie Consent plugin, this cookie stores user consent for cookies in the category "Others".
Cookie cookielawinfo-checkbox-non-necessary
Duration 11 months
Description This cookie is set by GDPR Cookie Consent plugin. The cookies is used to store the user consent for the cookies in the category "Non Necessary".
Cookie cookielawinfo-checkbox-analytics
Duration 1 year
Description Set by the GDPR Cookie Consent plugin, this cookie records the user consent for the cookies in the "Analytics" category.
Cookie cookielawinfo-checkbox-performance
Duration 1 year
Description Set by the GDPR Cookie Consent plugin, this cookie stores the user consent for cookies in the category "Performance".
Cookie cookielawinfo-checkbox-uncategorized
Duration 1 year
Description The cookie is set by the GDPR Cookie Consent plugin to record the user consent for cookies in the category "Uncategorized".
Cookie cookielawinfo-checkbox-functional
Duration 1 year
Description The GDPR Cookie Consent plugin sets the cookie to record the user consent for the cookies in the category "Functional".
Cookie cookielawinfo-checkbox-advertisement
Duration 1 year
Description Set by the GDPR Cookie Consent plugin, this cookie records the user consent for the cookies in the "Advertisement" category.
Cookie wpEmojiSettingsSupports
Duration session
Description WordPress sets this cookie when a user interacts with emojis on a WordPress site. It helps determine if the user's browser can display emojis properly.
Cookie VISITOR_PRIVACY_METADATA
Duration 6 months
Description YouTube sets this cookie to store the user's cookie consent state for the current domain.
Cookie viewed_cookie_policy
Duration 11 months
Description The cookie is set by the GDPR Cookie Consent plugin and is used to store whether or not user has consented to the use of cookies. It does not store any personal data.
Cookie PHPSESSID
Duration
Description This cookie is native to PHP applications. The cookie is used to store and identify a users' unique session ID for the purpose of managing user session on the website. The cookie is a session cookies and is deleted when all the browser windows are closed.
Cookie __cfduid
Duration 4 weeks
Description The cookie is set by CloudFare. The cookie is used to identify individual clients behind a shared IP address d apply security settings on a per-client basis. It doesnot correspond to any user ID in the web application and does not store any personally identifiable information.

Functional

Functional cookies help to perform certain functionalities like sharing the content of the website on social media platforms, collect feedbacks, and other third-party features.

Cookie yt-remote-connected-devices
Duration never
Description YouTube sets this cookie to store the user's video preferences using embedded YouTube videos.
Cookie ytidb::LAST_RESULT_ENTRY_KEY
Duration never
Description The cookie ytidb::LAST_RESULT_ENTRY_KEY is used by YouTube to store the last search result entry that was clicked by the user. This information is used to improve the user experience by providing more relevant search results in the future.
Cookie yt-remote-device-id
Duration never
Description YouTube sets this cookie to store the user's video preferences using embedded YouTube videos.
Cookie yt-remote-session-name
Duration session
Description The yt-remote-session-name cookie is used by YouTube to store the user's video player preferences using embedded YouTube video.
Cookie yt-remote-fast-check-period
Duration session
Description The yt-remote-fast-check-period cookie is used by YouTube to store the user's video player preferences for embedded YouTube videos.
Cookie yt-remote-session-app
Duration session
Description The yt-remote-session-app cookie is used by YouTube to store user preferences and information about the interface of the embedded YouTube video player.
Cookie yt-remote-cast-available
Duration session
Description The yt-remote-cast-available cookie is used to store the user's preferences regarding whether casting is available on their YouTube video player.
Cookie yt-remote-cast-installed
Duration session
Description The yt-remote-cast-installed cookie is used to store the user's video player preferences using embedded YouTube video.
Cookie na_id
Duration 1 year
Description This cookie is set by Addthis.com to enable sharing of links on social media platforms like Facebook and Twitter
Cookie vc
Duration 1 year
Description This cookie is set by addthis.com on sites that allow sharing on social media.
Cookie __atuvc
Duration 1 year
Description This cookie is set by Addthis to make sure you see the updated count if you share a page and return to it before our share count cache is updated.
Cookie __atuvs
Duration 30 minutes
Description This cookie is set by Addthis to make sure you see the updated count if you share a page and return to it before our share count cache is updated.
Cookie ouid
Duration 1 year
Description The cookie is set by Addthis which enables the content of the website to be shared across different networking and social sharing websites.

Analytics

Analytical cookies are used to understand how visitors interact with the website. These cookies help provide information on metrics the number of visitors, bounce rate, traffic source, etc.

Cookie ga*
Duration 1 year 1 month 4 days
Description Google Analytics sets this cookie to store and count page views.
Cookie _ga
Duration 2 years
Description This cookie is installed by Google Analytics. The cookie is used to calculate visitor, session, camapign data and keep track of site usage for the site's analytics report. The cookies store information anonymously and assigns a randoly generated number to identify unique visitors.
Cookie sbjs_migrations
Duration session
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie sbjs_current_add
Duration session
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie sbjs_first_add
Duration session
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie sbjs_current
Duration session
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie sbjs_first
Duration session
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie sbjs_udata
Duration session
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie sbjs_session
Duration 1 hour
Description Sourcebuster sets this cookie to identify the source of a visit and stores user action information in cookies. This analytical and behavioural cookie is used to enhance the visitor experience on the website.
Cookie tk_or
Duration 1 year 1 month 4 days
Description JetPack plugin sets this referral cookie on sites using WooCommerce, which analyzes referrer behaviour for Jetpack.
Cookie tk_r3d
Duration 3 days
Description JetPack installs this cookie to collect internal metrics for user activity and improve user experience.
Cookie tk_lr
Duration 1 year
Description JetPack plugin sets this referral cookie on sites using WooCommerce, which analyzes referrer behaviour for Jetpack.
Cookie tk_ai
Duration 1 year
Description JetPack sets this cookie to store a randomly-generated anonymous ID used only within the admin area and for general analytics tracking.
Cookie tk_tc
Duration session
Description JetPack sets this cookie to record details on how users use the website.
Cookie _gat_gtag_UA_5784146_31
Duration 1 minute
Description Google Used to distinguish users.
Cookie GPS
Duration 30 minutes
Description This cookie is set by Youtube and registers a unique ID for tracking users based on their geographical location
Cookie __gads
Duration 2 years
Description This cookie is set by Google and stored under the name dounleclick.com. This cookie is used to track how many times users see a particular advert which helps in measuring the success of the campaign and calculate the revenue generated by the campaign. These cookies can only be read from the domain that it is set on so it will not track any data while browsing through another sites.
Cookie uvc
Duration 1 year
Description The cookie is set by addthis.com to determine the usage of Addthis.com service.
Cookie ad-id
Duration 7 months
Description Provided by amazon-adsystem.com for tracking user actions on other websites to provide targeted content
Cookie _gat_gtag_UA_116563943_1
Duration 1 minute
Description Google uses this cookie to distinguish users.
Cookie _gid
Duration 1 day
Description This cookie is installed by Google Analytics. The cookie is used to store information of how visitors use a website and helps in creating an analytics report of how the wbsite is doing. The data collected including the number visitors, the source where they have come from, and the pages viisted in an anonymous form.

Performance

Performance cookies are used to understand and analyze the key performance indexes of the website which helps in delivering a better user experience for the visitors.

Cookie YSC
Duration
Description This cookies is set by Youtube and is used to track the views of embedded videos.
Cookie _gat
Duration 1 minute
Description This cookies is installed by Google Universal Analytics to throttle the request rate to limit the colllection of data on high traffic sites.

Advertisement cookies are used to provide visitors with relevant ads and marketing campaigns. These cookies track visitors across websites and collect information to provide customized ads.

Cookie COMPASS
Duration 1 hour
Description The COMPASS cookie is used by Yahoo to deliver targeted advertising based on user's online behavior.
Cookie NID
Duration 5 months
Description This cookie is used to a profile based on user's interest and display personalized ads to the users.
Cookie __Secure-YNID
Duration 6 months
Description Google cookie used to protect user security and prevent fraud, especially during the login process.
Cookie __Secure-ROLLOUT_TOKEN
Duration 6 months
Description YouTube sets this cookie to manage feature rollout and experimentation. It helps Google control which new features or interface changes are shown to users as part of testing and staged rollouts, ensuring consistent experience for a given user during an experiment.
Cookie yt.innertube::nextId
Duration never
Description YouTube sets this cookie to register a unique ID to store data on what videos from YouTube the user has seen.
Cookie yt.innertube::requests
Duration never
Description YouTube sets this cookie to register a unique ID to store data on what videos from YouTube the user has seen.
Cookie VISITOR_INFO1_LIVE
Duration 5 months
Description This cookie is set by Youtube. Used to track the information of the embedded YouTube videos on a website.
Cookie TapAd_TS
Duration 1 month
Description The cookie is set by Tapad.com. The purpose of the cookie is to track users across devices to enable targeted advertising.
Cookie TapAd_DID
Duration 1 month
Description The cookie is set by tapad.com. The purpose of the cookie is to track users across devices to enable targeted advertising
Cookie personalization_id
Duration 2 years
Description This cookie is set by twitter.com. It is used integrate the sharing features of this social media. It also stores information about how the user uses the website for tracking and targeting.
Cookie uid
Duration 1 year
Description This cookie is used to measure the number and behavior of the visitors to the website anonymously. The data includes the number of visits, average duration of the visit on the website, pages visited, etc. for the purpose of better understanding user preferences for targeted advertisments.
Cookie loc
Duration 1 year
Description This cookie is set by Addthis. This is a geolocation cookie to understand where the users sharing the information are located.
Cookie IDE
Duration 2 years
Description Used by Google DoubleClick and stores information about how the user uses the website and any other advertisement before visiting the website. This is used to present users with ads that are relevant to them according to the user profile.
Cookie di2
Duration 1 year
Description This cookie is set by addthis.com on sites that allows sharing on social media. The cookie is used to track user behavior anonymously to generate usage trends to improve relevance to their services and advertising.

Others

Other uncategorized cookies are those that are being analyzed and have not been classified into a category as yet.

Cookie pxcts
Duration session
Description Description is currently not available.
Cookie _pxttld
Duration session
Description Description is currently not available.
Cookie SGPBShowingLimitationDomain77659
Duration 2 days
Description Description is currently not available.
Cookie __Secure-YEC
Duration past
Description YouTube sets this cookie to stores the user's video player preferences using embedded YouTube video
Cookie S
Duration 1 hour
Description Used by Yahoo to provide ads, content or analytics.
Cookie test_cookie
Duration 11 months
Description This cookie is set by doubleclick.net. The purpose of the cookie is to determine if the users' browser supports cookies.
Cookie sc_at
Duration 1 year
Description Snapchat sets this cookie for showing relevant advertising based on the user’s movement.
Cookie TapAd_3WAY_SYNCS
Duration 1 month
Description TapAd sets this cookie for data synchronization with advertising networks.
Cookie _pin_unauth
Duration 1 year
Description Pinterest set this cookie to group actions for users who cannot be identified.
Cookie sc_anonymous_id
Duration 9 years
Description Soundcloud sets this cookie to enable visitors to embed content or files on the website.
Cookie um
Duration 1 year
Description Set by addthis.com.(Purpose not known)
Cookie DCRP_Categories
Duration 4 weeks
Description Description is currently not available.
Cookie vuid
Duration 2 years
Description Vimeo installs this cookie to collect tracking information by setting a unique ID to embed videos on the website.
Cookie X-AB
Duration 1 day
Description Adobe Analytics sets this cookie in context with multi-variate testing. This is a tool used to combine or change content on the website. This allows the website to find the best variation or edition of the site.
Cookie YTC
Duration 10 minutes
Description YouTube sets the YTC cookie to manage the embed and viewing of videos on the website.
Cookie sp_t
Duration 1 month
Description The sp_t cookie is set by Spotify to implement audio content from Spotify on the website and also registers information on user interaction related to the audio content.
Cookie sp_landing
Duration 1 day
Description The sp_landing is set by Spotify to implement audio content from Spotify on the website and also registers information on user interaction related to the audio content.
Cookie __asc
Duration 30 minutes
Description Alexa Metrics sets this cookie to track and report information to the Alexa analytics service.
Cookie __auc
Duration 1 year
Description Alexa Metrics sets this cookie to track and report information to the Alexa analytics service.
Cookie AWSESS
Duration
Description Awin sets this to ensure the same kind of advertisement is not shown to the user.
Cookie nevercache-b39818
Duration session
Description Description is currently not available.

REJECT Save My Preferences ACCEPT

Image 4: Logo News Hub

Premium Content

Read our exclusive articles

Facebook

Instagram

Discord Linkedin Reddit X

Image 5: Logo News Hub

Image 6: Logo MARKTECHPOST

Search AI news...

Home Editors Pick Agentic AIStateful Agentic Engineering Copilots with Letta: Core Memory, Shared State, Semantic Retrieval,...

Hand AI the chore. It gets it done. We gave it a wallet to pay its way on the web.Activate your wallet**→The job application Working✓ Filled Forgotten subscriptions Working✓ Found Snack sign-up form Working✓ Shared eBay price check Working✓ Checked300,000+ users** · +30% top-up bonus through Oct 31 Sponsored

Stateful Agentic Engineering Copilots with Letta: Core Memory, Shared State, Semantic Retrieval, Custom Tools, Streaming, and Portable Agent Workflows

Sana Hassan

October 24, 2026

Add as a preferred source on Google

In this tutorial, we build a stateful engineering copilot with Letta that demonstrates how agentic memory extends beyond conventional prompt-based conversations. We begin by creating inspectable core memory blocks that the agent can autonomously update, then expose those blocks for programmatic state synchronization and shared access across multiple agents. We further extend the architecture with archival vector memory for semantic retrieval, a custom server-side tool for deterministic risk estimation, streaming for observing agent execution, server-side message history for persistent conversations, and .af export for portability. Together, these components demonstrate a practical memory architecture in which persistent state, retrieval, tools, multi-agent collaboration, and execution traces work as an integrated agentic system.

Copy Code Copied Use a different Browser

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

We set up the Letta client, install the SDK, and add compatibility helpers that handle API variations across Letta versions. We dynamically select an available model and create a stateful engineering copilot with structured core memory blocks. We then inspect the initial memory state to establish the agent’s persistent context.

**TEXTTABLELAYOUTIs your PDF parser actually getting it right?omni_parse_bench · scores text, tables & layout · open source Star on GitHub→**Sponsored

Copy Code Copied Use a different Browser

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

We demonstrate self-editing memory by providing the copilot with durable information about the engineer, project, decisions, and known failure modes. We also update project state directly from application code and create shared read-only and writable blocks for multi-agent collaboration. We then connect a reviewer agent to the shared memory and observe how both agents exchange persistent working context.

Copy Code Copied Use a different Browser

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

We populate archival memory with tagged engineering knowledge and use semantic search to retrieve relevant information without relying on exact keyword matches. We then allow the agent to query this archival knowledge autonomously when answering an operational question. Finally, we create and attach a custom server-side release-risk tool so the agent can perform deterministic calculations and record the resulting decision in memory.

Copy Code Copied Use a different Browser

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)

We use Letta’s streaming interface to observe reasoning, tool calls, and assistant responses as the agent executes a task. We then inspect server-side conversation history to verify that the interaction state persists independently of the local Python process. We also demonstrate how the agent can be reconnected from a fresh process using only its persistent identifier.

Copy Code Copied Use a different Browser

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

We export the complete agent state to an .af file, providing a portable representation of the agent’s memory, tools, and conversation state. We then include optional cleanup logic that removes the agents and shared blocks created during the tutorial. We finish by displaying the persistent agent identifiers and the Letta ADE location for further inspection.

100K 200K 300K+users hand AI their chores.+30% wallet bonus · ends Oct 31 Claim 30% bonus**→**Sponsored

In conclusion, we now have a complete view of how Letta supports persistent, inspectable agent workflows rather than treating every interaction as an isolated prompt. We saw how core memory maintains high-value state in context, archival memory provides scalable retrieval, shared and read-only blocks coordinate multiple agents, and application code can directly synchronize external state. We also used custom tools and streaming to extend and observe agent behavior while server-side history and agent export provide persistence and portability. Through this workflow, we established a practical foundation for building engineering copilots that continuously retain context, reason over accumulated knowledge, collaborate across agents, and operate with explicit, controllable memory.

Check out the FULL CODES here. All credit goes to the researcher of this project. Also,feel free to follow us on**Twitterand don’t forget to join our150k+ML SubRedditand Subscribe toour Newsletter. Wait! are you on telegram?now you can join us on telegram as well.**

Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?Connect with us

Sana Hassan

+ posts Bio ⮌

Sana Hassan, a consulting intern at Marktechpost and dual-degree student at IIT Madras, is passionate about applying technology and AI to address real-world challenges. With a keen interest in solving practical problems, he brings a fresh perspective to the intersection of AI and real-life solutions.

$uv pip install omni-parse-bench**✓text✓tables✓layout→ scored per page Star on GitHub→**Sponsored

Previous articleNVIDIA Releases Nemotron 3.5 Bolt E4B 4GB: An Open Small Language Model for Local Voice Agents, Game Characters and Robots

Sana Hassan