在这个教程中，我们用Letta来构建一个具有状态的工程copilot，以展示代理记忆如何超越常规的即时对话。 我们首先创建可检查的核心内存块，agent可以自主地更新，然后暴露出这些块进行程序状态同步和跨越多个agents共享访问。 我们进一步扩展了架构，包括用于语义检索的归档向量内存、用于确定风险估计的自定义服务器侧工具、用于观察agent执行的流线、用于持续对话的服务器侧信息历史以及用于可移植性的.af导出。 这些组件一起展示了一种实用的内存架构，其中持久状态，检索，工具，多agent协作，以及执行痕迹作为一个集成的代理系统。
导入子进程，sys
subprocess.run ([sys. executive,"-m","pip","install","-q","-U","letta-client"],"检查=True").
导入 os, json，时间，文本包
从接通导入接通
从 letta 客户端导入 Letta
flex(*尝试，标签="):
错误=无
fn 在尝试中：
尝试：
返回 fn ()
例外作为e:
错误=e
提高运行时 Error( f) 所有调用变量失败 { label 或 ' call'}: {err}} 从错误中调用。
API KEY = os.environ.get ("LETTA API KEY")或取出("Letta API key (plateform.letta.com/api-keys):").
BASE_URL = os.environ.get("LETTA_BASE_URL", "").strip()
客户端=弹性(
羊肉：Letta(base url=BASE URL,api key=API KEY) 如果BASE URL else Letta(api key=API KEY)，则使用该词。
羊肉：Letta(base url=BASE URL，令牌=API KEY) 如果BASE URL 其它 Letta(token=API KEY)
标签=“客户输入”
)
df hr(标题):
打印 ("\n" + "=" * 92 + f"\n {title}\n" + "=" * 92)
def 为 text(内容):
如果存在( 内容，st):
返回内容
如果存在( 内容，列表):
返回 “. ”。 join( getattr( p, “ text”, “ “” ) 或 p 内容中的“ “ ” )
返回 str( 内容或“ ” )
def 显示( 响应，隐藏解释=虚假 ):
用于获取器(响应、“信息”、响应)或[:
type = getatr(m,"message type"，无)或 getatr(m,"type","'?)
而不是隐藏。.. 解释：
print ("推理："), text wrap.shorten (gettr (m,"解释","""或",","220")).
elif kind ====================================================================================================================================================================为====================================
tc = getatr(m,"tool call"，无)
print("工具调用：")，取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取
elif kind ================================================================================== (===================================================================================为=====================为=====================
print ("工具 ret:"), text wrap.shorten (str (gettr (m, "tool return", "")", "260") (中文(简体) ).
elif kind =================================================================== (=====================================================================================================为======================为=======================
打印("助手："), as text(getattr(m),"内容","")))
elif kind ="批准请求消息":
print ("批准":agent正等待您批准一个工具呼叫").
u = 得到(答复，“使用”，无)
如果为 u:
print (f) 阶梯 QQGetattr(u, “ step counts”, “? ”) 符号 QQGetattr(u, “total tokens”, “? ”) )
def 聊天( agent id，文本，max steps=12，安静=假):
r=弹性(
lambda：客户端.agents.messages.create(agent id=agent id，输入=text，最大 steps=max steps) 互联网档案馆的存檔，存档日期2013-12-21.
羊肉：客户端。 agents. messages.create(agent id=agent id，输入=text) 羊肉：
羊肉：客户端。 agents. messages.create(agent id=agent id) 羊肉：
消息=[{"作用":"用户","内容":text}]
标签="messages.create"
)
若不安静：
打印(f"\n 用户：{text}")
显示(r)
返回 r 时
活塞= ["openai/gpt-4.1","openai/gpt-4o-mini","anthropic/claude-sonnet-4-20250514"].
尝试：
handers = [getatr(m,"handle","none")在客户端中表示 m.models.list (]
MODEL = 下个((如果 h 在手柄中，H 在 PreFERRED 中为h)，无) 或手柄 [0]
除外：
模型 = 预选 [0]
打印(f) 连接到 ^ 模型手柄 = {MODEL}" )
hr ("1. 创建带有自定义内存块的状态 agent").
def make agent (**kw) (中文(简体) ).
返回弹性( E)
羊肉：客户端。 agents.create(模型=MODEL，嵌入="openai/text-embedding-3-small",**kw) 出品人：
羊肉：客户端。 agents. Creative(模型=MODEL,**kw)
标签="agents. Create"
)
copilot = 制造 agent ()
名称=f"copilot-{int(time.time ())}"
说明=“具有可检查长期内存的工程copilot”
标记= ["图","copilot"]
内存块=[
{"标签":"人"
"价值":"我Mnemo，一个工程copilot. 我是铁丝网，混凝土，我从来没有 "
"发明事实——如果我不知道我说的话，然后去查查".
"限制": 3000}
{"标签":"人类"
"描述":"关于我工作的工程师的可论证事实：姓名，角色，堆，"
"喜好，工作风格。 只要我学会稳定的东西，就更新这个".
"价值":"至今不详".
"限制": 3000}
{"标签":"项目状态"
"描述":"我们所运输的项目的生存状态：里程碑，阻塞器，"
"决定及其理由。 保持其密集而流畅；去掉被打乱的行".
"值":"(空)".
"限制":5000]
{"标签":"教训"
"叙事":"哈尔沁教与出错再不重复。 仅附录一行，每行一行".
"值":"(空)".
"限制":2000]
],
)
打印 ("agent id:", copilot.id)
def 获取块( agent id，标签):
返回弹性( E)
羊肉：客户端。 agents. blocks.retrieve(agent id=agent id，块 label=label)
羊肉：客户端。 agents. blocks.retrieve(agent id，标签)
标签="agents. blocks.retrieve"
)
def set block( agent id，标签，值):
返回弹性( E)
lambda：客户端。 agents. blocks.update(agent id=agent id，块 label=标签，值=值)
羊肉：客户端。 agents. blocks.modify(agent id=agent id，块 label=标签，值=值)
Lambda：客户端。 agents. blocks.update(agent id，标签，值=值)
标签="agents.blocks.update" (中文(简体) ).
)
def dump memory(agent id，标题="核心内存"):
块 = 有弹性 (lambda: client.agents. blocks.list (agent id=agent id) )
lambda：客户端。 agents.blocks.list(agent id)，标签="agents.blocks.list").
打印(f"\n────{标题}}\")
b 区块：
ro = "[只读]"，如果取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取取
print(f) {b. label}{ro} ({len(b. value or "')}/{gettr(b,'limit','?')}字符) )
打印(text wrap.缩进(textwrap.fill (b.value 或 ",","100",""))
返回块
turn memory(copilot.id,"出生时的核心记忆")
我们设置了Letta客户端，安装了SDK，并增加了兼容性帮助器，处理Letta版本的API变体。 我们动态地选择一个可用的模型，并创建一个具有结构化核心内存块的状态工程copilot. 然后检查初始内存状态，以建立agent的持续上下文。
hr ("2. 自编自导自导的记忆——说一次，然后读出块回").
聊天( copilot. id)
"嗨，我是Priya，北风的员工工程师。 我们正在运输"光环"，一种粗糙的摄入"
"服务，在11月14日。 我讨厌长长的解释——给我子弹或代码，永远不要作文章。 "
"Curent Blocker：卡夫卡消费者在负载下每~4分钟重新平衡". ).
聊天( copilot. id)
"更新：我们追踪到重平衡最大5秒民调。 中间慢批量承诺。"
" 决定：移出民调圈。 也记住我们已经尝试了碰碰"
”“课会。超时”和“无所作为”——不要再次暗示。 ”
dump memory(copilot.id),"2个转弯后的核心内存(注意移动到哪里)".
hr ("3. 您的代码写入上下文窗口" )
活 = 得到 block(copilot.id,"Project state")
设置块( copilot. id), “ project state”
(live.value or "") + "n [CI 09:41Z] 主干为RED：集成/kafka reable test 失败(flaky x3)".
聊天(copilot.id,"在我开始之日前应该知道什么？").
hr ("4. 在两个 agents 之间共享内存")
策略 = 客户端。 blocks.create(
标签=“政策”
说明=“不可谈判的工程政策。 只读：服从，不编辑".
值="1. 无直接向主推。\n2. 每个有回归测试的固定船。
"3. 客户数据从未离开欧盟区域".
限额=2000
仅读为真
)
抓取板 = 客户端。 blocks.create ()
标签="团队-scratchpad"
说明=“在copilot和审查员agent之间共享工作指出。”
"两个agents在这里读和写；保持条目前缀你的名字".
数值="(空)"
限额=4 000
)
(政策.id, crawpad.id):
flex( lambda：客户端。 agents. blocks. attach( agent id=copilot. id，块 id=bid)
羊肉：客户端。 agents. blocks.attach(copilot.id，出价)
标签="agents. blocks.attach").
审查者 = make agent ()
name=f"审查器-{int(time.time ())}"
描述=“与 copilot 共享内存的 Strict 代码审查员”
标记= [“审校”]
block ids=[政策.id，抓图.id]
内存 blocks=[{"标签":"人"
"价值":"我是Argus，一个偏执的工作人员评论员。 我寻找失败模式 "
"其他人都跳过，我逐个引用团队政策"].
)
print ("审查者 id:"，审查者.id)
聊天( copilot. id),“ 将您为重新平衡问题提出的修正写入团队 scratchpad”
"堵住，让评论员能看到。 保持为4行". )
print ("\n IPER在共享块中看到的(由其他 agent 所写):")
print(textwrap.indent (client.blocks.retrieve (scratchpad.id)). ) 值，" " " )
聊天(reviewer.id),"Read team scratchpad并审查提议的固定。 煽动其违反的任何政策，"
"然后在同一个街区附上你的判决。"
print (“\n 共享块在审查者答复后 ” )
print(textwrap.indent (client.blocks.retrieve (scratchpad.id)). ) 值，" " " )
打印( “此块所附带的正则 ” )
[a.name for a in flex(lambda: client.blocks.agents.list (block id=scratchpad.id)) 中一个在flex(lambda: client.blocks.
羊肉：客户端。 blocks.agents.list(scratchpad.id) 页面存档备份，存于互联网档案馆。
标签="块.agents.list")].
我们通过向copilot提供有关工程师、项目、决定和已知故障模式的持久信息来展示自我编辑记忆。 我们还直接从应用程序代码更新工程状态，并创建共享的只读可写块来进行多agent协作。 然后，我们将一个评论员agent与共享内存连接起来，并观察agents双方如何交换持续的工作环境。
hr ("5. 存档内存：散装知识，然后让 agent 检索).
认识 = [
("后尸检"2024-08-02：光环下降1.2M事件当消费者群体在一次"
"调出。 根因：在被屏蔽的民调回路内
["死后","kafka","halo"].
("ADR-017：光环在发送时使用最少的一分一秒，并用一分一分一分的写法键入 (tenant id, event id).
"通过卡夫卡的交易被否决了"
["adr", "architecture", "halo"]),
("运行本：要排出一个光环节点，设置 DRAIN=1，等后差<1000，再由SIGTERM. 永远不要签字 - "
"飞机舱只检查优雅的关闭。"
["跑书","通","哈罗".
("基准测试 2025-01:rdkafka 2.3 有8个分区在p99 32ms上持续了240k事件/s).
"c6i.4x大；瓶颈是JSON解析，而不是网络".
["基准测试","性能"].
("Priya在文件布局上的偏好：每个限定上下文一个模块，集成测试活性").
"旁边的模块，没有测试/特大指令",["首选","风格"].
("事件2025-03-19：一个欧盟租户的有效载荷被镜像到我们东一号40分钟。 "
"被地区包围 DLQ. 这就是为什么政策3存在",["事件","遵守"].
]
def 归档( agent id，内容，标记):
返回弹性( E)
lambda：客户端。 agents. passages. 插入(agent id=agent id，内容=内容，标签=tags)
lambda：客户端。 agents.passages.create(agent id=agent id，文本=内容，标签=tags)
lambda：客户端。 agents.passages.create(agent id=agent id，文本=内容)
标签="agents. Passages. 插入"
)
在“知识”中显示内容、标记：
存档( copilot. id，内容，标签)
打印(f) “插入了{len(KNOWLEDGE)}段落” )
达到 = 弹性( )
lambda：客户端.agents.passages.search(agent id=copilot.id，查询="部署时丢失消息").
lambda：客户端.agents.passages.list(agent id=copilot.id，搜索="部署时丢失消息").
标签="agents. passages. search" (中文(简体) ).
)
print ("\n 程序语义搜索“ 部署时丢失消息 ” )
(达到)[:3]:
print ("-"), text wrap.shorten (gettr (h,"content","none") 或 getatr (h,"text",""","",",",",",",")
聊天(copilot.id),"在我今晚重启节点之前：我们历史上有什么值得我做的"
"担心，什么是安全程序？"
hr ("6. 给 agent 一个自定义工具").
def 估计 release risk(开出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出
""减轻出船的风险p
本文由 AI宝库自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
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.
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.
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.
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 shipp
Auto-fetched and translated by AI Gem Box. All rights belong to the original author.