用 Letta 构建有状态工程 Copilot：核心记忆、共享状态、语义检索、自定义工具与流式输出

在 Google 上添加为首选来源

在本教程中，我们使用 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 : 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 差异。我们动态选择一个可用模型，并创建一个带有结构化核心记忆块的有状态工程 Copilot。随后我们检查初始记忆状态，以建立该智能体的持久上下文。

TEXTTABLELAYOUT你的 PDF 解析器真的解析正确了吗？omni_parse_bench · 对文本、表格和布局评分 · 开源在 GitHub 上加星→赞助复制代码已复制使用其他浏览器

hr("2. 自编辑记忆——先说一次，再把记忆块读回来") chat(copilot.id, "嗨，我是 Priya，Northwind 的资深工程师。我们将在 11 月 14 日上线 'Halo'，一个 Rust 摄取服务。我讨厌冗长的解释——给我要点或代码，绝不要长篇大论。当前阻塞点：Kafka 消费者在负载下大约每 4 分钟就会重新平衡一次。") chat(copilot.id, "更新：我们把重新平衡追溯到 5 秒的 max.poll.interval.ms 以及缓慢的批量提交。决定：把提交移出轮询循环。另外请记住，我们已经试过调高 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 处于 RED：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, "把你针对 rebalance 问题的拟议修复写入 team_scratchpad " "块，以便审查员可以看到。保持在 4 行以内。") print("\n🔁 REVIEWER 在共享块中看到的内容（由另一个智能体写入）：") print(textwrap.indent(client.blocks.retrieve(scratchpad.id).value, " ")) chat(reviewer.id, "读取 team_scratchpad 并审查拟议的修复。引用它违反的任何策略，" "然后将你的裁决追加到同一个块中。") print("\n🔁 审查员回复后的共享块：") print(textwrap.indent(client.blocks.retrieve(scratchpad.id).value, " ")) print("\n附加到此块的智能体：", [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 提供关于工程师、项目、决策和已知失效模式的持久信息，来演示自编辑记忆。我们还直接从应用代码更新项目状态，并为多智能体协作创建共享的只读块和可写块。然后，我们将一个审查员智能体连接到共享记忆，并观察两个智能体如何交换持久的工作上下文。复制代码已复制使用其他浏览器

hr("5. 归档记忆：批量加载知识，然后让智能体检索它") KNOWLEDGE = [ ("事后复盘 2024-08-02：在一次部署期间消费者组重新平衡时，Halo 丢弃了 120 万条事件。根本原因：轮询循环内的提交阻塞超过了 max.poll.interval.ms。", ["postmortem", "kafka", "halo"]), ("ADR-017：Halo 使用至少一次投递，并以 (tenant_id, event_id) 为键进行幂等写入。出于吞吐量原因，通过 Kafka 事务实现精确一次被否决。", ["adr", "architecture", "halo"]), ("运行手册：要排空一个 Halo 节点，请设置 DRAIN=1，等待滞后 str: """评估按时发布一个版本的风险。参数：open_bugs (int)：未解决的 P0/P1 缺陷数量。days_to_deadline (int)：距离发布日期的日历天数。test_coverage (float)：行覆盖率，为 0.0 到 1.0 之间的小数。返回：str：0-100 的风险评分和结论。 """ 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. 流式输出智能体的执行步骤") 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"已导出 → {path}（{len(blob)} 字节）——记忆、工具和历史随之迁移。") print("稍后可用以下方式重新导入：client.agents.import_file(file=open(path,'rb'))") except Exception as e: print("此 SDK/部署不支持导出：", e) CLEANUP = False if CLEANUP: for aid in (copilot.id, reviewer.id): try: client.agents.delete(aid) except Exception as e: print("智能体删除：", e) for bid in (policies.id, scratchpad.id): try: client.blocks.delete(bid) except Exception as e: print("块删除：", e) print("已删除。") else: hr("完成") print(f"copilot : {copilot.id}\nreviewer : {reviewer.id}") print("可在 ADE 中访问 https://app.letta.com 进行可视化检查")

我们将完整的智能体状态导出为 .af 文件，从而提供智能体记忆、工具和对话状态的可移植表示。随后，我们加入可选的清理逻辑，删除教程期间创建的智能体和共享块。最后，我们显示持久化的智能体标识符以及 Letta ADE 的位置，以便进一步检查。

总之，我们现在完整了解了 Letta 如何支持持久、可检查的智能体工作流，而不是把每次交互都当作孤立的提示。我们看到了核心记忆如何在上下文中维护高价值状态，归档记忆如何提供可扩展的检索，共享块和只读块如何协调多个智能体，以及应用代码如何直接同步外部状态。我们还使用了自定义工具和流式输出来扩展和观察智能体行为，而服务端历史和智能体导出则提供了持久性和可移植性。通过这一工作流，我们为构建工程 Copilot 奠定了实用基础：它们能够持续保留上下文、基于积累的知识进行推理、跨智能体协作，并以明确、可控的记忆运行。

Sana Hassan+ 发布文章简介 ⮌

Sana Hassan 是 Marktechpost 的咨询实习生，也是 IIT Madras 的双学位学生，热衷于将技术和 AI 应用于解决现实世界的挑战。他对解决实际问题有着浓厚兴趣，为 AI 与现实生活解决方案的交汇带来了全新视角。

Sana Hassan《Cactus Whistle 编码指南：一个 16.9 MB 的语音模型能否与 Whisper Base 抗衡？在 CPU 上的词错误率、速度、流式处理、时间戳、关键词以及语音到工具调用》

$uv pip install omni-parse-bench✓文本✓表格✓布局→ 按页评分在 GitHub 上加星→赞助

用 Letta 构建有状态工程 Copilot：核心记忆、共享状态、语义检索、自定义工具与流式输出

主题

标签

正文 · AI 翻译