在这个教程中，我们用Letta来构建一个具有状态的工程copilot，以展示代理记忆如何超越常规的即时对话。 我们首先创建可检查的核心内存块，agent可以自主更新，然后暴露这些块进行程序状态同步和跨越多个agents共享访问。 我们进一步扩展该架构，包括用于语义检索的档案矢量内存、用于确定风险估计的自定义服务器侧工具、用于观察agent执行的流线、用于持续对话的服务器侧信息历史以及用于可移植性的.af导出。 这些组件一起展示了一种实用的内存架构，其中持续状态，检索，工具，多agent协作，以及执行的痕迹作为一个集成的代理系统。
导入子进程，sys
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", "letta-client"], check=True)
导入 os、 json、 时间、 文本包
从 Geops 导入到 Geops
从 letta 客户端导入 Letta
flex(*尝试，标签="):
错误=无
fn 在尝试中：
尝试：
返回 fn ()
例外作为e:
错误=e
提高运行时 Error( f) 所有调用变量失败 { label 或 ' call'}: { err} ) 出错
API KEY = os.environ.get("LETTA API KEY")或 getpass("Letta API密钥(平台.letta.com/api-keys):").
BASE_URL = os.environ.get("LETTA_BASE_URL", "").strip()
客户端=弹性(
lambda: Letta( base url=BASE URL, api key=API KEY) 如果 BASE URL 其它 Letta( api key=API KEY) 使用accessdate=需要含有url= (帮助)
Lambda: Letta( base url=BASE URL，令牌=API KEY) 如果 BASE URL 其它 Letta(token=API KEY)
标签=“客户输入”
)
def hr(标题):
打印 ("\n" + "=" * 92 + f"\n {title}\n" + "=" * 92)
def 作为 text(内容):
如果存在( 内容，字符串):
返回内容
如果存在( 内容，列表):
返回“ ”。 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("工具调用："), getatr(tc, "name", "?"), text wrap.shorten(strap(getattr(tc, "参数", "")), 200))
elif kind ======================================================================================================================================================================================================为==================================
print (“ 工具 ret: ” ), text wrap.shorten (str (gettr (m, “tool return”, “” )), 260);
elif kind == "assistant_message":
打印(“ 助手：” ), as text( getattr( m, “content”, “) )
elif kind == "approval_request_message":
打印 ("批准：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，输入=文本，最大步数=max 步数)
lambda：客户端。 agents. messages.create(agent id=agent id，输入=文本)
lambda：客户端。 agents. messages.create(agent id=agent id).
消息= [{'作用':"用户","内容": text}]
标签="messages.create"
)
若不安静：
打印( f)\n 用户：{ text}} )
显示(r)
返回 r 时
PREFERRED = ["openai/gpt-4.1", "openai/gpt-4o-mini", "anthropic/claude-sonnet-4-20250514"]
尝试：
控件 = [Getattr( m, "handle, no)" 用于客户端中的 m. models.list()]
MODEL = 下一个((如果 h 在手柄中，H 在 PreFERRED 中为h)，无) 或手柄 [0]
除外情形：
模式 = 偏好 [0]
打印(f) 连接的% 型号手柄 = {MODEL} )
hr ("1. 创建带有自定义内存块的状态 agent")
def make agent(**kw): (中文(简体) ).
返回弹性( E)
lambda：客户端。 agents.create(模型=MODEL，嵌入="openai/文本-嵌入-3-小",**kw)
lambda：客户端。 agents.create(型号=MODEL,**kw)
标签="agents.create"
)
copilot = make_agent(
名称=f"copilot-{int(time.time())}"
说明=“具有可检查长期内存的工程copilot”
标记 = [“导线”、“copilot”]
内存块=[
{"标签":"人"
"价值":"我是Mnemo，一个工程copilot. 我是铁丝网混凝土我从来没有 "
"发明事实——如果我不知道我说的话，然后去查查".
"限制": 3000}
{"标签":"人类"
"描述":"关于我与之共事的工程师的双重事实：姓名，角色，堆叠，"
"喜好，工作风格。 只要我学会稳定的东西，就更新这个".
"价值":"至今不详".
"限制": 3000}
{“ 标签”: “ project state”
"说明":"我们所航运项目的生存状况：里程碑，阻塞器，"
"决定及其理由。 保持其密集和流畅；删除陈旧的线条".
"价值":"(空)"
"限制":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 设置 block( agent id，标签，值):
返回弹性( E)
lambda：客户端。 agents. blocks.update(agent id=agent id，块 label=标签，值=值)
lambda：客户端。 agents. blocks.modify(agent id=agent id，块 label=标签，值=值)
lambda：客户端。 agents. blocks.update(agent id，标签，值=值)
标签="agents. blocks.update"
)
def turn memory(agent id，标题="核心内存"):
块 = flex(lambda: client.agents. blocks.list(agent id=agent id)
lambda：客户端。 agents.blocks.list(agent id)，标签="agents.blocks.list").
打印 (f)\n\\\\\{标题}\\\”
b 区块：
ro = “ [只读] ” 如果 getatr( b, “只读”, “只读”, False) 其他 “ ”
print(f) {b. label}{ro} ({len(b. value or '')}/{getatr(b, 'limit', '?')} 字符) )
打印( textwrap.缩进( textwrap. fill( b. value 或 ", 100), " " )
返回块
turn memory(copilot.id,"出生时的核心记忆")
我们设置了Letta客户端，安装了SDK，并增加了可处理整个Letta版本API变异的兼容性帮助。 我们动态地选择一个可用的模型，并创建一个具有结构化核心内存块的状态工程copilot. 然后检查初始内存状态以建立agent的持久上下文。
hr("2. 自编自导的记忆——说一次，然后读回块").
聊天( copilot. id)
"嗨，我是普瑞娅，北风的员工工程师" 我们正在运输"光环"，一种锈蚀摄入"
"服务，在11月14日。 我讨厌长篇解释——给我子弹或密码，不要散文。 "
"当代阻塞器：卡夫卡消费者在负载下每~4分钟重新平衡一次". )
聊天( copilot. id)
"更新：我们追踪重平衡到最大5秒民调中间数慢批量承诺"
"决定：移出民调圈。 也记住我们已经试过了
"session.timeout.ms and it did nothing — don't suggest that again.")
turn memory(copilot.id),"2转后的核心内存(注意移动到哪里)"
hr ("3. 您的代码写入上下文窗口" )
live = 获得 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
只读=真
)
crawpad = 客户端。 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=[{"标签":"人"
"价值":"我是阿尔格斯，一个偏执的员工评论员。 我寻找失败模式 "
"其他人都跳过，我逐个引用团队政策。"]
)
print ("审查者 id:"，审查者.id)
聊天( copilot. id),“ 将您为重新平衡问题提议的固定写入团队 scratchpad ”
"堵住，让评论员能看到。 保持为4行。 )
print ("\n IPER在共享块中看到的(由其他 agent 写入):")
打印(textwrap.incent(client.blocks.retrieve(scratchpad.id))) 值，" " )
聊天(reviewer.id),"Read team scratchpad并审查提议的固定。 违反任何政策
"然后把判决附在同一个街区。"
print (“\n 共享块在审查者答复后 ” )
打印(textwrap.incent(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) 互联网档案馆的存檔，存档日期2013-09-02.
标签="块.agents.list")].
我们通过向copilot提供有关工程师，项目，决定和已知故障模式的持久信息，来展示自我编辑记忆。 我们还直接从应用程序代码更新工程状态，并为多agent合作创建共享的只读和可写区块。 然后我们把一个评论员agent与共享内存连接起来，观察agents两个如何交换持续的工作环境。
hr ("5. 存档内存：散装知识，然后让 agent 检索它").
知识 = [
("死后2024-08-02：光环下降1.2M事件当消费者群体在一个"
"调遣。 根因子：在民调环路被阻断的内，通过最大。民调.interval.ms".
["死后","kafka","halo"].
("ADR-017：光环使用最少的一送即用同位素键写(tenant id，事件 id)). "
"通过卡夫卡的交易被否决了"
["adr", "architecture", "halo"]),
("运行本：排出一个光环节点，设置DRAIN=1，等待滞后<1000，然后SIGTERM. 永远不要笑 - "
"飞机舱只检查关闭优雅。"
["运行本","ops","halo"].
("基准测试 2025-01:rdkafka 2.3,8个分区在p99 32ms上持续了240k事件/s).
"c6i.4x大；瓶颈是JSON剖析，而不是网络".
["基准测试","性能"].
("Priya在文件布局上的偏好：每个限定上下文一个模块，集成测试直播").
"旁边的模块，没有测试/巨型目录",["首选","风格"]).
("事件2025-03-19：一个欧盟租户的有效载荷被反射到我们-东-140分钟。 "
"被区域包围 DLQ. 这就是为什么政策3存在",["事件","遵守"].
]
def 归档( agent id，内容，标记):
返回弹性( E)
lambda：客户端。 agents. passages. 插入(agent id=agent id，内容=内容，标签=tags) 互联网档案馆的存檔，存档日期2013-09-02.
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]:
打印 (“ ” ), text wrap.shorten( getattr( h, "content"，无) 或 getattr( h, "text", "", "", 150))
聊天(copilot.id),"在我今晚重启节点之前：我们历史上有什么我应该做的"
"担心，什么是安全程序？"
hr ("6. 给 agent 一个自定义工具").
def 估计 release risk(open bugs: int, days to deadline: int, test coverage: fload) - > str:
""减轻沉船的风险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 Atlas. All rights belong to the original author.