RadarAI· MarkTechPost（RSS）·· 4 小时前AI 评分58
Letta 实战教程：用核心记忆、共享状态与语义检索构建有状态工程副驾驶
AI 导读
MarkTechPost 发布 Letta 实战教程，演示如何构建有状态工程副驾驶智能体，涵盖可自主更新的核心记忆块、跨智能体共享记忆、档案向量记忆的语义检索、自定义服务端工具、流式执行和服务端消息历史。教程最后将完整智能体状态导出为 .af 文件实现便携迁移，并给出完整代码。
正文 · AI 翻译
MarkTechPost· Sana Hassan·2026-10-24 18:29· 1 小时前AI 评分58
使用 Letta 构建有状态智能体工程副驾驶：核心记忆、共享状态、语义检索、自定义工具、流式传输与可移植智能体工作流
AI 导读
本教程演示如何用 Letta 构建有状态的工程副驾驶智能体，涵盖可自主更新的核心记忆块、跨智能体共享记忆、档案向量记忆的语义检索、自定义服务端工具、流式执行和服务端消息历史。文末将完整智能体状态导出为 .af 文件实现便携迁移，并给出完整代码。
正文 · 原文
在本教程中，我们使用 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")
我们将完整的 agent 状态导出为 .af 文件，提供 agent 记忆、工具和对话状态的可移植表示。随后我们加入可选的清理逻辑，删除教程期间创建的 agent 和共享块。最后，我们展示持久化的 agent 标识符以及 Letta ADE 位置，以便进一步检查。
总之，我们现在完整地了解了 Letta 如何支持持久、可检查的 agent 工作流，而不是将每次交互都视为孤立的提示。我们看到了核心记忆如何在上下文中维护高价值状态，归档记忆如何提供可扩展的检索，共享块和只读块如何协调多个 agent，以及应用代码如何直接同步外部状态。我们还使用了自定义工具和流式传输来扩展和观察 agent 行为，而服务端历史和 agent 导出则提供了持久性和可移植性。通过这一工作流，我们为构建工程 copilot 奠定了实用基础，使其能够持续保留上下文、对积累的知识进行推理、跨 agent 协作，并以显式、可控的记忆运行。
查看完整代码。所有功劳归于本项目的研究者。另外，欢迎在Twitter上关注我们，别忘了加入我们的150k+ ML SubReddit并订阅我们的 Newsletter。等等！你在 Telegram 上吗？现在你也可以在 Telegram 上加入我们。
需要与我们合作推广你的 GitHub 仓库、Hugging Face 页面、产品发布或网络研讨会等？联系我们
文章使用 Letta 构建有状态 Agentic 工程 Copilot：核心记忆、共享状态、语义检索、自定义工具、流式传输与可移植 Agent 工作流首次发表于MarkTechPost。
来源：MarkTechPost · marktechpost.com
来源：RadarAI · aihot.news