在本教程中，我们使用 Letta 构建一个有状态的工程 copilot，展示智能体记忆如何超越传统的基于提示词的对话。我们首先创建可检查的核心记忆块，智能体可以自主更新这些块，然后将其暴露出来，用于程序化状态同步和跨多个智能体的共享访问。我们进一步扩展该架构，加入用于语义检索的归档向量记忆、用于确定性风险估计的自定义服务端工具、用于观察智能体执行的流式传输、用于持久对话的服务端消息历史，以及用于可移植性的 .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 : 智能体正在等待你批准一次工具调用")
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. 创建一个带自定义记忆块的有状态智能体")
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="具备可检查长期记忆的工程副驾驶",
tags=["tutorial", "copilot"],
memory_blocks=[
{"label": "persona",
"value": "我是 Mnemo，一名工程副驾驶。我说话简洁、具体，从不"
"编造事实——如果我不知道某件事，我会说出来，然后去查。",
"limit": 3000},
{"label": "human",
"description": "关于与我共事的工程师的持久事实：姓名、角色、技术栈、"
"偏好、工作风格。每当我了解到稳定信息时就更新这里。",
"value": "目前未知。",
"limit": 3000},
{"label": "project_state",
"description": "我们正在交付的项目的实时状态：里程碑、阻塞项、
决策及其理由。保持内容密集且最新；删除过时的行。",
"value": "(空)",
"limit": 5000},
{"label": "lessons",
"description": "来之不易的教训和绝不再犯的错误。仅追加，每行一条。",
"value": "(空)",
"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 差异。我们动态选择一个可用模型，并创建一个带有结构化核心记忆块的有状态工程 copilot。随后我们检查初始记忆状态，以建立该智能体的持久上下文。
复制代码已复制使用其他浏览器
hr("2. 自编辑记忆——先对话一次，再把记忆块读回来")
chat(copilot.id,
"嗨，我是 Priya，Northwind 的资深工程师。我们将在 11 月 14 日发布 'Halo'，一个 Rust 摄取
服务。我讨厌冗长的解释——给我要点或代码，绝不要长篇大论。
当前阻塞项：Kafka 消费者在负载下大约每 4 分钟就发生一次再均衡。")
chat(copilot.id,
更新：我们把这次再平衡追溯到了 5s 的 max.poll.interval.ms 加上缓慢的批量提交。
决定：把提交移出 poll 循环。另外记住我们已经试过调大
session.timeout.ms，没有任何效果——不要再建议这个了。）
dump_memory(copilot.id, "两轮之后的核心记忆（记录什么移动到了哪里）")
hr("3. 你的代码写入上下文窗口")
live = get_block(copilot.id, "project_state")
set_block(copilot.id, "project_state",
(live.value or "") + "\n[CI 09:41Z] main 是红的：integration/kafka_rebalance_test 失败（已不稳定 3 次）。")
chat(copilot.id, "开始一天工作之前，有什么我应该知道的吗？")
hr("4. 两个智能体之间的共享记忆")
policies = client.blocks.create(
label="policies",
description="不可协商的工程政策。只读：遵守，绝不编辑。",
value="1. 禁止直接推送到 main。\n2. 每个修复都附带回归测试。\n"
"3. 客户数据绝不离开欧盟区域。",
limit=2000,
read_only=True,
)
scratchpad = client.blocks.create(
label="team_scratchpad",
description="copilot 与 reviewer 智能体之间的共享工作笔记。"
"两个智能体都在这里读写；条目请以你的名字作为前缀。",
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="与 copilot 共享记忆的严格代码审查者",
tags=["tutorial", "reviewer"],
block_ids=[policies.id, scratchpad.id],
memory_blocks=[{"label": "persona",
"value": "我是 Argus，一个偏执的资深审查者。我寻找别人都跳过的失败模式，"
"并且按编号引用团队政策。"}],
)
print("reviewer id:", reviewer.id)
chat(copilot.id, "把你针对再平衡问题的修复方案写进 team_scratchpad "
"块，好让 reviewer 能看到。控制在 4 行以内。")
print("\n REVIEWER 在共享块中看到的内容（由另一个智能体写入）：")
print(textwrap.indent(client.blocks.retrieve(scratchpad.id).value, " "))
chat(reviewer.id, "读取 team_scratchpad 并审查提出的修复方案。引用它违反的任何政策，"
然后把你的结论追加到同一个 block。
print("\n 审阅者回复后的共享 block：")
print(textwrap.indent(client.blocks.retrieve(scratchpad.id).value, " "))
print("\n 附加到此 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")])
我们通过向 copilot 提供关于工程师、项目、决策和已知故障模式的持久信息，来演示自编辑记忆。我们还直接从应用代码更新项目状态，并为多智能体协作创建共享的只读和可写 block。随后，我们将一个审阅者智能体连接到共享记忆，并观察两个智能体如何交换持久的工作上下文。
复制代码已复制使用其他浏览器
hr("5. 归档记忆：批量加载知识，然后让智能体检索它")
KNOWLEDGE = [
("事后复盘 2024-08-02：在一次部署期间消费者组重新平衡时，Halo 丢弃了 120 万条事件。
根本原因：poll 循环内的提交阻塞超过了 max.poll.interval.ms。",
["postmortem", "kafka", "halo"]),
("ADR-017：Halo 采用至少一次投递，并以 (tenant_id, event_id) 为键进行幂等写入。
出于吞吐量原因，通过 Kafka 事务实现精确一次被否决。",
["adr", "architecture", "halo"]),
("运行手册：要排空一个 Halo 节点，设置 DRAIN=1，等待 lag<1000，然后发送 SIGTERM。绝不要 SIGKILL——
在途批次只在优雅关闭时才会被检查点保存。",
["runbook", "ops", "halo"]),
("基准测试 2025-01：rdkafka 2.3 在 8 个分区下，于 c6i.4xlarge 上以 p99 32ms 持续达到 24 万事件/秒；
瓶颈是 JSON 解析，而不是网络。",
["benchmark", "performance"]),
("Priya 对文件布局的偏好：每个限界上下文一个模块，集成测试放在模块旁边，
不要 tests/ 巨型目录。", ["preference", "style"]),
("事件 2025-03-19：一个欧盟租户的负载被镜像到 us-east-1 长达 40 分钟。
通过将 DLQ 固定到区域修复。这就是策略 3 存在的原因。", ["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, "我们有 9 个未解决的 P0/P1 缺陷，距离 11 月 14 日还有 12 天，覆盖率为 0.61。"
用工具评估风险，然后把结论记录到 project_state 中。
我们用带标签的工程知识填充归档记忆，并使用语义搜索来检索相关信息，而不依赖精确的关键词匹配。然后，我们允许智能体在回答运维问题时自主查询这些归档知识。最后，我们创建并挂载一个自定义的服务端发布风险工具，使智能体能够执行确定性计算并将结果决策记录到记忆中。
复制代码已复制使用其他浏览器
hr("7. 流式输出智能体的步骤")
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, "用一行给出 Halo 的状态，仅从记忆中获取。不使用工具。")
except Exception as e:
print("此 SDK 版本不支持流式输出：", e)
hr("8. 服务端对话历史")
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 仅用 id 从新进程重新连接：")
print(" client.agents.messages.create(agent_id='%s', input='...')" % copilot.id)
我们使用 Letta 的流式接口来观察智能体执行任务时的推理过程、工具调用和助手响应。随后我们检查服务端的对话历史，以验证交互状态独立于本地 Python 进程而持久存在。我们还演示了如何仅凭其持久标识符，从一个全新进程重新连接该智能体。
复制代码已复制使用其他浏览器
hr("9. 导出 / 导入智能体")
try:
dump = flex(lambda: client.agents.export_file(agent_id=copilot.id),
lambda: client.agents.export_file(copilot.id), label="agents.export_file")
blob = dump if isinstance(dump, (str, bytes)) else json.dumps(dump, default=str)
path = "/content/copilot.af" if os.path.isdir("/content") else "copilot.af"
with open(path, "wb" if isinstance(blob, bytes) else "w") as f:
f.write(blob)
print(f"exported → {path} ({len(blob)} bytes) — memory, tools and history travel with it.")
print("re-import later with: client.agents.import_file(file=open(path,'rb'))")
except Exception as e:
print("export not available on this SDK/deployment:", e)
CLEANUP = False
if CLEANUP:
for aid in (copilot.id, reviewer.id):
try: client.agents.delete(aid)
except Exception as e: print("agent delete:", e)
for bid in (policies.id, scratchpad.id):
try: client.blocks.delete(bid)
except Exception as e: print("block delete:", e)
print("deleted.")
else:
hr("Done")
print(f"copilot : {copilot.id}\nreviewer : {reviewer.id}")
print("Inspect them visually in the ADE at https://app.letta.com")
我们将完整的智能体状态导出为 .af 文件，从而提供智能体内存、工具和对话状态的可移植表示。随后我们加入了可选的清理逻辑，用于删除教程期间创建的智能体和共享块。最后我们展示持久化的智能体标识符以及 Letta ADE 的位置，以便进一步检查。
总之，我们现在已经完整地了解了 Letta 如何支持持久、可检查的智能体工作流，而不是把每次交互都当作孤立的提示词来处理。我们看到了核心记忆如何在上下文中维护高价值状态，归档记忆如何提供可扩展的检索，共享块和只读块如何协调多个智能体，以及应用代码如何直接同步外部状态。我们还使用了自定义工具和流式传输来扩展和观察智能体行为，而服务端历史和智能体导出则提供了持久性和可移植性。通过这一工作流，我们为构建工程副驾驶奠定了实用基础，使其能够持续保留上下文、基于积累的知识进行推理、跨智能体协作，并以显式、可控的记忆运行。
在此处查看完整代码。所有功劳归于该项目的研究者。另外，欢迎在 Twitter 上关注我们，别忘了加入我们拥有 15 万+ 成员的 ML SubReddit 并订阅我们的新闻通讯。等等！你在 Telegram 上吗？现在你也可以在 Telegram 上加入我们。
需要与我们合作推广你的 GitHub 仓库、Hugging Face 页面、产品发布或网络研讨会等？联系我们
文章《Stateful Agentic Engineering Copilots with Letta: Core Memory, Shared State, Semantic Retrieval, Custom Tools, Streaming, and Portable Agent Workflows》首次发布于 MarkTechPost。