先搭最小 Agent 回路——NeoHorse-1 别急着接生产流量

想把 NeoHorse-1 接进现有系统，第一步不是改主流程，而是在主链路之外单独跑一条最小 Agent 回路：固定输入、只读工具、可观测输出。只要这条回路能在旁路里稳定完成一次「读数据—推理—给结论」，再讨论要不要接生产；一旦发现结果不稳、延迟不可控或必须依赖写操作，直接停用即可，线上数据和主业务响应都不受影响。

先做旁路验证：NeoHorse-1 跑在独立进程或容器里，数据源只给只读副本或离线快照，工具清单里不放任何写操作。用固定输入输出契约跑样本任务，记录成功率、平均轮次、超时次数和人工复核差异。任一指标变差或出现输出校验失败，先用开关停用回路并转人工接管，主业务不受影响。

把 NeoHorse-1 放在旁路容器或独立进程

隔离的核心是两件事：进程不共享主业务的生命周期，凭证不共享主业务的写权限。旁路进程可以放在独立的容器、独立的 systemd 服务或一台开发机上，网络策略上只允许它出站到模型服务地址，不允许它连生产主库。目录按「代码 / 样本 / 输出 / 日志」分开，输出目录是 append-only 的，方便对比和回放。

agent-sandbox/
├── app/
│   ├── main.py        # 回路入口，只读
│   ├── contract.py    # 输入输出校验
│   ├── tools.py       # 只读工具集合
│   └── config.py
├── data/
│   ├── samples/       # 离线样本输入
│   └── out/           # 结果落盘，与生产表隔离
├── logs/
└── .env.example

环境变量只写占位，密钥通过本地环境或密钥管理注入，不要写进仓库。数据源配置指向只读账号、只读副本或离线快照，旁路进程里不出现任何写权限凭证。

NEOHORSE_BASE_URL=
NEOHORSE_API_KEY=
NEOHORSE_MODEL=NeoHorse-1
AGENT_ENABLED=false
AGENT_DRY_RUN=true
AGENT_MAX_TURNS=4
AGENT_TIMEOUT_SECONDS=30
READONLY_DB_DSN=      # 只读账号，指向只读副本或快照库

验证方式很直接：启动旁路进程后确认它连不上主库写入，执行一次样本任务后确认生产表没有新增记录。如果做不到这一点，先别往下走。

定义最小任务输入输出契约

契约的作用是让每次验证都能对齐比较：同样的输入结构、同样的输出字段，换模型版本或换提示词时才看得出差异。输入里把工具白名单和轮次上限固化，避免回路自己扩权。

{
  "task_id": "demo-0001",
  "task_type": "read_and_judge",
  "question": "把这里替换成实际要判断的问题",
  "context_ref": {
    "source": "readonly_snapshot",
    "table": "orders_snapshot",
    "filter": {"status": "pending", "limit": 20}
  },
  "constraints": {
    "max_turns": 4,
    "allowed_tools": ["query_rows", "calc", "compare"]
  }
}

输出字段建议固定为下面这些，status 用枚举值，校验不通过就按失败处理，不要让它带着半成品继续往下走。

{
  "task_id": "demo-0001",
  "status": "ok",
  "answer": "最终结论文本",
  "evidence": [{"tool": "query_rows", "arguments": {}, "row_count": 12}],
  "turns": 3,
  "elapsed_ms": 4210,
  "model": "NeoHorse-1",
  "warnings": []
}

{
  "task_id": "demo-0001",
  "status": "error",
  "error_code": "TOOL_TIMEOUT",
  "message": "query_rows 超过单次工具超时",
  "turns": 2,
  "partial_evidence": []
}

status 只用 ok / partial / error 三个值；partial 表示结论可用但证据不全，需要人工复核。字段名按实际任务替换，但结构不要频繁改，否则灰度数据没法对比。

写一个只读工具调用的 Agent 循环

循环骨架的重点不是模型调用本身，而是轮次上限、工具白名单和输出校验这三道闸。模型接入那一行留成占位，替换成团队实际使用的客户端即可，只传只读工具声明，不传写工具。

# app/main.py —— 旁路只读回路骨架
import json, os, time
from contract import validate_input, validate_output, error_payload
from tools import TOOLS, READONLY_SPECS

MAX_TURNS = int(os.getenv("AGENT_MAX_TURNS", "4"))
TIMEOUT = int(os.getenv("AGENT_TIMEOUT_SECONDS", "30"))

def call_model(messages, tool_specs):
    """替换为实际模型调用：只传消息 + 只读工具声明。"""
    raise NotImplementedError

def run_agent(task: dict) -> dict:
    validate_input(task)
    messages = [
        {"role": "system", "content": "只能用只读工具，不得假设已写入任何数据。"},
        {"role": "user", "content": json.dumps(task, ensure_ascii=False)},
    ]
    evidence, started = [], time.time()

    for turn in range(1, MAX_TURNS + 1):
        if time.time() - started > TIMEOUT:
            return error_payload(task, "GLOBAL_TIMEOUT", "整体超时", turn, evidence)

        reply = call_model(messages, READONLY_SPECS)
        messages.append(reply)

        if reply.get("type") == "final":
            out = {"task_id": task["task_id"], "status": "ok",
                   "answer": reply["answer"], "evidence": evidence,
                   "turns": turn,
                   "elapsed_ms": int((time.time() - started) * 1000)}
            validate_output(out)          # 字段缺失或枚举非法就抛错
            return out

        call = reply["tool_call"]
        tool = TOOLS.get(call["name"])
        if tool is None or not tool.readonly:
            return error_payload(task, "TOOL_NOT_ALLOWED", call["name"], turn, evidence)

        result = tool.run(**call.get("arguments", {}))
        evidence.append({"tool": call["name"],
                         "arguments": call["arguments"],
                         "row_count": result.get("row_count")})
        messages.append({"role": "tool", "name": call["name"],
                         "content": json.dumps(result, ensure_ascii=False)})

    return error_payload(task, "MAX_TURNS_EXCEEDED", "超过轮次上限", MAX_TURNS, evidence)

工具侧同样要收口：表名走白名单，limit 强制上限，查询参数化，连接用只读账号。这几行不写，回路就只是换了地方读生产库。

# app/tools.py
READONLY_TABLES = {"orders_snapshot", "tickets_snapshot"}

def query_rows(table: str, filt: dict, limit: int = 20) -> dict:
    if table not in READONLY_TABLES:
        return {"error": "TABLE_NOT_ALLOWED", "row_count": 0}
    limit = min(int(limit), 50)          # 强制上限，避免整表拉取
    rows = read_only_conn.execute(table, filt, limit)  # 只读副本上的参数化查询
    return {"rows": rows, "row_count": len(rows)}

记录灰度对照指标

每跑完一个任务就落一行 JSON 日志，字段跟输出契约对齐，后面统计不用再解析自然语言。建议先看趋势，再结合环境确认是否需要调参或停用。

{"ts":"<iso8601>","task_id":"demo-0001","status":"ok","turns":3,
 "elapsed_ms":4210,"tools":["query_rows"],"review":"same"}

成功率：统计 status=ok 占全部任务的比例，partial 单独计数，不要混进成功。
平均轮次：从 turns 字段取平均，重点看有没有任务长期贴着 max_turns。
超时次数：按 error_code 里的 GLOBAL_TIMEOUT、TOOL_TIMEOUT 分类计数。
人工复核差异：抽样人工看结论，标记 same / different / unclear 三类，different 的任务把 task_id 单独留档。

计数方式用命令行就够了，例如 grep 数状态行、awk 求和 turns 和 elapsed_ms。这里的阈值应由团队结合自己的样本量和业务容忍度来定，不要照搬别人的数字；判断依据是这批样本里差异集中在哪些任务类型，而不是某个绝对百分比。

设置回滚开关与人工接管点

开关必须是外置配置，改完能立刻生效，不需要重新发版。旁路进程默认关闭，每次验证手动打开，验证结束再关掉。

# config/agent.yaml —— 旁路进程启动时读取
agent:
  enabled: false            # 总开关，默认关闭
  dry_run: true             # 只生成结论，不触发任何下游动作
  max_turns: 4
  timeout_seconds: 30
  allowed_tools: ["query_rows", "calc", "compare"]
  fallback: human_queue     # 停用后任务转到人工队列，不静默丢弃

人工接管点建议至少覆盖这几种情况：同一个任务连续出现 error 或超时；输出校验失败，比如 status 枚举非法、evidence 为空却返回 ok；人工复核中 different 的任务集中在同一类问题上；单次任务耗时明显偏离这批样本的常见区间；工具返回空数据或连接异常。触发后先关 enabled，再按 fallback 把待处理任务转人工，不做自动重试。

停止后数据保留方式要提前定好：输入消息、模型回复、工具调用参数与返回、最终输出、日志行都按 task_id 写进 data/out 和 logs/，只追加不覆盖，不回写任何生产表。这样回滚之后还能离线复盘，也方便下次打开开关时用同一批样本重跑对比。