独立仓 Agent 开发手把手 9 步
Day 19 讲了"两种起 agent 方式"。本篇把其中的"独立仓形态"展开成一条可照抄跑通的完整实战路径——从下模板到 obelisk 上线。这是 Day 19「独立仓形态」的实战放大版:你不在小区(monorepo)里盖房,而是在小区外自建一间小屋,但通过「物业供应链(Nexus)」订购同款承重构件(可信底座)。跟着 9 步照抄就能跑通。
ai-trust-toolkit-bom[core]==0.6.0),同款承重构件送货上门。~/.netrc = 你的供货仓门禁卡(卡号打错就进不去 → 401);telemetry = 屋里装个用量上报器,让小区物业大屏也能看到你这间小屋的效果;安全自检 7 问 = 你拿物业万能钥匙(服务凭据)帮人开门时,绝不能让没权限的人经你的手开了不该开的门。这个 skill 是什么
平台带了一个叫 standalone-repo-agent 的 skill(一个 SKILL.md)。回忆 Day 19:它是"在你自己的独立 git 仓库(不在平台 monorepo)里从零起一个 agent"的手把手指南。跟着抄就能跑通,例子叫 demo-agent,换成你的名字即可。活范例是 bmc-agent(CLI+MCP+telemetry+发 Nexus 全套)。
🎯 它解决的场景:你没有平台 monorepo 的权限,但想做一个用平台可信底座(ai-trust-toolkit)的 agent——在自己仓里开发、跑在自己机器上(CLI/MCP 形态,数据不出门),框架通过公司 Nexus 一行依赖接全。
两种形态:monorepo 内 vs 独立仓
先厘清和 Day 19 的关系——平台有两条起 agent 的路,本篇专讲第二条:
① monorepo 内加(Day 19 主线)
agent 进平台的apps/,成为 workspace 成员,改 toolkit 立即生效。用 agent-onboarding skill。平台核心团队走这条。② 独立仓开发(本篇)
在你自己的 git 仓,通过 Nexus 一行 BOM 依赖用 toolkit。用standalone-repo-agent skill。外部业务团队(无 monorepo 权限)走这条。ai-trust-toolkit-bom[core]==0.6.0 拉包)。BOM 保证两条路用的是同一套可信底座,只是获取方式不同——一个本地 workspace 引用,一个 Nexus 拉包。别混淆:要在 monorepo 内加 agent 用 agent-onboarding,不是本 skill。本篇还有一个"分叉"——起 agent 也有手动 vs 一键两种:
手动起(本篇 · 推荐新手)
下模板 → 改名 → 自己填图。每一步看得见、改起来透明。一键生成(agent-creator 驱动)
用 portal「新建 agent」向导,填名 + 选 BOM extras → 自动吐整套工程 → 下 zip。懒得手动/要批量时用。9 步总览(会自己走一遍)
下面时间线会依次点亮,先建立"从零到上线要经过哪 9 步"的整体感(下面各讲逐步展开):
本地环境检查
Python ≥ 3.10 + uv + git。这是个跑在你本机的本地 agent(数据不出门)。
起仓
平台直链下模板 → 改名(your_agent → 你的)→ git init。
配 Nexus 拉取凭据
用只读账号 pypi_pull 写进 ~/.netrc(一次性)。
装 + 配 LLM key + 跑通
uv sync → 填 ANTHROPIC_API_KEY → uv run demo-agent run "你好"。
写你的业务图(核心)
改 nodes/ + builder.py,LLM 走 get_llm()。
接 telemetry
record_effect + push_invoke_metrics,让平台看见你的效果。
选额外入口形态
CLI(默认)/ +MCP(给 IDE)/ +pod(常驻端点),多选叠加。
测
uv run --extra test pytest(模板自带冒烟测试)。
安全自检
涉权必做:7 问逐条答,命中红线走独立评审。
上线
先测试环境验过,再生产(obelisk 灰度 / Nexus 发布)。
步骤 0-1 · 环境与起仓
步骤 0 · 本地环境检查
这是个本地 agent——开发和运行都在你本机(默认 CLI 形态),代码/数据不出门。开工确认两样:Python ≥ 3.10(python3 --version)+ uv(包管理+venv):
curl -LsSf https://astral.sh/uv/install.sh | sh # 装 uv(macOS 也可 brew install uv)
凭据三类,都别提交进 git:Nexus 只读拉取账号(步骤 2,团队共用)· LLM key(业务方自己填)· 发布写凭据(找 @Valiant)。
步骤 1 · 起仓(2 分钟)
# A) 平台免登录直链下模板(不用任何 git 仓权限)
curl -fL http://gov-agents-ui.bmaicsaws-prod.com/portal/v1/agent-template/download.zip -o /tmp/tpl.zip
unzip /tmp/tpl.zip && mv standalone-agent-template demo-agent && cd demo-agent
# 2) 改名:your-agent / your_agent → 你的
mv your_agent demo_agent
grep -rl 'your[-_]agent' . | xargs sed -i '' 's/your-agent/demo-agent/g; s/your_agent/demo_agent/g'
# ↑ macOS 用 sed -i '';Linux 用 sed -i(去掉那对空引号)
# 3) git init
git init && git add -A && git commit -m "init demo-agent"
pyproject.toml 有几段关键(模板已带,别删)——它们让 uv 从公司 Nexus 拉框架:
[project]
dependencies = ["ai-trust-toolkit-bom[core]==0.6.0", "langchain-anthropic>=0.3.0", "typer>=0.12.0", ...]
[[tool.uv.index]] # 框架从公司 Nexus 拉
name = "bm-nexus"
url = "http://nexus.bitmartpro.com/repository/local-pipy/simple/"
[tool.uv]
index-strategy = "unsafe-best-match" # 跨 Nexus+PyPI 挑满足约束的最优版
[tool.uv.sources]
ai-trust-toolkit-bom = { index = "bm-nexus" } # 内部包钉死走 Nexus(防被公共 PyPI 同名顶替)
ai-trust-toolkit-bom[core]==0.6.0 就拉到整套可信底座(critic/failsafe/memory/llm…),不用自己 pin 一堆子包。index-strategy = unsafe-best-match 是因为 Nexus 上有些公共包是旧版,这行让 uv 能跨 Nexus 和 PyPI 挑出满足约束的最优版本。[[tool.uv.index]] name=bm-nexus → 告诉 uv「除了 PyPI,还有一个公司供货仓地址」[tool.uv.sources] ai-trust-toolkit-bom = {index="bm-nexus"} → 「这个内部包钉死走 Nexus」,防止公共 PyPI 上有同名包把它顶替掉(供应链投毒防护)index-strategy="unsafe-best-match" → 「跨两个仓一起挑满足约束的最优版本」→ 效果:
uv sync 时框架从 Nexus 拉、其它公共库从 PyPI 拉,各取所长。步骤 2-3 · Nexus 凭据与跑通
步骤 2 · 配 Nexus 拉取凭据(一次性)
用平台的只读拉取账号 pypi_pull(团队共用,uv 自动读 ~/.netrc)。用 printf 写、login/password 不加引号:
printf 'machine nexus.bitmartpro.com\nlogin pypi_pull\npassword 08gk5p2Q\n' >> ~/.netrc
chmod 600 ~/.netrc
cat ~/.netrc # 确认:login pypi_pull(无引号)· password 无引号
uv sync 报 401 多半是 netrc 写歪了——用户名打错、值带了引号、或 machine 行漏了。cat ~/.netrc 逐字核对。这也是 skill 触发词里专门列了 "uv sync 401 / Nexus 拉不下来" 的原因。步骤 3 · 装 + 配 LLM key + 跑通
uv sync # 从 Nexus 拉框架 + 建 venv
cp .env.example .env # 编辑 .env:ANTHROPIC_API_KEY=sk-ant-...
# 或 export ANTHROPIC_API_KEY=sk-ant-...(CI/临时;导出的 env 优先于 .env)
uv run demo-agent run "你好" # 跑通!
✅ 期望:success=True · XXms + 一句 LLM 回复。没填 key 也能跑——走 echo 兜底打 (echo)你好,也算跑通。
from ai_trust_toolkit.llm import get_llm,成本/观测自动接、测试可换 FakeLLM、换厂商不动代码。"没 key 走 echo 兜底"就是这层抽象的好处:底座帮你处理了"没配 key"的降级,你的业务代码不用管。key 哪来:用你自己的(业务方申请),别用别人的、别进 git。👶 小白:monorepo 里改 toolkit 立即生效,我这独立仓拿的是发布版——那平台底座升级了,我怎么跟上?
👨🏫 老师:这正是两种形态的核心差别。小区内(monorepo)是 editable 装、改承重构件全楼即时生效;你这间小屋外接的是物业发布到 Nexus 的成品构件,升级 = 把 ai-trust-toolkit-bom[core]==0.6.0 改成新版本号、重跑 uv sync 订新货即可。好处是你不会被平台内部的半成品改动波及——只在你主动升版本时才动,稳定可控。代价是要手动跟版本,不像小区内那样自动同步。
步骤 4 · 写你的业务图(核心)
模板已是分层骨架(跟 Day 05 讲的标准 agent 结构一致):节点在 nodes/,builder.py 只组装,共享字段在 state.py。别对着示意图脑补——直接读模板里的真代码。先看它的黑板 examples/standalone-agent-template/your_agent/state.py,薄得只有两个字段:
class AgentState(TypedDict, total=False): # state.py
input: str
output: str
# ← 加你的中间字段,例如:parsed: dict / findings: list / verdict: str
再看示例节点 your_agent/nodes/run.py——这是模板里的完整真实代码(不是简化版),它悄悄替你处理了"没配 key"的情况:
# your_agent/nodes/run.py —— 示例节点(完整)
def run_node(state: AgentState) -> AgentState:
text = state.get("input", "")
try:
from ai_trust_toolkit.llm import get_llm # ← 走 toolkit 注入点
from langchain_core.messages import HumanMessage, SystemMessage
resp = get_llm("sonnet").invoke(
[SystemMessage(content=RUN_SYSTEM), HumanMessage(content=text)])
out = getattr(resp, "content", str(resp))
if isinstance(out, list): # content block list 兼容
out = next((b.get("text", "") for b in out if isinstance(b, dict)), str(out))
except Exception: # noqa: BLE001 — 无 key / 无依赖 · 回显
out = f"(echo · 未接 LLM){text}"
return {"output": out}
input。② get_llm("sonnet")——回扣 Day 11 铁律:业务代码永远不 import anthropic,只走 toolkit 的 get_llm,成本/观测/FakeLLM 全自动接。③ 那个 if isinstance(out, list) 是处理"新版 Anthropic 返回 content block 列表"的兼容层。④ 最关键——整段包在 try/except 里,没 key 或缺依赖时不报错,回显 (echo)你好。这就是为什么 L05 说"没填 key 也能跑通":降级逻辑就写在这。最后看组装图 your_agent/builder.py——单节点 demo,就是 Day 03 那套 StateGraph → add_node → set_entry_point → add_edge → compile:
def build_graph() -> Any: # builder.py
from langgraph.graph import END, StateGraph
g = StateGraph(AgentState)
g.add_node("run", run_node)
g.set_entry_point("run")
g.add_edge("run", END)
return g.compile()
多步流程(scan→处理→verify→report):每个节点一个 nodes/*.py,在 builder.py 里多 add_node + add_edge 串起来,节点间传的字段加到 state.py。跟 monorepo 内的 agent 一模一样——学会 Day 05 那户,这里零成本迁移。
skill 给了几条"要更复杂时怎么长"的指引,正好回扣前面学的模块:
| 需求 | 怎么做(对应 gov-agents 教程) |
|---|---|
| 多视角并行 LLM 分析 | 加 specialists/(Day 05 专家工厂) |
| 调下游系统(库/工单/交易所) | 写 tools/ 只读 client:白名单+时间窗+超时、失败 fallback 不阻断、入 LLM 前脱敏(Day 08 闸门 + Day 11 safe_tool_result) |
| 按任务路由 / 串多 agent | supervisor 编排:进程内多专家 或 跨 agent HandoffSignal(Day 04);现成编排器 supervisor-router-agent |
| agent 要存自己的数据 | 起"自有库":Memory + MySQL 双 backend、get_store() 单例、CREATE TABLE IF NOT EXISTS 幂等(Day 09 记忆思想) |
| 长 prompt | 抽到 prompts/(Python 常量,Day 05 版本化 prompt) |
governance_store.py(自有治理项库)+ 日志专家查 Quickwit(tools/quickwit_client.py)。步骤 5-7 · telemetry / 形态 / 测
步骤 5 · 接 telemetry(让平台看见你的效果)
模板的 your_agent/core.py 里已经写好了"跑图 + 上报"的整套包装(run_agent),你只需把效果 key 换成自己的。先读它的真实核心:
# your_agent/core.py —— 形态无关的 invoke 包装
def run_agent(initial_state, *, form="cli", telemetry=True, ...) -> RunResult:
start = time.monotonic()
graph = (graph_factory or build_graph)()
final = graph.invoke(initial_state, config={"recursion_limit": 50})
success = bool(final.get("output")) # ← 换成你的"成功"判定
result = RunResult(success=success, status="ok" if success else "failsafe",
duration_ms=int((time.monotonic()-start)*1000), final_state=final)
if telemetry:
_report(result, form)
return result
def _report(result: RunResult, form: str) -> None:
"""上报 portal KPI · best-effort · 失败静默(toolkit 缺失/网络都不抛)。"""
try:
from ai_trust_toolkit import bind_effect, record_effect
from ai_trust_toolkit.portal_telemetry import push_invoke_metrics
with bind_effect():
if result.success:
record_effect("done", 1) # ← 换成你的领域效果 key(upgraded / scanned)
push_invoke_metrics(result, form=form)
except Exception: # noqa: BLE001 — telemetry 不阻断业务
pass
run_agent 跑完图,把耗时/成功与否打包成一个 RunResult dataclass。② _report 里 with bind_effect(): 开一个"效果累加器",record_effect("done", 1) 记一笔领域效果。③ push_invoke_metrics(result, form) 把这次调用连同效果推给平台。④ 整个 _report 包在 try/except Exception: pass 里——这是本篇最该记住的一行:telemetry 挂了(没装 toolkit、网络断、平台宕),你的业务照跑,一声不吭。那 record_effect 到底怎么工作?翻 toolkit 真源码 ai_trust_toolkit/effect.py:50:
_current_effect: ContextVar[dict|None] = ContextVar("...", default=None) # effect.py:29
def record_effect(key: str, n: float = 1) -> None: # effect.py:50
acc = _current_effect.get()
if acc is None:
return # ← AC-1.4 · 没 bind_effect / 非 invoke 路径 · 静默 no-op 不抛
if not isinstance(key, str) or not key:
return
acc[key] = acc.get(key, 0.0) + float(n) # 同 invoke 内多次调累加
_current_effect 是一个 ContextVar(任务本地变量)——bind_effect() 那个 with 块进去时给它塞个空 dict,块内不管在哪个节点、哪个并发子任务里调 record_effect,都累加到同一个 dict(LangGraph 并发节点会 copy 上下文,所以并发也正确,跟 Day 11 的成本记账 _current_tracker 是同一套范式)。关键那句 if acc is None: return:如果你在非 invoke 路径(比如单元测试、脚本里)误调 record_effect,它不会炸,直接静默 no-op。这保证了"埋点代码放哪都安全"。再看上报函数 ai_trust_toolkit/portal_telemetry.py:100 的两处硬约束:
DEFAULT_PORTAL_URL = "https://gov-agents-platform.bmaicsaws-prod.com/portal/v1" # :34
_TIMEOUT_S = 2 # AC-TELEMETRY.9 · 不可改 · 防 portal 慢拖业务 # :35
def push_invoke_metrics(result, form, ...) -> None: # portal_telemetry.py:100
payload = { "agent_id": agent_id, "deploy_form": form,
"duration_ms": getattr(result, "duration_ms", 0), # ← 全走 getattr · 缺字段给默认
"success": getattr(result, "success", False),
"metrics": _collect_effect_metrics(result), ... }
try:
urllib.request.urlopen(req, timeout=_TIMEOUT_S) # 2s 超时
except (urllib.error.URLError, OSError, TimeoutError) as e:
log.debug("telemetry push failed (silent): %s", e) # ← AC-TELEMETRY.8 · 静默
_TIMEOUT_S = 2 且标"不可改",就是给"给别人看的用量表"设一个死线——2 秒推不上去就放弃,绝不拖累"你自己的活"。次要功能永远不能拖累主要功能。getattr(result, "x", default) 取字段,而不是直接 result.x?
因为 push_invoke_metrics 要能吃任何形状的 result——你的 RunResult、bmc-agent 的 result、甚至一个普通 dict-like。用 getattr(..., 默认值),某个 agent 没有 duration_ms 字段也不会崩,给个默认值继续。这就是"鸭子类型 + 优雅缺省":上报函数不强求调用者的类型,谁都能用、老 agent 加了新字段也向后兼容(回扣 Day 20 那条 effect_metrics proposal 里说的 "老 agent 无字段走 None · BC")。_collect_effect_metrics(portal_telemetry.py:72)合并两个来源:result.effect_metrics(兜底)和 ContextVar 累加值(主)。源码里明确 ContextVar 优先覆盖 result,且两路都空时返回 None 而不是空 dict(return merged or None)——空 dict 和 None 在下游语义不同,这个细节避免了"上报一个空 metrics 让平台以为有数据但都是 0"。✅ 期望:跑完后平台 /agents/demo-agent 几分钟内出现一条调用 + 你的效果数。本地调试不想上报加 --no-telemetry(CLI 里 run_agent(..., telemetry=not no_telemetry),见 your_agent/cli.py)。这正是 gov-agents Day 11 讲的 record_effect/ContextVar 自动采集。
步骤 6 · 选额外入口形态(按需,多选叠加)
先 CLI 跑通业务,形态后加不返工(模板都预留好了):
| 你的场景 | 选 |
|---|---|
| 自己/小范围本机用,数据敏感不出门 | 只 CLI(默认) |
| 让队友在 Cursor / Claude Code 里一键调 | CLI + MCP(起个 demo-agent-mcp 包,照 bmc-agent-mcp) |
| 被系统/定时/批量/跨团队调用,要常驻端点 | CLI + pod(加 fastapi/uvicorn,server.py 已有 build_router()) |
步骤 7 · 测
uv run --extra test pytest # 模板自带冒烟测试
--extra test。uv sync 默认不装 test 依赖,直接 uv run pytest 会报 Failed to spawn: pytest。✅ 期望:绿(2 passed)。步骤 8 · 安全自检(涉权必做)
你的 agent 用服务凭据调下游(GitLab/工单/文档/业务库)时,有个核心风险——越权放大:没权限的人经你的 agent 就拿到了那个权限。命中下面任一 → 必须走独立安全评审:
逐条答(粘进你的方案/提案):
会放大权限吗?没权限的人经 agent 能做到本来做不了的事?
下游带谁的身份?用服务凭据则怎么校验调用者有权?
服务凭据是最小权限吗?(只读/限仓库·范围)
高危操作有 dry-run / 人审 / 二次授权吗?
每次操作可追溯到真实用户吗?
越权/失败默认拒绝吗?
敏感数据出域/过 LLM 吗?
🔒 一条原则记死:经 agent 的有效权限 ≤ 调用者本人。
步骤 9 · 上线(先测试环境,验过再生产)
发布产物按形态分两类,都先测试环境、验过再生产:
9.1 打产物
# CLI / MCP:build wheel + 发 Nexus
uv build
uv publish --publish-url http://nexus.bitmartpro.com/repository/bm-pypi/ \
--username '<NEXUS_USER>' --password '<NEXUS_TOKEN>' dist/*
# pod 形态:build 镜像 → 推镜像仓
--publish-url 和凭据必须跟 uv publish 同一条命令。分开写环境变量会漏 → 误发公共 PyPI 报 403。写凭据(不是步骤 2 那个只读账号)找 @Valiant。9.2 → 9.3 测试环境 → 生产
测试环境(先)
发布系统 obelisk(obelisk.bmaiaws-infra.com/home/auto-deploy)选你的 project + 测试环境建发布单。pod 验健康检查 + 调一次 + 看平台出 invocation;CLI/MCP 让小范围工程师 pipx install 试。⚠️ telemetry 指向测试环境平台,别混进生产 KPI。
生产环境(测试通过后)
pod:obelisk 同 project 选生产环境 → 灰度 canary 推进 0→10→50→100%。CLI/MCP:稳定版广播,工程师 pipx install/upgrade。✅ 上线后跑一次 → 生产平台出真实 KPI + 业务效果。
小结 + 一页检查清单
🧠 本篇你应该能回答
standalone-repo-agentskill 解决什么场景?(无 monorepo 权限,在自己仓用 Nexus BOM 起本地 agent)- 它和
agent-onboarding、agent-creator的分工?(独立仓 / monorepo 内 / 一键生成) - 三大坑:
uv sync 401(netrc)、pytest要--extra test、uv publish参数同命令行。 - 为什么涉权要过 7 问安全自检?核心原则?(有效权限 ≤ 调用者本人)
- 上线两条链(Nexus 发包 / obelisk 出镜像)+ 先测试后生产灰度。
✅ 一页检查清单(照抄跑通)
[ ] 步0:python3 --version ≥3.10 + uv --version 有
[ ] 步1-2:下模板 → 改名干净(grep -r your_agent . 无输出)→ 配 ~/.netrc(pypi_pull)
[ ] 步3:uv sync → 填 ANTHROPIC_API_KEY → uv run demo-agent run "你好" 跑通
[ ] 步4:业务图换成你的(nodes/ + builder.py)· LLM 走 get_llm()
[ ] 步5:telemetry 接上(record_effect + push_invoke_metrics)
[ ] 步6:按场景选形态(CLI 默认 / +MCP / +pod)
[ ] 步7:uv run --extra test pytest 绿
[ ] 步8:涉权 → 过 7 问安全自检 / 独立评审
[ ] 步9:先 obelisk 测试环境验过 → 再生产(灰度 / pipx 广播)