BoundFlow 实战教程：给你的 AI Agent 装上运营控制面板

当你的 AI 编码 Agent（Claude Code、Codex）从”试玩”进入”无人值守”模式时，你会面临一系列现实问题：Agent 是不是在无限循环？它已经花了多少钱？它会不会执行一个不可逆的操作？

传统方案是在 Agent 代码中手动加各种检查——但每次增加一个 Agent 就要重复一遍，且代码逻辑和策略逻辑混在一起，修改策略等于重新部署。

BoundFlow 正是为了解决这个问题——一个开源的 AI Agent 控制面板，让你用声明式策略（而非硬编码）来管理 Agent 的费用上限、模型切换、人工审批门和自动回滚。

BoundFlow 是什么

BoundFlow 不是一个提示词框架、推理服务或 Agent 构建器——它是运行在 Agent 之外的运营层。简单说，你的 Agent 负责”思考和执行”，BoundFlow 负责”监督和干预”。

核心能力一览：

核心架构

BoundFlow 采用控制面-数据面分离架构。你的 Worker（运行 Agent 的进程）通过 gRPC 连接到 BoundFlow 控制面：

  你的 SDK/客户端 ──── gRPC ────▶  BoundFlow 后端（控制面）
                                 调度 · 分派 · 治理 · 审计
  你的 Worker ───── gRPC 流 ◀───▶
  运行 Agent+工具
  用自己的 API Key

控制面负责调度工作流、执行策略、记录审计日志；Worker 用你自己的 API Key 实际运行 Agent，控制面从不接触你的密钥。

快速安装

启动后端

echo "BOUNDFLOW_DB_PASSWORD=$(openssl rand -hex 16)" > .env

docker compose -f docker-compose.dist.yml up -d

docker compose -f docker-compose.dist.yml run --rm server -mode=provision -name=me
export BOUNDFLOW_API_KEY="<生成的 Key>"

安装 SDK

pip install boundflow
export ANTHROPIC_API_KEY="<你的 Anthropic Key>"

实战：构建带审批门的客服工作流

下面构建一个真实的客服工单处理工作流——Agent 先分析工单，判断是否需要退款；需要时停在审批门前等人工批准，且整个工作流的费用上限为每轮 $0.25。

第一步：定义 Agent 和工作流

from boundflow import (
    AgentDefinition, BoundFlowWorker, Complete,
    AwaitApproval, Next, ControlPlaneClient,
)
from boundflow.anthropic_client import AnthropicLlmClient

worker = BoundFlowWorker(
    llm=AnthropicLlmClient(...),  # 从环境变量读取端点 + 密钥
)

@worker.workflow("triage", version=1)
async def triage(ctx):
    ctx.add_context("ticket", "客户反馈账单多扣了 $50")
    await ctx.run_agent(AgentDefinition(
        name="analyst",
        model="claude-haiku-4-5",
        system_prompt="分析工单，判断是否需要退款。",
    ))
    return Complete()

第二步：添加审批门

当 Agent 判断需要退款时，工作流停在审批门前，等待人工确认：

@worker.workflow("refund", version=1)
async def refund(ctx):
    await ctx.run_agent(analyst)        # 步骤 1：分析工单

    return AwaitApproval(               # 停下来等待审批
        on_approve=Next("issue_refund", ctx.context),
        on_reject=Complete(),
        justification="批准这笔 $50 退款？",
    )

@worker.operation("refund", "issue_refund")
async def issue_refund(ctx):
    # 只有在人工批准后才会执行
    ...
    return Complete()

工作流在执行 AwaitApproval 时处于暂停状态——没有不可逆的操作已经在进行，确保安全性。

第三步：应用运行时策略

通过控制面 API 设置三层策略：

from boundflow import (
    RuntimePolicy, AgentRule, AgentMetric, Op, SetModel,
    WorkflowRule, WorkflowMetric, SetVersion,
)

cp = ControlPlaneClient(...)

await cp.set_agent_runtime_policy(
    wf_id, "analyst",
    RuntimePolicy(max_cost_usd=0.25),
)

await cp.set_agent_lifecycle_policy(wf_id, "analyst", [
    AgentRule(
        metric=AgentMetric.COST_USD,
        op=Op.GT, threshold=0.20, window=5,
        action=SetModel(value="claude-haiku-4-5"),
    ),
])

await cp.set_workflow_lifecycle_policy(wf_id, [
    WorkflowRule(
        metric=WorkflowMetric.NUM_FAILURES,
        threshold=3,
        action=SetVersion(target=1),
    ),
])

三层策略叠加的效果是：Agent 运行费用超过 $0.20 时自动降级到 Haiku，超过 $0.25 时直接终止；连续失败 3 次时整个工作流回滚到 v1 版本。

第四步：观察与审计

BoundFlow 所有运行数据通过 OpenTelemetry 导出——对接 Jaeger、Tempo、Langfuse 等任意兼容后端：

from boundflow.trace import OTelTraceSink

worker = BoundFlowWorker(
    llm=...,
    trace_sink=OTelTraceSink(tracer),
)

审批记录存储在服务端的持久化审计日志中，可通过 API 按审批 ID 查询：

records = await cp.get_approval_audit(approval_id="...")

CLI 命令快捷管理：

boundflow workflow list           # 查看所有工作流及状态
boundflow workflow runs       # 查看运行记录和结果

更多集成：LangChain 和 LangGraph

BoundFlow 的 SDK 内置了 LangChain 和 LangGraph 的支持：

from langchain_anthropic import ChatAnthropic
from boundflow.langchain_client import LangChainLlmClient

worker = BoundFlowWorker(
    llm=LangChainLlmClient(ChatAnthropic(model="claude-haiku-4-5")),
)

安装：pip install "boundflow[langchain]"

总结

BoundFlow 的价值在于将”策略”从”代码”中分离出来：

• 运行时策略：费用上限、调用次数、Token 限制——硬性保障
• 生命周期策略：基于历史指标自动切换模型或版本
• 人工审批：敏感操作强制门控，审批记录可追溯
• 自愈能力：失败次数超阈值时自动回滚或暂停

部署方式灵活——可以 Docker 自托管控制面，也可以用 BoundFlow Cloud（早期访问阶段）。无论哪种方式，你的 API Key 和推理流量始终在你自己的环境中，控制面只负责调度和治理。

对于运行多个 Agent 工作流的团队来说，BoundFlow 提供了一个比在各 Agent 中硬编码策略更优雅、更可维护的选择。

相关链接：

• BoundFlow GitHub 仓库
• BoundFlow 文档
• BoundFlow 快速上手指南