feat(tier2): B门重机器并行落地——控制面有界resume + Studio观测 + MCP工具面 + 基线client/Prompt门

B 门(AgentScope 重机器)四单元并行建设,文件不相交,均过 6c6g 静态校验(py_compile +
惰性 import 安全 + 机器门)。真跑收敛/Studio/门验证属 mini-desktop,作为下一步 e2e。

B1 控制面有界 resume(service/control_plane.py · 皇冠):
  补上 service/app.py 与 bootstrap.py 文件头标的 followup——CLI run_studio:404-458 的外层有界
  resume(M3 看一次 verdict 就停 → 带反馈踹回去续修)此前在服务态(create_app :8200)缺失。
  drive_generation() 复用 bootstrap REST 原语 + run 纯代码门:每回合消费 SSE(REPLY_END/
  EXCEED_MAX_ITERS)等 chat run 结束 → 独立跑 run.run_gates 机器判门 → 门绿据 on-disk 源工程
  重建七要素 + persist 落库 / 否则 verdict_feedback 续修,直到门绿/预算耗尽/熔断。不靠内存
  session 判收敛(服务态九工具每回合工厂新建 Tier2Session),贴合「门机器判、控制面绝不自评翻绿」。

C1 AgentScope Studio 观测(observability/studio_sink.py + trace.py):
  把统一 trace 事件按 OTLP(经核实 Studio = OpenTelemetry,非私有 REST;HTTP 3000 /v1/traces)
  映射成 span 推 Studio,traceId→gen_ai.conversation.id 聚成一条 run。私有 TracerProvider 避免
  抢 otel 全局单例。默认关闭(无 TIER2_STUDIO_URL 即 no-op)、observe-only、绝不改 decision。

C2 MCP 工具面(worker/mcp_tools.py + toolkit.py 加性挂点):
  图说 E 族——build_toolkit 末尾加配置门控挂点,默认关闭时 build_mcp_clients() 返回 []、
  Toolkit(tools, mcps=[]) 与原构造等价,九工具(spike 已验证主链)行为零改。Toolkit 接受 mcps=
  形参已对源码核实(tool/_toolkit.py:93)。

C3 基线 client + Prompt 一致性机器门(config.py + genconfig.py + generation.yaml + ci/):
  build_baseline_model(Opus 经 new-api Anthropic 原生)供 A 门「证路」,build_model M3 路零回归;
  check-prompt-registry.sh 文件级三向一致 + import 级命中 registry,6c6g 实跑 exit 0。

修复 design_team 配置外置兑现(design_team.py):
  leader_max_iters/per_expert_cap/timeout_s 改 None→genconfig.get('design_team',...),此前硬编码
  默认 240 导致改 generation.yaml 的 timeout 不生效;已验 genconfig 真读到 YAML 值。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
zizi 2026-06-24 07:46:59 +00:00
parent 1a4c6970b0
commit bd8b2c80a6
11 changed files with 1471 additions and 7 deletions

178
tier2/ci/check-prompt-registry.sh Executable file
View File

@ -0,0 +1,178 @@
#!/usr/bin/env bash
# tier2 prompt 一致性守门 —— 把「prompts.py 该加载的每个 prompt == registry/外置文件、无漂移」编译成机器门
# (对齐项目 AGENTS.md §6 条款10:文档治理规则必须编译成机器门,违规红线拦截,而非只作散文)。
#
# 背景:tier2 的 8 条 system prompt 已外置到 contracts/prompts/09-tier2-richgame/*.md + registry.yaml,
# worker/prompts.py 经契约 #8 Prompt Registry 运行时按 id 读正文,读不到/脏 → 回落 roles.py 内置原文。
# 外置后最容易出的漂移:① 改了 registry 加了新 id,roles.py 没接(孤儿条目);② roles.py 加载某 id,
# registry 没登记(回落内置原文、外置形同虚设);③ registry 的 file 指向的 .md 不存在(运行时静默回落)。
# 本门把这三类漂移在 wave-close / pre-commit 拦下来。
#
# 校验内容(纯文件级三向一致性比对;不需 runtime/网络/重依赖,6c6g 本机与 CMI(mini-desktop)均可跑):
# A) roles.py 里 prompts.load("tier2.*") 声明加载的 id 集合
# == registry.yaml 里 file 指向 09-tier2-richgame/ 的 tier2.* 条目 id 集合(双向无差集);
# B) registry 每个 tier2.* 条目的 file 字段所指 .md 在 contracts/prompts/ 下真实存在;
# C) contracts/prompts/09-tier2-richgame/ 下每个 .md 都被某 registry 条目引用(无未登记的孤儿 .md)。
# 任一项漂移 → 非零退出 + 指出哪个 prompt / 哪个文件漂了;全一致 → 退出 0。
#
# ⚠️ import 级校验(bare-import worker/prompts.py + 逐 id load() 断言「命中 registry 非回落内置原文、
# 且正文与 .md 逐字节一致」)是更强的一层:它能抓出「frontmatter 剥离口径变了 / registry 解析器口径变了」
# 这类纯文件比对看不出的语义漂移。worker/prompts.py 顶层不牵 agentscope(只 os/re/threading/pathlib),
# 故 6c6g 本机可裸 import;本门【尽力】跑这层(python3 可用且 prompts.py 可裸 import 时),跑不了就
# 降级为纯文件级比对并打印「import 级校验需在能 import worker 包的环境(mini-desktop)补跑」的提示——
# 纯文件级比对仍是硬门(漂移即拦),import 级只是增强、缺它不放过文件级漂移。
set -euo pipefail
# ROOT = tier2/(本脚本在 tier2/ci/);与 check-forbidden-import.sh 同款定位。
ROOT="$(cd "$(dirname "$0")/.." && pwd)"
# 仓库根 = tier2/ 的上一级(contracts/ 在仓根下)。
REPO_ROOT="$(cd "$ROOT/.." && pwd)"
ROLES_PY="$ROOT/gen-worker/worker/roles.py"
PROMPTS_PY="$ROOT/gen-worker/worker/prompts.py"
REGISTRY="$REPO_ROOT/contracts/prompts/registry.yaml"
PROMPT_DIR="$REPO_ROOT/contracts/prompts/09-tier2-richgame"
fail=0
# 前置:三个关键路径必须存在(缺任何一个都是配置外置被破坏,直接红线)。
for p in "$ROLES_PY" "$REGISTRY" "$PROMPT_DIR"; do
if [ ! -e "$p" ]; then
echo "❌ prompt-registry: 关键路径缺失:$p" >&2
fail=1
fi
done
if [ "$fail" -ne 0 ]; then
exit 1
fi
# ── 提取三个 id/文件集合 ────────────────────────────────────────────────────────────
# (1) roles.py 声明加载的 tier2.* id(prompts.load("tier2.xxx", ...) 的第一个串参)。
roles_ids=$(grep -oE 'prompts\.load\("tier2\.[A-Za-z0-9._-]+"' "$ROLES_PY" \
| sed -E 's/.*prompts\.load\("//; s/"$//' | sort -u)
# (2) registry.yaml 里 file 指向 09-tier2-richgame/ 的条目:抽出 (id, file) 对。
# registry 是「- id: xxx」起头、随后若干「 key: val」的列表块;我们只关心 tier2.* 且 file 落在
# 09-tier2-richgame/ 的条目(其余阶段 01~07 的条目与本门无关,不碰)。
# awk:遇 `- id:` 记当前 id;遇 `file:` 记当前 file;到下一个 `- id:` 或文件尾时,若 file 在
# 09-tier2-richgame/ 下就输出「id<TAB>file」。剥行内注释(# 之后)与首尾空白。
reg_pairs=$(awk '
function trim(s){ sub(/#.*/,"",s); gsub(/^[ \t]+|[ \t]+$/,"",s); return s }
/^[[:space:]]*-[[:space:]]+id[[:space:]]*:/ {
if (cur_id != "" && cur_file ~ /^09-tier2-richgame\//) print cur_id "\t" cur_file;
line=$0; sub(/^[^:]*:[[:space:]]*/,"",line); cur_id=trim(line); cur_file=""; next
}
/^[[:space:]]+file[[:space:]]*:/ {
line=$0; sub(/^[^:]*:[[:space:]]*/,"",line); cur_file=trim(line); next
}
END { if (cur_id != "" && cur_file ~ /^09-tier2-richgame\//) print cur_id "\t" cur_file }
' "$REGISTRY")
reg_ids=$(printf '%s\n' "$reg_pairs" | awk -F'\t' 'NF>=1 && $1!=""{print $1}' | sort -u)
# (3) 09-tier2-richgame/ 下真实存在的 .md 文件名(相对 contracts/prompts/ 的路径,与 registry file 字段同口径)。
fs_files=$(cd "$REPO_ROOT/contracts/prompts" && ls 09-tier2-richgame/*.md 2>/dev/null | sort -u || true)
# ── 校验 A:roles.py 声明集 == registry tier2 条目集(双向无差集)───────────────────────
only_roles=$(comm -23 <(printf '%s\n' "$roles_ids") <(printf '%s\n' "$reg_ids") || true)
only_reg=$(comm -13 <(printf '%s\n' "$roles_ids") <(printf '%s\n' "$reg_ids") || true)
if [ -n "$only_roles" ]; then
echo "❌ prompt-registry[A]: roles.py 加载了这些 id,registry 却未登记(外置失效→运行时静默回落内置原文):" >&2
printf '%s\n' "$only_roles" | sed 's/^/ /' >&2
fail=1
fi
if [ -n "$only_reg" ]; then
echo "❌ prompt-registry[A]: registry 登记了这些 tier2 条目,roles.py 却没加载(孤儿条目):" >&2
printf '%s\n' "$only_reg" | sed 's/^/ /' >&2
fail=1
fi
# ── 校验 B:registry 每个 tier2 条目的 file 指向的 .md 真实存在 ───────────────────────────
while IFS=$'\t' read -r id file; do
[ -z "$id" ] && continue
if [ ! -f "$REPO_ROOT/contracts/prompts/$file" ]; then
echo "❌ prompt-registry[B]: registry 条目 '$id' 的 file 指向不存在的文件:$file" >&2
fail=1
fi
done <<< "$reg_pairs"
# ── 校验 C:09-tier2-richgame/ 下每个 .md 都被某 registry 条目引用(无孤儿 .md)──────────────
reg_files=$(printf '%s\n' "$reg_pairs" | awk -F'\t' 'NF>=2 && $2!=""{print $2}' | sort -u)
orphan_md=$(comm -23 <(printf '%s\n' "$fs_files") <(printf '%s\n' "$reg_files") || true)
if [ -n "$orphan_md" ]; then
echo "❌ prompt-registry[C]: 这些 .md 在 09-tier2-richgame/ 下却没被任何 registry 条目引用(孤儿文件):" >&2
printf '%s\n' "$orphan_md" | sed 's/^/ /' >&2
fail=1
fi
# ── import 级增强校验(尽力跑;跑不了优雅降级,不影响上面文件级硬门的结论)──────────────────
# bare-import worker/prompts.py(顶层不牵 agentscope),对每个 roles.py 声明的 id 调 load(id, SENTINEL):
# 返回值 != SENTINEL 即「命中 registry(非回落内置原文)」。这能抓出「文件存在但解析/剥 frontmatter
# 口径漂移导致 load 实际回落」这类纯文件比对看不出的问题。需 python3 且 prompts.py 可裸 import。
import_checked=0
if command -v python3 >/dev/null 2>&1; then
ids_csv=$(printf '%s\n' "$roles_ids" | paste -sd, -)
# PROMPTS_DIR 显式指向仓内 09 的父目录,避免受调用环境 env TIER2_PROMPTS_DIR 干扰。
import_out=$(WORKER_DIR="$ROOT/gen-worker/worker" REPO_PROMPTS="$REPO_ROOT/contracts/prompts" IDS="$ids_csv" \
python3 - <<'PY' 2>&1 || true
import os, sys, importlib.util
worker_dir = os.environ["WORKER_DIR"]
ids = [x for x in os.environ.get("IDS", "").split(",") if x]
# 直接按文件路径加载 prompts.py 模块(避免触发 worker 包的相对 import / agentscope)。
spec = importlib.util.spec_from_file_location("_tier2_prompts_probe", os.path.join(worker_dir, "prompts.py"))
mod = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(mod)
except Exception as e: # 任何 import 期异常 → 标记不可跑,降级(由 shell 侧打印降级提示)
print("PROBE_SKIP:" + type(e).__name__ + ":" + str(e))
sys.exit(0)
# 强制用仓内 prompts 目录(不受外部 env 干扰)。
os.environ["TIER2_PROMPTS_DIR"] = os.environ["REPO_PROMPTS"]
mod.reload()
SENTINEL = "<<__FALLBACK_SENTINEL__>>"
bad = []
for pid in ids:
body = mod.load(pid, SENTINEL)
if body == SENTINEL or not body:
bad.append(pid)
if bad:
print("PROBE_FALLBACK:" + ",".join(bad))
else:
print("PROBE_OK:" + str(len(ids)))
PY
)
# prompts.py 的 best-effort 告警会打到 stdout 与探针标记行混在一起;只取最后一行 PROBE_* 标记判定。
import_marker=$(printf '%s\n' "$import_out" | grep -E '^PROBE_(OK|FALLBACK|SKIP):' | tail -n1)
[ -z "$import_marker" ] && import_marker="$import_out"
case "$import_marker" in
PROBE_OK:*)
import_checked=1
echo "✓ prompt-registry[import]: ${import_out#PROBE_OK:} 条 prompt 经 prompts.load 命中 registry(非回落内置原文)"
;;
PROBE_FALLBACK:*)
echo "❌ prompt-registry[import]: 这些 id 经 prompts.load 仍回落内置原文(registry 命中失败=漂移):" >&2
echo " ${import_out#PROBE_FALLBACK:}" >&2
fail=1
;;
PROBE_SKIP:*)
echo "⚠ prompt-registry[import]: prompts.py 裸 import 失败(${import_out#PROBE_SKIP:})→ 仅做文件级比对;import 级校验请在能 import worker 包的环境(mini-desktop)补跑。"
;;
*)
echo "⚠ prompt-registry[import]: 探针输出异常,降级为文件级比对(请在 mini-desktop 补跑 import 级校验)。输出:$import_out"
;;
esac
else
echo "⚠ prompt-registry[import]: 本机无 python3 → 仅做文件级比对;import 级校验请在 mini-desktop 补跑。"
fi
if [ "$fail" -ne 0 ]; then
echo "❌ prompt-registry 守门未通过:tier2 prompt 与 registry/外置文件存在漂移(见上)。" >&2
exit 1
fi
n_ids=$(printf '%s\n' "$roles_ids" | grep -c . || true)
if [ "$import_checked" -eq 1 ]; then
echo "✓ prompt-registry 守门通过(tier2 ${n_ids} 条 prompt:文件级三向一致 + import 级命中 registry)"
else
echo "✓ prompt-registry 守门通过(tier2 ${n_ids} 条 prompt:文件级三向一致;import 级增强校验本机未跑,见上提示)"
fi

View File

@ -25,6 +25,8 @@ model:
default_model_name: MiniMax-M3 # 单写 agent 默认模型(M3 路;model_name_from_env 默认值)
deepseek_flash: deepseek-v4-flash # 便宜主力档名(spike 主问「便宜到什么程度还守得住」)
deepseek_pro: deepseek-v4-pro # 强便宜救场档名(退路树 Q1 读它的过门率)
baseline_opus: claude-opus-4-8 # 强基线 Opus(Anthropic 原生经 new-api;A 门「M3 证路(软)」主用,跑同一条 studio 线证「路本身通」)
baseline_fable: claude-fable-5 # 强基线 Fable(列出但不假设网关一定可用;build_baseline_model 取不到即诚实报错、不静默)
max_tokens: 16000 # 输出上限(非上下文窗;build_model 硬约束 max_tokens > thinking_budget)
thinking_enable: true # M3 是否开 thinking 分离(便宜档无此约束,只 M3 路用)
thinking_budget: 8000 # M3 thinking 预算(必须 < max_tokens,否则 build_model 抛)

View File

@ -34,6 +34,14 @@ from .trace import (
TraceAdapter,
TraceStep,
to_trace_step,
with_studio,
)
# Studio 观测接线(C1):把统一 trace 事件推到 AgentScope Studio 的旁路 sink 工厂。
# 注意:studio_sink 顶层零重依赖(otel SDK 在函数内惰性 import),6c6g 仍可裸 import 本包。
from .studio_sink import (
make_studio_sink,
resolve_studio_url,
StudioSink,
)
__all__ = [
@ -54,4 +62,9 @@ __all__ = [
"TraceAdapter",
"TraceStep",
"to_trace_step",
"with_studio",
# ── Studio 观测接线(C1)──
"make_studio_sink",
"resolve_studio_url",
"StudioSink",
]

View File

@ -0,0 +1,337 @@
"""observability/studio_sink.py —— tier2 富游戏自治线 · 把统一 trace 事件推到 AgentScope Studio(C1 观测接线)。
这份解决什么问题
tier2 一次生成 run Tier2TraceMiddleware / TraceAdapter.ingest 把每步事件映射成统一 trace 形状
(observability/trace.py TraceStep:traceId/step/cost/verdict/timestamp + 推理/动作/观察扩展段),
现状只落内存 adapter.steps收口由编排器读 summary本模块把这同一条事件流再接一个旁路 sink 把每步
推成 OTLP span 发给 AgentScope Studio, tier2 生成 run Studio Trace 页可视化(服务于创始人
迭代过程拿证据/日志做优化)
Studio 真实接入方式(源码/reference 核实,绝不臆造)
AgentScope Studio 的可观测是 **OpenTelemetry / OTLP 协议** (不是某私有 REST 形状):
- studio README:Tracing: OpenTelemetry-based trace visualization for LLM calls, token usage,
and agent invocations(/root/oss/agentscope-studio/README.md)
- studio tracing 教程(/root/oss/agentscope-studio/docs/tutorial/en/develop/tracing.md):Studio 默认暴露
· OTLP/Trace/gRPC:localhost:4317(env OTEL_GRPC_PORT )
· OTLP/Trace/HTTP:localhost:3000(env PORT )
Studio currently only supports receiving Trace type data.
- 该教程Advanced Integration: Import Custom Trace Data段给了任意来源接 Studio 的官方配方
OpenTelemetry Python SDK OTLPSpanExporter(endpoint=Studio HTTP)+ BatchSpanProcessor +
TracerProvider,自己 start_as_current_span span(原文示例 endpoint="http://localhost:3000")
- 接收侧实现在 studio server(/root/oss/agentscope-studio/packages/server/src/index.ts:136 提到
traces will be received via HTTP endpoint /v1/traces;OTLP HTTP exporter 自动在 base 末尾拼 /v1/traces)
AgentScope 框架原生那条(agentscope.init(studio_url=...) + middleware.TracingMiddleware)同样落到这个
OTLP 端点(middleware/_tracing/_trace.py: opentelemetry 全局 tracer span;setup init SDK provider)
本模块接的是哪条 / 为什么
tier2 已经有自己一套结构化 trace 事件(TraceStep), C1 的任务是 TraceAdapter ingest 的事件接到
Studio 所以走 **Advanced Integration 自组 span 这条**(把每条 TraceStep 映射成一个 gen-ai 语义约定 span),
而不是去替换/抢装 agentscope.init 的全局 TracerProvider(那是编排器/服务启动层的事,会与框架原生 middleware
set_tracer_provider 全局单例) sink 自带一个 **独立私有 TracerProvider**(不碰 otel 全局),
span 经独立 OTLPSpanExporter 直发 Studio HTTP,与框架原生 trace 互不干扰可单独开关
observe-only 铁律(C1 红线)
trace / Studio 是纯旁路:绝不改 verdict / decision / L1 / 任何门判定,绝不中断生成主链
- 取不到 Studio 地址 sink 直接降级为 no-op(__call__ 立即 return),只留内存 adapter,绝不报错;
- 推送失败(Studio 没起 / 网络错 / otel 依赖) best-effort 记一次告警日志后继续,绝不抛绝不阻塞;
- 本模块只TraceStep 往外发,从不回写任何决策字段
惰性 import(6c6g 红线)
opentelemetry SDK / exporter 在函数体内 import(6c6g 静态校验环境无这些 wheel),本模块顶层零重依赖
保证 6c6g py_compile 且裸 import真正建 provider / span 只在装了依赖的 mini-desktop / Mac 上发生
开关 / 地址来源
Studio HTTP 地址按以下顺序取(默认关闭 = 取不到即 no-op):
env TIER2_STUDIO_URL(单点覆盖,临时开一次实验最省事);
service/infra_config.py [studio].http_url(若该分区/项存在;infra.yaml 内网入仓口径);
显式传给 make_studio_sink(studio_url=...) 的参数
三者都没有 resolve_studio_url() 返回 None make_studio_sink 返回 no-op sink(现有 CLI 真跑不受影响)
"""
from __future__ import annotations
import os
from typing import Any, Callable, Optional
# ── env 开关键(与 genconfig/infra_config 的 env 风格一致:全大写、TIER2_ 前缀)──
# 取到非空即视为「开 Studio 观测」并用作 Studio HTTP base(如 http://100.64.0.7:3000)。
ENV_STUDIO_URL = "TIER2_STUDIO_URL"
# ── tier2 这条 trace 在 Studio 里的服务名(OTLP resource service.name;Studio 按它聚合归类)──
SERVICE_NAME = "tier2-gen-worker"
# ── 一次性告警去重(同一 tag 只 print 一次,避免长 run 刷屏;对齐 genconfig._warn_once 风格)──
_WARNED: set[str] = set()
def _warn_once(tag: str, msg: str) -> None:
"""同一 tag 的告警只落一次(best-effort 告警去重,绝不抛)。"""
if tag in _WARNED:
return
_WARNED.add(tag)
print(f"[tier2-studio-sink] {msg}", flush=True)
def resolve_studio_url(explicit: Optional[str] = None) -> Optional[str]:
"""解析 Studio HTTP base 地址:env > infra_config[studio].http_url > 显式参数;都没有 → None(=关闭)。
:param explicit: 调用方显式传入的 Studio HTTP 地址(最低优先级,作兜底)
:return: 归一化后的 Studio HTTP base(去尾斜杠);取不到任何来源 None(make_studio_sink 据此降级 no-op)
"""
# ① env 单点覆盖(最高优先级)。
env_val = os.environ.get(ENV_STUDIO_URL)
if env_val and env_val.strip():
return env_val.strip().rstrip("/")
# ② infra_config.py 的 [studio].http_url(惰性 import:infra_config 顶层无重依赖,但仍 best-effort 兜)。
try:
# 包内/直跑兼容导入(直跑 observability 下脚本时 service 包仍可解析)。
try:
from service import infra_config # type: ignore
except Exception: # pragma: no cover —— 直跑兜底:把 gen-worker/ 加进 sys.path 再取
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from service import infra_config # type: ignore
cfg_val = infra_config.get("studio", "http_url", None)
if cfg_val and str(cfg_val).strip():
return str(cfg_val).strip().rstrip("/")
except Exception as exc: # best-effort:读配置任何异常都不阻断,落一次告警后继续
_warn_once("infra_config", f"读 infra_config[studio].http_url 失败(忽略):{exc}")
# ③ 显式参数兜底。
if explicit and explicit.strip():
return explicit.strip().rstrip("/")
return None
# ── TraceStep 工具结果状态 → OTLP span 状态码的映射(observe 段 verdict 决定 span 是否标 ERROR)──
# 注意:这只是「把工具/门结果状态如实反映到 span 颜色」,不改任何生成侧判定(observe-only)。
_ERROR_VERDICTS = {"error", "interrupted", "denied"}
class StudioSink:
"""把统一 trace 事件(TraceStep dict)推成 OTLP span 发给 AgentScope Studio 的旁路 sink。
用法(插进现有 TraceAdapter sink ,零改 ingest / 事件结构):
sink = make_studio_sink(trace_id="<gen-task-trace>") # 取不到 Studio 地址 → 返回 no-op
adapter = TraceAdapter(trace_id="<gen-task-trace>", sink=sink)
async for event in agent.reply_stream(user_msg):
adapter.ingest(event) # 映射 + 落内存 + 经本 sink 推 Studio,全程 best-effort
sink.shutdown() # 收口:flush 残留 span(best-effort,不抛)
sink 契约 = Callable[[dict], None]( TraceAdapter sink 一致):吃一条 TraceStep dict 发一个 span
任何环节失败( otel 依赖 / provider 失败 / 发送失败)一律 best-effort:落一次告警自动降级 no-op绝不抛
"""
def __init__(self, trace_id: str, studio_url: str) -> None:
"""
Args:
trace_id: 本次生成的 traceId(贯穿;作为 span gen_ai.conversation.id,Studio 据它把同一 run 的步归一条 trace)
studio_url: Studio HTTP base(已归一化;OTLP HTTP exporter 会自动在末尾拼 /v1/traces)
"""
self.trace_id = trace_id
self.studio_url = studio_url
self._enabled = True # best-effort 自降级开关:首次建 provider 失败即永久置 False
self._provider: Any = None # 私有 TracerProvider(惰性建;不碰 otel 全局单例)
self._tracer: Any = None # 从私有 provider 取的 tracer
self._initialized = False # 是否已尝试过初始化(成功或失败都置 True,避免每条都重试)
def _ensure_tracer(self) -> bool:
"""惰性建私有 TracerProvider + OTLPSpanExporter(HTTP→Studio);成功返回 True,失败 best-effort 降级返回 False。
为什么用私有 provider 而非 otel 全局 set_tracer_provider: sink tier2 自己的 TraceStep 另发一份给
Studio的旁路,不应抢占 otel 全局单例(那是 agentscope.init / 框架原生 TracingMiddleware 的地盘);
私有 provider 经自带 exporter 直发,互不干扰可独立开关
"""
if self._initialized:
return self._enabled
self._initialized = True
try:
# 惰性 import:6c6g 无 opentelemetry SDK / exporter wheel,本模块仍可被裸 import(此处才触发依赖)。
from opentelemetry.sdk.trace import TracerProvider # noqa: PLC0415
from opentelemetry.sdk.trace.export import ( # noqa: PLC0415
BatchSpanProcessor,
)
from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( # noqa: PLC0415
OTLPSpanExporter,
)
from opentelemetry.sdk.resources import Resource # noqa: PLC0415
# resource.service.name = tier2-gen-worker:Studio 按服务名聚合归类这条线的 trace。
resource = Resource.create({"service.name": SERVICE_NAME})
provider = TracerProvider(resource=resource)
# OTLP/HTTP exporter:endpoint 传 Studio HTTP base,SDK 默认在末尾拼 /v1/traces(对齐 studio /v1/traces 接收)。
# 显式给 traces 全路径,避免不同 SDK 版本对 base/自动拼路径行为不一致(reference 示例用 base,
# 这里给全路径更稳;两者 studio 都收)。
exporter = OTLPSpanExporter(endpoint=self.studio_url + "/v1/traces")
# 批处理 processor:异步攒批后台发,推送不卡 ingest 主路(observe-only 不阻塞生成)。
provider.add_span_processor(BatchSpanProcessor(exporter))
# 从私有 provider 取 tracer(不经 otel.trace.get_tracer 全局,避免取到框架/默认 provider)。
self._provider = provider
self._tracer = provider.get_tracer("tier2-gen-worker", "1.0.0")
_warn_once(
"init-ok",
f"Studio 观测已开(traceId={self.trace_id}{self.studio_url}/v1/traces)",
)
return True
except Exception as exc: # best-effort:缺依赖 / 建 provider 失败 → 永久降级 no-op,不抛、不阻塞
self._enabled = False
_warn_once(
"init-fail",
f"建 OTLP→Studio 通道失败,本 run 降级为不推 Studio(只留内存 trace):{exc}",
)
return False
def __call__(self, trace_step: dict) -> None:
"""吃一条 TraceStep dict → 发一个 OTLP span 给 Studio(全程 best-effort,绝不抛、绝不阻塞主链)。
映射( TraceStep 公共核心子集 + 扩展段铺成 gen-ai 语义约定 + agentscope 扩展约定的 span 属性):
- span.name = ext.action.tool / 相位 kind / raw.event,组成可读步名( "step 3 · write_source")
- gen_ai.conversation.id = traceId(Studio 据它把同一 run 的步归一条 trace 与框架 TracingMiddleware 同键)
- agentscope.function.* / gen_ai.usage.* = ext 三段(推理/动作/观察)+ cost.tokens 铺成属性,JSON 序列化
- span 状态:observe verdict {error/interrupted/denied} ERROR(如实反映工具/门结果,不改判定)
"""
if not self._enabled:
return # 已降级 no-op(取不到地址 / 初始化失败):直接返回,observe-only 不影响主链
if not self._ensure_tracer():
return # 首次初始化失败已落告警并永久降级
try:
self._emit_span(trace_step)
except Exception as exc: # best-effort:单条 span 发送失败只告警一次,不连累后续、不抛
_warn_once(
"emit-fail",
f"推 Studio span 失败(best-effort 计告警,traceId={self.trace_id}):{exc}",
)
def _emit_span(self, trace_step: dict) -> None:
"""把单条 TraceStep 组成一个 span 并即时收尾(同步 start→set→end,交 BatchSpanProcessor 异步发出)。"""
import json # noqa: PLC0415 —— 仅序列化属性用,函数内 import 无妨
from opentelemetry.trace import StatusCode # noqa: PLC0415
step = trace_step.get("step")
ext = trace_step.get("ext") or {}
raw = ext.get("raw") or {}
# ── span 名:优先工具名 / 相位 kind / 原始事件类型,拼上步序,便于在 Studio 列表里一眼认出这一步在干嘛 ──
label = (
(ext.get("action") or {}).get("tool")
or (ext.get("action") or {}).get("kind")
or (ext.get("observation") or {}).get("kind")
or (ext.get("reasoning") or {}).get("kind")
or raw.get("event")
or "step"
)
span_name = f"step {step} · {label}"
# 用私有 tracer 起 span(end_on_exit 默认 True:with 退出即 end;本步是「一条已成事实的 trace 记录」,
# 无需跨步嵌套,即起即收最简单、也不会因异常漏 end)。
with self._tracer.start_as_current_span(span_name) as span:
# ── 公共核心子集 → 标准/约定属性 ──
# traceId 作 gen_ai.conversation.id(与框架原生 TracingMiddleware 同键:Studio 据此把同一 run 的步聚一条 trace)。
span.set_attribute("gen_ai.conversation.id", str(self.trace_id))
if step is not None:
span.set_attribute("agentscope.tier2.step", int(step))
ts = trace_step.get("timestamp")
if ts:
span.set_attribute("agentscope.tier2.timestamp", str(ts))
# ── cost.tokens → gen_ai.usage.*(Studio 的 token 用量视图按这两个标准键展示)──
cost = trace_step.get("cost") or {}
tokens = (cost or {}).get("tokens") or {}
if "in" in tokens:
span.set_attribute("gen_ai.usage.input_tokens", int(tokens.get("in") or 0))
if "out" in tokens:
span.set_attribute("gen_ai.usage.output_tokens", int(tokens.get("out") or 0))
# ── tier2 扩展三段 → agentscope.function.* + tier2 自有属性(JSON 序列化,Studio 在 Metadata 区展示)──
# 用 agentscope.function.input/output 这两个 reference 明确列出的「Opt-In」扩展键承载推理/动作/观察,
# 让 Studio 能按 AgentScope 扩展约定展示;原始三段也各存一份 tier2 私有键便于反查。
for seg_key in ("reasoning", "action", "observation"):
seg = ext.get(seg_key)
if seg:
span.set_attribute(
f"agentscope.tier2.{seg_key}",
json.dumps(seg, ensure_ascii=False),
)
# 动作段的工具名也铺到标准键(gen_ai.tool.name),Studio 工具调用视图按它高亮。
tool = (ext.get("action") or {}).get("tool")
if tool:
span.set_attribute("gen_ai.tool.name", str(tool))
# 原始事件类型(反查用)。
if raw.get("event"):
span.set_attribute("agentscope.tier2.raw_event", str(raw.get("event")))
# ── verdict → span 状态:如实反映工具/门结果(observe-only,不改任何生成判定)──
verdict = trace_step.get("verdict")
if verdict is not None:
span.set_attribute("agentscope.tier2.verdict", str(verdict))
if str(verdict).lower() in _ERROR_VERDICTS:
# 工具失败 / 被拒 / 中断 → span 标 ERROR(只是把已发生的结果反映到 trace 颜色,不影响主链)。
span.set_status(StatusCode.ERROR, str(verdict))
else:
span.set_status(StatusCode.OK)
def shutdown(self) -> None:
"""收口:flush + 关私有 provider,把 BatchSpanProcessor 攒批里残留的 span 发出(best-effort,绝不抛)。
编排器在一次 run 收口处调一次(对齐 TraceAdapter.summary 那一步);不调也不会崩(进程退出时 batch 可能丢尾批,
observe-only 容忍 trace 丢尾批不影响生成产物)
"""
if self._provider is None:
return
try:
# force_flush:把攒批里没发的 span 立刻发出(给个上限超时,避免收口卡住)。
self._provider.force_flush(timeout_millis=5000)
except Exception as exc: # best-effort:flush 失败只告警
_warn_once("flush-fail", f"force_flush 残留 span 失败(忽略):{exc}")
try:
self._provider.shutdown()
except Exception as exc:
_warn_once("shutdown-fail", f"关 Studio provider 失败(忽略):{exc}")
# ── no-op sink:取不到 Studio 地址时返回它,插进 TraceAdapter.sink 槽等于「没接 Studio」(只留内存 adapter)──
class _NoopStudioSink:
"""关闭态占位 sink:__call__ 什么都不做、shutdown 什么都不做。让调用方无需判 None、统一拿 sink + 调 shutdown。"""
enabled = False
def __call__(self, trace_step: dict) -> None: # noqa: D401 —— 故意 no-op
return
def shutdown(self) -> None: # noqa: D401 —— 故意 no-op
return
def make_studio_sink(
trace_id: str,
*,
studio_url: Optional[str] = None,
) -> Callable[[dict], None]:
"""构造一个 Studio 旁路 sink:解析到 Studio 地址 → StudioSink;取不到 → no-op sink(默认关闭)。
这是给编排器/run 的唯一入口:无论开没开 Studio,都拿到一个可直接塞进 TraceAdapter(sink=...) 的可调用对象,
且都带 .shutdown()(no-op 时也安全可调) 调用方零分支判断,observe-only 不影响主链
:param trace_id: 本次生成 traceId(贯穿)
:param studio_url: 显式 Studio HTTP base(最低优先级;一般留空, env TIER2_STUDIO_URL / infra_config)
:return: 一个 Callable[[dict], None] 且带 .shutdown() sink 对象(StudioSink _NoopStudioSink)
"""
url = resolve_studio_url(studio_url)
if not url:
# 取不到地址 = 默认关闭:返回 no-op(现有 CLI 真跑不受任何影响)。
return _NoopStudioSink()
return StudioSink(trace_id=trace_id, studio_url=url)
if __name__ == "__main__": # pragma: no cover —— 本地自检:无 Studio 地址时应拿到 no-op,且 __call__/shutdown 不炸
s = make_studio_sink(trace_id="demo-trace-001")
print("sink type:", type(s).__name__) # 期望 _NoopStudioSink(未设 TIER2_STUDIO_URL)
s({"traceId": "demo-trace-001", "step": 0, "ext": {"raw": {"event": "ReplyStartEvent"}}})
s.shutdown()
print("noop sink okobserve-only 关闭态:吃事件不报错、shutdown 不报错)")

View File

@ -377,6 +377,59 @@ class TraceAdapter:
}
def with_studio(
trace_id: str,
*,
sink: Optional[Callable[[dict], None]] = None,
keep_in_memory: bool = True,
studio_url: Optional[str] = None,
) -> "TraceAdapter":
"""便利件:建一个把每步既落原 sink、又旁路推 AgentScope Studio 的 TraceAdapter(C1 观测接线)。
为什么是便利件而不改 TraceAdapter 构造:TraceAdapter sink 槽是单 sink 契约(sink(dict)->None),
本件不破坏它 而是把 sinkStudio sink用一个 _fanout 串成一个 sink 再塞进去(对调用方透明)
Studio sink 默认关闭(取不到 Studio 地址即 no-op, studio_sink.make_studio_sink),所以:
- 没开 Studio( TIER2_STUDIO_URL / infra_config[studio]):行为 == 直接 TraceAdapter(sink=sink),零影响;
- 开了 Studio:每步在原 sink 之外多推一份 OTLP span Studio(observe-only,best-effort,不阻塞主链)
studio_sink 在函数内**惰性 import**:6c6g opentelemetry 依赖时本模块顶层仍可裸 import(只有真用到本件
Studio 地址可达时才触发 otel 依赖)Studio sink .shutdown() 由编排器在 run 收口处调
(经返回 adapter _studio_sink 属性拿到,或调用方自己持有 sink shutdown)
:param trace_id: 本次生成 traceId(贯穿)
:param sink: 原有落库 sink(JSONL/入库/message_bus );None 不另落,(可能) Studio + 内存
:param keep_in_memory: 是否在 adapter.steps 留内存副本(透传 TraceAdapter)
:param studio_url: 显式 Studio HTTP base(一般留空, env / infra_config)
:return: 配好 fanout sink TraceAdapter; ._studio_sink 属性 = Studio sink(供收口调 shutdown)
"""
# 惰性 import:把 otel 依赖关进本函数,保证 trace.py 顶层 6c6g 仍可裸 import。
from .studio_sink import make_studio_sink # noqa: PLC0415
studio_sink = make_studio_sink(trace_id, studio_url=studio_url)
def _fanout(trace_step: dict) -> None:
"""把一条 trace_step 同时喂给原 sink 与 Studio sink;**两者各自 best-effort,互不连累**。
注意: fanout 仍被 TraceAdapter._emit_to_sink 的外层 try 兜一次;但为 sink 失败不连累推 Studio
Studio 失败不连累原 sink,这里再各包一层 任一侧异常只告警不抛不阻断另一侧与主链(observe-only)
"""
if sink is not None:
try:
sink(trace_step)
except Exception as exc: # 原 sink 失败:不连累 Studio 推送,也不抛(交外层计 dropped)
print(f"[tier2-trace] 原 sink 写失败(fanout 兜,traceId={trace_id}):{exc}", flush=True)
# Studio sink 自身已 best-effort(取不到地址 = no-op;发送失败只告警),这里再兜一层防御性 try。
try:
studio_sink(trace_step)
except Exception as exc: # 理论上 studio_sink 自己已兜;防御性再兜,绝不让观测旁路波及主链
print(f"[tier2-trace] Studio sink 旁路异常(忽略,traceId={trace_id}):{exc}", flush=True)
adapter = TraceAdapter(trace_id, sink=_fanout, keep_in_memory=keep_in_memory)
# 把 Studio sink 挂到 adapter 上,供编排器在 run 收口处 adapter._studio_sink.shutdown()(flush 残留 span)。
adapter._studio_sink = studio_sink # type: ignore[attr-defined]
return adapter
if __name__ == "__main__": # pragma: no cover —— 本地自检:用桩事件(dict 形态)跑映射链路
import json

View File

@ -0,0 +1,425 @@
"""service/control_plane.py —— tier2 富游戏自治线 · 服务态【有界 resume 控制面】(本批皇冠;B1)。
它补的是哪个 followup
CLI 主链 worker.agent_loop.studio.run_studio(:404-458) Agent ReAct 循环之外套了一圈
有界 resume:AgentScope 2.0.2 原生 ReAct 模型产出无 tool_call 的纯文本回合即退出
(agentscope/agent/_agent.py:612)实测 M3 调一次 run_gates 看到 decision=fix 就产空文本收尾
循环退出(看一次 verdict 就放弃不自纠)run_studio for attempt in range(max_resumes+1):
replycheckpoint门绿则踹 finish / 否则带 verdict 反馈 resume 续修兜住了它
**服务态(service/app.py create_app :8200)没有这道循环**:它的 chat 是每回合一次
fire-and-forget(_service/_chat.py),由消费方(控制面)决定要不要再 POST /chatservice/app.py
文件头(:34-42) service/bootstrap.py(:25-32)都把它列为 followup本模块就是来补这个 followup
在服务消费方实现等价 CLI max_resumes 的有界 resume 控制面,**绝不重写任何 REST 调用 / 生成逻辑**,
import 复用 bootstrap(REST 原语)+ run(纯代码门 / 落库)
CLI 态的关键差异 为什么不靠内存 session 判收敛
服务态九工具(scaffold_init/write_source/build/run_gates/finish) _tier2_tools_factory 每个
chat 回合工厂新建一个 Tier2Session(app.py:_tier2_tools_factory:129-154, session_id game_id),
其内存态 session.finished / last_verdict **跨回合不保留**(每回合一个新 Tier2Session 实例)
所以本控制面不靠内存 session判收敛,而是每回合 chat run 结束后独立跑 run.run_gates 判门
门是机器判的(run_gates 纯代码 LLM on-disk 源工程),这正是项目门机器判控制面绝不
自评翻绿的哲学落库同理: on-disk 源工程重建 source_project + run.persist_source_project,
不依赖内存 session.finished
game_id run_gates 评门目标的绑定(诚实写清)
服务态九工具实际写文件的工程目录 = run._workdir(<game_id>),而九工具的 game_id 绑的是 AgentScope
session_id(app.py:_tier2_tools_factory:150game_id session_id)故本控制面评门时,
**run_gates game_id 必须用 start_new_game 返回的 session_id**这样评的就是九工具刚写的那个工程目录,
二者指向同一物理目录(game-runtime/games/_tier2-gen/<session_id>),不会评错对象
收敛 / 停机判据(对齐 run_studio:436-449)
外层 for attempt in range(max_resumes+1):
a. 等本回合 chat run 结束:消费 SSE GET /sessions/{session_id}/stream,读到本轮 reply
REPLY_END 事件(agentscope/event/_event.py:EventType.REPLY_END;也认 EXCEED_MAX_ITERS 为结束)
即视为这轮 agent 跑完了带总超时兜底;SSE 断流也据当前 on-disk 状态继续评门,绝不卡死
b. 独立评门:res = run.run_gates(session_id, play_spec);v = res['verdict']
c. 门绿(v.decision=='accept' layerResults.L1.passed) on-disk 源工程落库,break 成功
d. attempt>=max_resumes 预算耗尽,break(停机原因 budget_exhausted)
e. 否则:fb = run.verdict_feedback(v);bootstrap.resume_session(..., feedback=续跑指令),回到 a
熔断/异常(resume / SSE 长时间无事件 / run_gates 异常) best-effort 记日志落部分结果返回,绝不静默吞
惰性 import 红线本模块顶层 import 标准库bootstrap/run/store/genconfig 这些包内模块顶层会
牵出 agentscope / httpx(6c6g 未装),故对它们一律在函数体内 import,保证 6c6g
`python -m py_compile service/control_plane.py` 且能裸 `import service.control_plane` 不炸
真跑门(run_gates / chrome / esbuild)只在 mini-desktop;6c6g run_gates 返回结构化失败(诚实失败)
"""
from __future__ import annotations
import json
import time
from typing import Any
# ── SSE 事件常量(源码核验 agentscope/event/_event.py:EventType)──
# EventBase.model_config use_enum_values=True ⇒ model_dump(mode="json") 的 "type" 字段是字符串值。
# ReplyEndEvent.type = EventType.REPLY_END → 序列化后 type=="REPLY_END",标志本回合 agent.reply_stream 收尾。
# ExceedMaxItersEvent(type=="EXCEED_MAX_ITERS")是另一种「本回合结束」收尾(内层 ReAct 撞 max_iters)。
_EVENT_REPLY_END = "REPLY_END"
_EVENT_EXCEED_MAX_ITERS = "EXCEED_MAX_ITERS"
# 本回合「结束」事件集合(读到其一即视为这轮 chat run 跑完,可评门)。
_TURN_END_EVENTS = frozenset({_EVENT_REPLY_END, _EVENT_EXCEED_MAX_ITERS})
def _infer_role(rel: str) -> str:
"""据工程内相对路径推断 A3 fileTree[].role(与 run.load_fixture_scaffold 的 _role 同口径)。
服务态控制面据 on-disk 文件重建 source_project,需要给每个文件标 role(交付契约 fileTree[].role)
fixture 脚手架同一套推断规则,保证重建出的 fileTree role 与生成期一致( split-brain)
"""
if rel == "src/main.js":
return "entry"
if "/scenes/" in rel:
return "scene"
if "/systems/" in rel or rel == "src/game-core.js":
return "system"
if "/data/" in rel:
return "config"
if "/util/" in rel or rel.endswith("seeded-random.js"):
return "lib"
if "/assets/" in rel:
return "asset-manifest"
return "other"
def _collect_on_disk_project(game_id: str) -> dict | None:
"""据 on-disk 工程目录(run._workdir/<game_id>)重建源工程交付形状(七要素)+ 文件全文清单。
为什么由控制面据 on-disk 重建,而不取内存 session.finished服务态九工具每回合工厂新建 Tier2Session,
内存态跨回合不保留(见文件头)门绿后要落库的源工程,只能从九工具刚写盘的工程目录读回重建
重建的形状对齐 worker.toolkit.finish 组装的 A3 源项目契约七要素(schemaVersion/projectType/fileTree/
entry/buildProfile/depLock/contentHash/addressing),保证落库口径与 CLI finish 一致(F3 无漂移)
Returns:
{"source_project": {七要素}, "file_list": [{path, content}]} None(目录不存在/无源文件)
"""
# 惰性 import:run / toolkit 顶层牵出 agentscope(6c6g 未装),只在真要落库时才 import。
from worker import run as _run # noqa: PLC0415
from worker import toolkit as _toolkit # noqa: PLC0415
wd = _run._workdir(game_id)
if not wd.exists():
return None
# 收集 on-disk 源文件(排除构建产物 bundle / evidence / play-spec / index.html 这些非源文件)。
# 只收 src/ 与 data/ 下的源(对齐 toolkit 累积的工程文件树语义),其余是 build/run_gates 副产物。
file_list: list[dict] = []
for p in sorted(wd.rglob("*")):
if not p.is_file():
continue
rel = str(p.relative_to(wd)).replace("\\", "/")
# 跳过构建 / 真玩副产物(非交付源文件)。
if rel in ("bundle.iife.js", "play-spec.json", "index.html"):
continue
if rel.startswith("evidence/") or "/node_modules/" in ("/" + rel):
continue
try:
file_list.append({"path": rel, "content": p.read_text(encoding="utf-8")})
except Exception as e: # noqa: BLE001 —— 单文件读失败记日志、不中断重建其余文件
print(f"[tier2-control] ⚠ 重建源工程读文件失败 {rel}: {type(e).__name__}: {e}", flush=True)
if not file_list:
return None
# 组装 A3 源项目契约七要素(与 toolkit.finish:344-357 同口径;contentHash 复用 toolkit._content_hash)。
file_tree = [{"path": f["path"], "role": _infer_role(f["path"])} for f in file_list]
chash = _toolkit._content_hash(file_list)
source_project = {
"schemaVersion": "tier2-1.0",
"projectType": "tier2-phaser",
"fileTree": file_tree,
"entry": "src/main.js",
"buildProfile": {
"bundler": "esbuild", "format": "iife",
"globalName": _run.DEFAULT_GLOBAL_NAME, "minify": True, "target": "es2019",
},
# depLock.phaser:据 on-disk 无法可靠读出精确版本,用 toolkit 默认值(与 finish 同源)。
"depLock": {"phaser": "3.80.1"},
"contentHash": chash,
"addressing": {"store": "mysql+oss", "fetchById": "contentHash"},
}
return {"source_project": source_project, "file_list": file_list}
async def _wait_for_turn_end(base_url: str, agent_id: str, session_id: str, *,
user_id: str, timeout_s: float, idle_timeout_s: float) -> dict:
"""消费 SSE GET /sessions/{session_id}/stream,等本回合 chat run 结束(读到 REPLY_END / EXCEED_MAX_ITERS)。
源码核验(agentscope/app/_router/_session.py:425-534):
- 端点 = GET /sessions/{session_id}/stream,**需带 query agent_id**(:432);
- 返回 text/event-stream,帧形如 `data: {json}\\n\\n`,空闲每 30s 发心跳注释 `:\\n\\n`(:446/522);
- 流先 replay 当前 run 的缓冲事件, live 订阅;订阅是长连(不随单回合结束而关,只随 bus 关闭结束),
所以本函数靠读到本回合的 REPLY_END/EXCEED_MAX_ITERS自行收尾,不等服务端关流
兜底纪律(绝不卡死):
- 总超时 timeout_s:超时即返回 {ended: False, reason: 'total_timeout'},由上层据 on-disk 继续评门;
- 空闲超时 idle_timeout_s:连续这么久没收到任何 data (只有心跳/无字节) 视为断流,返回
{ended: False, reason: 'idle_timeout'},同样据 on-disk 继续评门(SSE 断流不致命)
Returns:
{ended: bool, reason: str, endEvent: dict|None}ended=True 表示读到本回合结束事件
"""
import httpx # noqa: PLC0415 —— 惰性 import 红线(6c6g 未必装 httpx)
headers = {"X-User-Id": user_id} if user_id else {}
url = f"{base_url}/sessions/{session_id}/stream"
params = {"agent_id": agent_id}
t_start = time.perf_counter()
last_event_at = t_start
print(f"[tier2-control] 订阅 SSE 等本回合结束: session={session_id} agent={agent_id} "
f"(总超时 {timeout_s:.0f}s / 空闲超时 {idle_timeout_s:.0f}s)", flush=True)
try:
# read=None:逐行读流不受单次读超时打断;总时长 / 空闲由本函数显式判(下方 elapsed/idle)。
timeout_cfg = httpx.Timeout(connect=15.0, read=None, write=15.0, pool=15.0)
async with httpx.AsyncClient(timeout=timeout_cfg) as http:
async with http.stream("GET", url, params=params, headers=headers) as resp:
if resp.status_code >= 300:
body = (await resp.aread())[:300]
print(f"[tier2-control] ⚠ SSE 订阅非 2xx: {resp.status_code}: {body!r}", flush=True)
return {"ended": False, "reason": f"sse_http_{resp.status_code}", "endEvent": None}
async for line in resp.aiter_lines():
now = time.perf_counter()
# 总超时:本回合等太久,据 on-disk 继续(不卡死)。
if now - t_start > timeout_s:
print(f"[tier2-control] ⚠ SSE 等本回合结束总超时(>{timeout_s:.0f}s),"
"据 on-disk 继续评门。", flush=True)
return {"ended": False, "reason": "total_timeout", "endEvent": None}
if not line:
# 空行(SSE 帧分隔)——不更新空闲计时(只有真 data 帧才算「有进展」)。
if now - last_event_at > idle_timeout_s:
print(f"[tier2-control] ⚠ SSE 空闲超时(>{idle_timeout_s:.0f}s 无 data 帧),"
"疑断流,据 on-disk 继续评门。", flush=True)
return {"ended": False, "reason": "idle_timeout", "endEvent": None}
continue
if line.startswith(":"):
# 心跳注释帧(:\n\n);连接活着但本回合还没结束,据空闲超时判断是否断流。
if now - last_event_at > idle_timeout_s:
print(f"[tier2-control] ⚠ SSE 只收到心跳、空闲超时(>{idle_timeout_s:.0f}s),"
"据 on-disk 继续评门。", flush=True)
return {"ended": False, "reason": "idle_timeout", "endEvent": None}
continue
if not line.startswith("data:"):
continue
# 收到一个 data 帧:解析事件,刷新空闲计时。
last_event_at = now
payload = line[len("data:"):].strip()
if not payload:
continue
try:
event = json.loads(payload)
except Exception as e: # noqa: BLE001 —— 单帧解析失败记日志、不中断订阅
print(f"[tier2-control] ⚠ SSE 帧解析失败(已跳过): {type(e).__name__}: {e}", flush=True)
continue
etype = event.get("type")
# 只认本回合的结束事件(REPLY_END / EXCEED_MAX_ITERS);其余事件(模型流 / 工具调用)略过。
if etype in _TURN_END_EVENTS:
print(f"[tier2-control] SSE 读到本回合结束事件 type={etype}"
f"(reply_id={event.get('reply_id')});评门。", flush=True)
return {"ended": True, "reason": etype, "endEvent": event}
# 流自然结束(bus 关闭等)——本回合可能已收尾,据 on-disk 评门。
print("[tier2-control] ⚠ SSE 流结束但未读到本回合结束事件,据 on-disk 继续评门。", flush=True)
return {"ended": False, "reason": "stream_closed", "endEvent": None}
except Exception as e: # noqa: BLE001 —— SSE 任何异常都 best-effort:不卡死,据 on-disk 继续评门
print(f"[tier2-control] ⚠ SSE 订阅异常(据 on-disk 继续评门): {type(e).__name__}: {e}", flush=True)
return {"ended": False, "reason": f"sse_error:{type(e).__name__}", "endEvent": None}
# ── 续跑指令(等价 CLI run_studio:454-458 的「停了但门没绿,按失败门继续修、门绿再 finish」)──
def _resume_feedback_text(verdict_feedback: str) -> str:
"""把失败门反馈包成续跑指令(口径对齐 studio.py:454-458)。"""
return (
"你刚才停下了,但验收门还没全绿——不要放弃。这是上次 run_gates 的失败门:\n"
f"{verdict_feedback}\n"
"请在循环里:据失败门 write_source 针对性修(数据表 schema 错就先 validate_datatable 看平台要的 key),"
"build→run_gates→read_verdict,直到门绿再 finish。一步步来,先修最关键的致命门。")
async def drive_generation(
base_url: str,
game_id: str,
brief: str,
play_spec: dict | None = None,
*,
max_resumes: int | None = None,
model_name: str = "MiniMax-M3",
writer_max_iters: int | None = None,
user_id: str = "tier2",
sse_turn_timeout_s: float = 1200.0,
sse_idle_timeout_s: float = 300.0,
) -> dict:
"""服务态有界 resume 控制面:驱动一款富游戏从启动到门绿落库(等价 CLI run_studio 的外层 resume)。
bootstrap 的现成 REST 原语 + run 的纯代码门驱动 service/app.py create_app;**绝不重写 REST/生成逻辑**
收敛判据全靠 run.run_gates 机器判门(零自评绝不放松门),落库据 on-disk 源工程(不靠内存 session)
Args:
base_url: 运行中的 Agent Service 根地址( http://100.64.0.7:8200)
game_id: 调用方语义上的游戏标识(仅用于日志/返回;真实评门 game_id = 框架分配的 session_id,见下)
brief: 一句话题面
play_spec: 真玩驱动规格(透传 run.run_gates;None 走默认 business-sim driver)
max_resumes: 外层 resume 上限;None genconfig.get('iteration','max_resumes',6)(对齐 studio)
model_name: M3 模型名(默认 MiniMax-M3, new-api Anthropic 原生)
writer_max_iters: 单写 ReAct 放开轮数;None genconfig.get('iteration','writer_max_iters',40)
user_id: 多租户用户标识( X-User-Id )
sse_turn_timeout_s: 单回合 SSE 等结束的总超时(兜底,绝不卡死)
sse_idle_timeout_s: 单回合 SSE 空闲( data )断流判定超时
Returns:
{game_id, agent_id, session_id, finished(bool=门绿+落库), attempts, last_verdict,
store_addressing, wall_s, stopped_reason}
"""
# 惰性 import:bootstrap(REST 原语,牵出 httpx/worker)/ run(门 + 落库,牵出 agentscope)/ genconfig。
from worker import genconfig, run as _run # noqa: PLC0415
from . import bootstrap # noqa: PLC0415
t0 = time.perf_counter()
# 旋钮口径对齐 studio:max_resumes / writer_max_iters 缺省从 generation.yaml 读。
if max_resumes is None:
max_resumes = genconfig.get("iteration", "max_resumes", 6)
if writer_max_iters is None:
writer_max_iters = genconfig.get("iteration", "writer_max_iters", 40)
result: dict[str, Any] = {
"game_id": game_id, "agent_id": None, "session_id": None,
"finished": False, "attempts": 0, "last_verdict": None,
"store_addressing": None, "wall_s": 0.0, "stopped_reason": None,
}
# ── ① 启动:bootstrap.start_new_game(注册凭据→建 agent→建 session→发 kick)──
# 关键集成点:评门用的 game_id 必须 = 框架分配的 session_id(九工具据 session_id 当 game_id 管目录,
# 见 app.py:_tier2_tools_factory:150),否则 run_gates 评的不是九工具刚写的那个工程目录。
try:
started = await bootstrap.start_new_game(
base_url, brief, user_id=user_id, model_name=model_name,
writer_max_iters=writer_max_iters)
except Exception as e: # noqa: BLE001 —— 启动失败 best-effort 落结果、记日志,不静默吞
result["stopped_reason"] = f"start_failed:{type(e).__name__}: {e}"
result["wall_s"] = round(time.perf_counter() - t0, 1)
print(f"[tier2-control] game={game_id} 启动失败:{type(e).__name__}: {e}", flush=True)
return result
agent_id = started.get("agent_id")
session_id = started.get("session_id")
result["agent_id"] = agent_id
result["session_id"] = session_id
# 评门 game_id = 框架 session_id(与九工具工程目录一致)。
gate_game_id = session_id
print(f"[tier2-control] game={game_id} 启动: agent={agent_id} session={session_id} "
f"(评门 game_id=session_id;max_resumes={max_resumes})", flush=True)
if not session_id or not agent_id:
result["stopped_reason"] = "start_missing_ids(start_new_game 未回 agent_id/session_id)"
result["wall_s"] = round(time.perf_counter() - t0, 1)
return result
last_verdict: dict | None = None
# ── ② 外层有界 resume 循环(对齐 run_studio:424-458)──
for attempt in range(max_resumes + 1):
result["attempts"] = attempt + 1
# a. 等本回合 chat run 结束(SSE;超时/断流据 on-disk 继续,绝不卡死)。
await _wait_for_turn_end(
base_url, agent_id, session_id, user_id=user_id,
timeout_s=sse_turn_timeout_s, idle_timeout_s=sse_idle_timeout_s)
# b. 独立评门:run.run_gates 纯代码判 on-disk 源工程(零 LLM、确定性、绝不自评翻绿)。
try:
gate = _run.run_gates(gate_game_id, play_spec)
except Exception as e: # noqa: BLE001 —— 评门异常 best-effort:记日志,带空 verdict 进续修分支
print(f"[tier2-control] game={game_id} attempt={attempt} run_gates 异常:"
f"{type(e).__name__}: {e}", flush=True)
gate = {"rc": 1, "verdict": None, "log": f"run_gates 异常:{type(e).__name__}: {e}"}
v = gate.get("verdict") or {}
last_verdict = v or last_verdict
result["last_verdict"] = last_verdict
# c. 门绿判定(decision==accept 且 L1.passed)。
l1 = ((v.get("layerResults") or {}).get("L1") or {})
if v.get("decision") == "accept" and l1.get("passed"):
print(f"[tier2-control] game={game_id} attempt={attempt} 验收门全绿,据 on-disk 落库。", flush=True)
# 据 on-disk 源工程落库(不依赖内存 session.finished)。
built = _collect_on_disk_project(gate_game_id)
if built:
addr = _run.persist_source_project(
gate_game_id, built["source_project"], built["file_list"], now_ts=time.time())
result["store_addressing"] = addr
result["finished"] = bool(addr) # 落库成功才算真 finished(门绿 + 落库)
if not addr:
print(f"[tier2-control] game={game_id} ⚠ 门绿但落库失败(产物仍在 GEN_DIR workdir);"
"stopped_reason=persist_failed。", flush=True)
result["stopped_reason"] = "gates_green_but_persist_failed"
else:
result["stopped_reason"] = "gates_green_persisted"
else:
# 门绿却读不到 on-disk 源工程(异常态)——诚实标注,不伪造落库。
print(f"[tier2-control] game={game_id} ⚠ 门绿但 on-disk 工程目录读不到源文件,无法落库。",
flush=True)
result["stopped_reason"] = "gates_green_but_no_on_disk_source"
break
# d. 预算耗尽:已是最后一次 attempt 仍未门绿 → 停。
if attempt >= max_resumes:
print(f"[tier2-control] game={game_id} resume 预算耗尽({max_resumes} 次)仍未门绿,停。",
flush=True)
result["stopped_reason"] = "budget_exhausted"
break
# e. 门没绿 → 带 verdict 失败反馈 resume 续修(头号自纠机制;等价 studio.py:451-458)。
fb = _run.verdict_feedback(v, gate.get("log") or "") if v else \
("尚无 verdict(疑真玩未产出 / harness 未就位)。" + (gate.get("log", "")[:400]))
print(f"[tier2-control] game={game_id} attempt={attempt} 门未绿"
f"(decision={v.get('decision')}),带反馈 resume 续修。", flush=True)
try:
await bootstrap.resume_session(
base_url, agent_id, session_id, user_id=user_id,
feedback=_resume_feedback_text(fb))
except Exception as e: # noqa: BLE001 —— resume 抛 = 熔断:best-effort 落部分结果返回,绝不静默吞
print(f"[tier2-control] game={game_id} attempt={attempt} resume 异常(熔断):"
f"{type(e).__name__}: {e}", flush=True)
result["stopped_reason"] = f"resume_failed:{type(e).__name__}: {e}"
break
if result["stopped_reason"] is None:
# 循环正常走完未命中任何分支(理论不达;兜底标注)。
result["stopped_reason"] = "loop_ended"
result["wall_s"] = round(time.perf_counter() - t0, 1)
print(f"[tier2-control] game={game_id} 结束: finished={result['finished']} "
f"attempts={result['attempts']} reason={result['stopped_reason']} "
f"wall={result['wall_s']}s", flush=True)
return result
def main(argv: list[str] | None = None) -> int:
"""CLI 入口:python -m service.control_plane <game_id> --base-url ... --brief ...。
惰性 import asyncio / argparse(顶层零重依赖红线;此处也只用标准库)真跑在 mini-desktop
"""
import argparse # noqa: PLC0415
import asyncio # noqa: PLC0415
parser = argparse.ArgumentParser(
description="tier2 服务态有界 resume 控制面:驱动一款富游戏到门绿落库。")
parser.add_argument("game_id", help="游戏标识(用于日志/返回;真实评门用框架分配的 session_id)。")
parser.add_argument("--base-url", required=True,
help="运行中的 Agent Service 根地址(如 http://100.64.0.7:8200)。")
parser.add_argument("--brief", required=True, help="一句话题面。")
parser.add_argument("--max-resumes", type=int, default=None,
help="外层 resume 上限(默认从 generation.yaml iteration.max_resumes 读)。")
parser.add_argument("--writer-max-iters", type=int, default=None,
help="单写 ReAct 放开轮数(默认从 generation.yaml iteration.writer_max_iters 读)。")
parser.add_argument("--model-name", default="MiniMax-M3", help="模型名(默认 MiniMax-M3)。")
parser.add_argument("--user-id", default="tier2", help="多租户用户标识(默认 tier2)。")
args = parser.parse_args(argv)
res = asyncio.run(drive_generation(
args.base_url, args.game_id, args.brief,
max_resumes=args.max_resumes, writer_max_iters=args.writer_max_iters,
model_name=args.model_name, user_id=args.user_id))
# 结构化结果打到 stdout(可被上层脚本捕获)。
print(json.dumps(res, ensure_ascii=False, indent=2), flush=True)
# 退出码:门绿落库成功 = 0,否则 1(供编排脚本判成败)。
return 0 if res.get("finished") else 1
if __name__ == "__main__":
import sys # noqa: PLC0415
sys.exit(main())

View File

@ -179,9 +179,9 @@ async def run_design_team(
model_factory,
*,
model_sink: list | None = None,
leader_max_iters: int = 12,
per_expert_cap: int = 2,
timeout_s: float = 240.0,
leader_max_iters: int | None = None,
per_expert_cap: int | None = None,
timeout_s: float | None = None,
) -> str:
"""跑工作室星形设计团队,返回一份连贯的富游戏设计稿(str;形状同原 _design_stage)。
@ -189,9 +189,12 @@ async def run_design_team(
汇总成功 = leader 跑完后产出非空最终设计稿;否则抛 DesignTeamError 由调用方 degrade
预算约束( agent token,必须有上限):
- leader_max_iters:leader ReAct 总轮数上限(= 调专家 + 汇总的天花板;默认 12,够调四专家各 2 + 汇总);
- per_expert_cap:单专家被调上限(默认 2; leader 反复刷同一专家);
- timeout_s:整团队墙钟硬超时(默认 240s),兜住模型挂起/leader 不收敛
- leader_max_iters:leader ReAct 总轮数上限(= 调专家 + 汇总的天花板;够调四专家各 2 + 汇总);
- per_expert_cap:单专家被调上限( leader 反复刷同一专家);
- timeout_s:整团队墙钟硬超时(),兜住模型挂起/leader 不收敛
三者均为 None 运行时读 generation.yaml design_team (default 12 / 2 / 240.0;口径同 studio.py
iteration.* 旋钮)**这是配置外置的兑现点**: generation.yaml design_team.timeout_s 即生效,
不必改代码(此前硬编码默认 240 导致 YAML timeout 不生效,已修)
任一预算触发都不让设计阶段挂死:超时/异常 DesignTeamError studio degrade 回单 agent
Args:
@ -211,6 +214,17 @@ async def run_design_team(
if model_sink is None:
model_sink = [] # 容错:调用方没传则内部建(此时成本汇总会漏算 team token,仅兜底用)
# ── 旋钮外置(运行时读 generation.yaml 的 design_team 段;口径同 studio.py 读 iteration.*)──
# 显式传入(非 None)优先;未传 → 从 genconfig 读,default 对齐历史硬编码值(12 / 2 / 240.0)。
# 修复点:此前这三个是写死的函数默认值,studio 调用时不传 → generation.yaml 的 design_team.* 形同虚设
# (改 YAML 调 timeout 不生效)。现在 None→genconfig.get,配置外置真正兑现。
if leader_max_iters is None:
leader_max_iters = genconfig.get("design_team", "leader_max_iters", 12)
if per_expert_cap is None:
per_expert_cap = genconfig.get("design_team", "per_expert_cap", 2)
if timeout_s is None:
timeout_s = genconfig.get("design_team", "timeout_s", 240.0)
# 每专家调用计数(跨同名工具多次调用累计;预算闸据它判超限)。
counter: dict[str, int] = {}

View File

@ -51,6 +51,22 @@ from agentscope.agent import ContextConfig # noqa: E402 —— 历史压缩配
DEEPSEEK_FLASH = genconfig.get("model", "deepseek_flash", "deepseek-v4-flash") # 主力便宜档
DEEPSEEK_PRO = genconfig.get("model", "deepseek_pro", "deepseek-v4-pro") # 强便宜档(退路树 Q1 读它的过门率)
# ── 强基线档默认模型名(C3;A 门「M3 证路(软)」)──
# 用强模型(Opus/Fable)跑同一条 studio/控制面线,把「路本身不通」和「便宜模型太弱」分开:
# 若强基线都过不了门,说明是路/harness 的问题;若强基线能过、便宜档不过,才是便宜模型天花板(退路树 R3)。
# 两者都是 Anthropic 原生经 new-api(与 M3 同协议);Fable 列出但不假设网关一定可用(取不到诚实报错,见 build_baseline_model)。
# 退路树 fallback_tree.STRONG_BASELINE_KEYS 用 "Opus"/"Fable" 当聚合 key;这里的常量是它们在 new-api 上的真实模型名。
BASELINE_OPUS = genconfig.get("model", "baseline_opus", "claude-opus-4-8") # 强基线 Opus(A 门主用)
BASELINE_FABLE = genconfig.get("model", "baseline_fable", "claude-fable-5") # 强基线 Fable(列出,不假设可用)
# 强基线聚合 key(fallback_tree 用)→ new-api 真实模型名 的映射。
# build_baseline_model 接受「聚合 key(Opus/Fable,大小写不敏感)」或「真实模型名」两种入参:
# 传 key → 查本表换成真名;传真名 → 原样用(便于调用方直接点名网关上的具体模型)。
_BASELINE_NAME_BY_KEY = {
"opus": BASELINE_OPUS,
"fable": BASELINE_FABLE,
}
class RecordingChatModel(AnthropicChatModel):
"""记录每次模型调用 usage 的 AnthropicChatModel 子类(per-call token 取证,供成本台账)。
@ -315,6 +331,72 @@ def build_model_openai(
)
def build_baseline_model(
model_name: str,
*,
base_url: str | None = None,
api_key: str | None = None,
max_tokens: int | None = None,
thinking_budget: int | None = None,
stream: bool = False,
max_retries: int | None = None,
record: bool = True,
) -> AnthropicChatModel:
"""构建「强基线」模型客户端(C3;A 门「M3 证路(软)」用)——让同一条 studio/控制面线换 Opus/Fable 等强模型跑。
为什么要它:A 门要把路本身不通便宜模型太弱分开把强模型(claude-opus-4-8 )接到与 M3
完全同一条生成线上同样经 build_model Anthropic 原生路同样的 studio.py 装配若强基线都过不了门,
问题在路/harness;若强基线能过便宜档不过,才坐实便宜模型天花板(退路树 fallback_tree R3 判定即读此)
build_model 的关系(加性,不动 build_model 的签名与行为):
- 本函数复用 build_model 的全部机制(凭据解析 client.resolve_base_url/get_api_key代理旁路
RecordingChatModel usage 取证Anthropic 原生 /v1/messages 协议),只是把模型名换成强基线;
- **thinking 默认关**:基线只为软证路通,默认 thinking_enable=False, thinking_budget 默认 0
max_tokens 不受必须 > thinking_budget约束( 0 build_model 0<max_tokens 恒成立);
调用方若想让 Opus 也开 thinking,显式传 thinking_budget>0 即可(此时 build_model 会强制 max_tokens>budget)
Args:
model_name: 强基线档可传聚合 key 'Opus'/'Fable'(大小写不敏感, _BASELINE_NAME_BY_KEY 换成
new-api 真名),或直接传真名( 'claude-opus-4-8',原样用)
base_url / api_key: build_model;None client 默认解析(已装代理旁路)
max_tokens: 输出上限;None generation.yaml model.max_tokens(默认 16000)
thinking_budget: None 默认 0( thinking,纯证路);>0 thinking build_model 强制 max_tokens>budget
stream / max_retries / record: build_model
Returns:
AnthropicChatModel(record=True 时为 RecordingChatModel,基线跑的成本也进台账)
诚实报错纪律(Fable):本函数不预判某基线在网关上一定可用它把模型名透传给 build_model,真正的
"网关有没有这个模型 / key 有没有权限" new-api 在首次真调时裁决一旦取不到(404/403/连不上),
build_model 内部的 anthropic SDK 会抛 RecordingChatModel 透传上来**绝不静默吞**,
调用方据真异常判该基线在本网关不可用(而非误以为跑出了 0 过门率)本机(6c6g)无网关 anthropic,
import/构造可成(惰性),真调一定在 mini-desktop
"""
# 聚合 key(Opus/Fable)→ new-api 真名;传的若已是真名(不在表里)则原样用。
resolved_name = _BASELINE_NAME_BY_KEY.get(model_name.strip().lower(), model_name)
# thinking 默认关(基线纯证路):budget=None → 0;此时 build_model 的「max_tokens>budget」恒成立。
# 显式传 budget>0 → 开 thinking,build_model 会强制校验 max_tokens>budget。
budget = 0 if thinking_budget is None else thinking_budget
thinking_on = budget > 0
print(
f"[tier2-config] build_baseline_model: 入参='{model_name}' → 真名='{resolved_name}' "
f"thinking={'on' if thinking_on else 'off'}(budget={budget}) record={record}",
flush=True,
)
# 复用 build_model 全部机制(Anthropic 原生路 + 凭据 + 代理旁路 + 成本取证),只换模型名与 thinking 开关。
return build_model(
resolved_name,
base_url=base_url,
api_key=api_key,
max_tokens=max_tokens,
thinking_enable=thinking_on,
thinking_budget=budget,
stream=stream,
max_retries=max_retries,
record=record,
)
def model_name_from_env(default: str | None = None) -> str:
"""单写 agent 的模型名:env TIER2_MODEL 覆盖 > 显式 default > generation.yaml 的 model.default_model_name(现值 MiniMax-M3)。

View File

@ -73,6 +73,12 @@ _BUILTIN_DEFAULTS: dict[str, dict[str, Any]] = {
"default_model_name": "MiniMax-M3", # model_name_from_env 默认(M3 路)
"deepseek_flash": "deepseek-v4-flash", # DEEPSEEK_FLASH 便宜主力档
"deepseek_pro": "deepseek-v4-pro", # DEEPSEEK_PRO 强便宜救场档
# ── 强基线档(C3;A 门「M3 证路(软)」用强模型跑同一条 studio/控制面线,把「路不通」与「便宜模型太弱」分开)──
# Opus = Anthropic 原生经 new-api(与 M3 同路);Fable 同协议但 **不假设网关一定可用**
# (build_baseline_model 取不到即诚实报错,不静默)。退路树 fallback_tree.STRONG_BASELINE_KEYS 读 "Opus"/"Fable" 当 key,
# 这里两键存的是 new-api 上的真实模型名(消费方:worker/config.py build_baseline_model 的 baseline_name → 真名映射)。
"baseline_opus": "claude-opus-4-8", # 强基线 Opus(Anthropic 原生路;A 门软证路主用)
"baseline_fable": "claude-fable-5", # 强基线 Fable(列出但不假设网关可用;取不到诚实报错)
"max_tokens": 16000, # build_model / build_model_openai 默认输出上限
"thinking_enable": True, # build_model 默认开 thinking 分离
"thinking_budget": 8000, # build_model 默认 thinking 预算(须 < max_tokens)

View File

@ -0,0 +1,337 @@
"""mcp_tools.py —— tier2 单写 toolkit 的【可选】MCP 工具接入层(图说 E 族 · skills/工具面可扩展性落地)。
这份解决什么问题
tier2 单写 agent 现在用 build_toolkit 装九个本地工具(scaffold_init/write_source/build/run_gates/finish),
那是 spike feie-005 已验证过九门的主链,绝不能动图说 E 族另规划了 skills/工具面的可扩展性
MCP(Model Context Protocol)接入外部工具(如资产检索 / 素材库 / 联网查询类),让生成 agent 在九工具
之外按需多一组外部能力本模块就是那条接入线:按配置注册若干 MCP server 的工具进 toolkit收敛成一处,
默认关闭不开 MCP ,toolkit 与现在字节级一致(九工具零改),只有显式开关打开才会多挂 MCP 工具
AgentScope 2.0.2 真实 MCP API(逐一核对源码,绝不臆造)
来源:/root/oss/agentscope/src/agentscope/mcp/{_mcp_client.py,_config.py,__init__.py}
/root/oss/agentscope/src/agentscope/tool/{_toolkit.py,_tool_group.py}
- MCP 客户端类 = agentscope.mcp.MCPClient(BaseModel),构造参数(全部核实):
name(str,须匹配 ^[a-zA-Z0-9_-]+$,会拼成模型可见的工具名 mcp__{name}__{tool})
is_stateful(bool,STDIO 必须 True;HTTP True/False)
mcp_config(StdioMCPConfig | HttpMCPConfig, type 字段判别)
可选 enable_tools / disable_tools(list[str])execution_timeout(float)
- 传输配置两类(agentscope.mcp.{StdioMCPConfig,HttpMCPConfig}):
StdioMCPConfig(command, args?, env?, cwd?) 本地进程,stdio 传输;
HttpMCPConfig(url, headers?, timeout?=30.0) 远端 HTTP,SSE / streamable-http url 后缀自动判
(url /sse /messages/ 结尾 SSE,否则 streamable-http)
- 注册进 toolkit 的唯一姿势(2.0.2 Toolkit.register_tool / register_mcp_client 这类后置方法):
构造器 Toolkit(tools=[...], mcps=[MCPClient, ...]) mcps 落进保留的 "basic" (_toolkit.py:127-134)
toolkit 在运行时惰性对每个 client list_tools() 枚举其工具(_toolkit.py:_get_available_tools L519-521),
stateless HTTP client 不必预先 connect();只有 is_stateful=True client 必须在构造 toolkit 前已 connect
(构造器会校验:stateful 且未连接直接 ValueError,_toolkit.py:145-151)
- "basic" 组对象暴露在 toolkit.tool_groups[0].mcps(list[MCPClient],_tool_group.py:38),可后置 append
这是 toolkit 构造完成后唯一能再加 MCP 的合法路径( async 场景连 stateful 客户端后挂入)
两个入口(对应两种调用时机)
build_mcp_clients(configs) -> list[MCPClient](同步):
构造MCPClient 对象( connect list_tools), build_toolkit 这种同步调用点把结果塞进
Toolkit(mcps=...)默认只接 stateless 客户端(HTTP is_stateful=False,无需 await connect);遇到声明
stateful 的配置同步路径下无法 connect跳过并告警(让它走 async 路径)
async def register_mcp_tools(toolkit, *, configs)(异步):
异步调用点(已在事件循环里):构造 client, stateful 的先 await connect(), append
toolkit "basic" mcps这是 spec 要求的 register_mcp_tools 入口,覆盖 STDIO / stateful HTTP
门控:默认关闭(本相红线)
配置从 tier2/config/generation.yaml 新增 mcp (genconfig 同源 config_path(),env TIER2_GENCONFIG 可改指);
也支持 env 单点开关 TIER2_GEN__MCP__ENABLED判定 enabled 的三要素任一不满足即整体 no-op(返回空列表):
- mcp.enabled 必须显式为真(默认 false);
- mcp.servers 必须是非空列表
mcp 段缺失 / / 解析失败 no-op + 一次性告警,绝不抛绝不中断生成主链(best-effort,对齐 genconfig 纪律)
惰性 import(6c6g 红线)
6c6g 没装 agentscope[full] / mcp 客户端依赖;agentscope.mcp 一律在函数体内import,模块顶层零重依赖,
保证 6c6g py_compile + import 本模块import 失败 当作MCP 不可用no-op + 告警,不阻断主链
最小示例配置(接一个 MCP server;默认不依赖任何外部 MCP 进程)
tier2/config/generation.yaml 末尾加(整段默认不存在 = 默认关闭):
mcp:
enabled: false # 总开关:默认 false。改 true 才会挂 MCP 工具(九工具行为永远不受影响)
servers:
# 例 A:远端 HTTP MCP(stateless,无需 connect;最适合同步 build_toolkit 路径)
- name: asset-search # 仅 [a-zA-Z0-9_-];会拼成工具名 mcp__asset-search__<tool>
transport: http # http | stdio
url: http://100.64.0.x:9xxx/mcp # streamable-http;以 /sse 结尾则自动走 SSE
stateful: false # HTTP 默认 stateless,同步路径可用
# headers: { Authorization: "Bearer xxx" } # 可选;凭据走 docs/内网凭据与端点.md,勿硬编码
# timeout: 30.0
# enable_tools: [search_asset] # 可选白名单
# 例 B:本地 STDIO MCP(必 stateful,只能走 async register_mcp_tools 路径)
- name: local-fs
transport: stdio
command: mcp-server-filesystem
args: ["--root", "/some/dir"]
# stateful: true # STDIO 强制 stateful(填不填都按 true)
"""
from __future__ import annotations
from typing import Any
# 顶层只引标准库 + genconfig(后者顶层也零重依赖);agentscope.mcp 一律在函数体内惰性 import(6c6g 无)。
try:
from . import genconfig as _genconfig # 包内相对导入(worker.mcp_tools)
except Exception: # pragma: no cover —— 兜底:被当顶层模块跑时退化为绝对导入
import importlib
_genconfig = importlib.import_module("worker.genconfig")
# ── env 单点总开关名(与 genconfig 的 TIER2_GEN__<AREA>__<KEY> 同前缀口径)──
_ENV_ENABLED = "TIER2_GEN__MCP__ENABLED"
def _warn_once(tag: str, msg: str) -> None:
"""复用 genconfig 的一次性告警(同 [tier2-genconfig] 渠道)防刷屏;失败兜底用 print。"""
try:
_genconfig._warn_once(tag, "[mcp] " + msg) # 复用同一去重集合,统一可追溯前缀
except Exception: # pragma: no cover —— genconfig 内部结构变动时的兜底
print(f"[tier2-genconfig] [mcp] {msg}", flush=True)
def _read_mcp_section() -> dict[str, Any]:
"""运行时读 generation.yaml 的【结构化】mcp 段(genconfig.get 只取标量,故此处直接读原始段)。
复用 genconfig._load_yaml()( mtime 缓存 + best-effort,任何异常都返回空 dict);本函数再做一层
防御:取不到 / dict 返回空 dict绝不抛
"""
try:
data = _genconfig._load_yaml() # 已缓存 + best-effort;读不到/脏 → {}
section = data.get("mcp") if isinstance(data, dict) else None
return section if isinstance(section, dict) else {}
except Exception as exc: # noqa: BLE001 —— 读配置绝不中断主链
_warn_once("mcp-read", f"读 mcp 配置段失败({type(exc).__name__}: {exc})→ 视为未配置(no-op)。")
return {}
def _enabled(section: dict[str, Any]) -> bool:
"""MCP 接入是否开启:env 单点开关 > yaml mcp.enabled;默认 false。
env TIER2_GEN__MCP__ENABLED 优先(便于一次性实验);其次 yaml 段里的 enabled两者都按宽松真值判定,
缺省一律 false(默认关闭,本相红线)
"""
import os
raw = os.environ.get(_ENV_ENABLED)
if raw is None:
raw = section.get("enabled")
if isinstance(raw, bool):
return raw
if raw is None:
return False
# 字符串/数值字面量按常见真值判定(避免 bool('false')==True 的坑)。
return str(raw).strip().lower() in ("1", "true", "yes", "on")
def _server_list(section: dict[str, Any]) -> list[dict]:
"""取 mcp.servers 列表(每项一个 server 配置 dict);非列表 / 空 → 返回 []。"""
servers = section.get("servers")
if not isinstance(servers, list):
return []
# 只保留 dict 项(过滤写错的标量项,best-effort 不抛)。
return [s for s in servers if isinstance(s, dict)]
def _build_one_client(spec: dict, *, allow_stateful: bool):
"""据单个 server 配置 dict 构造一个 MCPClient(惰性 import agentscope.mcp;失败/不合法返回 None)。
Args:
spec: server 配置(name / transport / 传输专属字段 / 可选 stateful·enable_tools·disable_tools·timeout)
allow_stateful: 当前调用路径能否承载 stateful 客户端
同步路径(build_mcp_clients) False:stateful await connect,同步下做不到 跳过该 client;
异步路径(register_mcp_tools) True:由调用方负责后续 connect
Returns:
构造好的 MCPClient(尚未 connect);任何不合法 / 无法在本路径承载 返回 None(并告警)绝不抛
"""
# 惰性 import:6c6g 无 agentscope.mcp;import 失败 → 当作 MCP 不可用,返回 None。
try:
from agentscope.mcp import MCPClient, StdioMCPConfig, HttpMCPConfig
except Exception as exc: # noqa: BLE001 —— 6c6g 等无重依赖环境:不可用即跳过,绝不阻断主链
_warn_once("mcp-import", f"agentscope.mcp 不可用({type(exc).__name__})→ 跳过 MCP 接入(no-op)。")
return None
name = (spec.get("name") or "").strip()
if not name:
_warn_once(f"mcp-noname:{id(spec)}", "某 MCP server 配置缺 name → 跳过该项。")
return None
transport = str(spec.get("transport") or "").strip().lower()
try:
if transport == "stdio":
# STDIO 传输:必 stateful(MCPClient.model_post_init 会强制校验)。同步路径无法 connect → 跳过。
if not allow_stateful:
_warn_once(
f"mcp-stdio-sync:{name}",
f"server '{name}' 是 stdio(必 stateful,需 await connect),同步 build_toolkit 路径无法承载"
" → 跳过;如需用它,请走 async register_mcp_tools 路径(在事件循环内连接)。",
)
return None
command = (spec.get("command") or "").strip()
if not command:
_warn_once(f"mcp-stdio-nocmd:{name}", f"stdio server '{name}' 缺 command → 跳过。")
return None
args = spec.get("args")
env = spec.get("env")
cwd = spec.get("cwd")
cfg = StdioMCPConfig(
command=command,
args=args if isinstance(args, list) else None,
env=env if isinstance(env, dict) else None,
cwd=cwd if isinstance(cwd, str) else None,
)
is_stateful = True # stdio 强制 stateful
elif transport == "http":
url = (spec.get("url") or "").strip()
if not url:
_warn_once(f"mcp-http-nourl:{name}", f"http server '{name}' 缺 url → 跳过。")
return None
headers = spec.get("headers")
timeout = spec.get("timeout")
cfg = HttpMCPConfig(
url=url,
headers=headers if isinstance(headers, dict) else None,
timeout=float(timeout) if isinstance(timeout, (int, float)) else 30.0,
)
# HTTP 默认 stateless(同步路径可用);仅当显式 stateful=true 才标 stateful。
is_stateful = bool(spec.get("stateful")) is True
if is_stateful and not allow_stateful:
_warn_once(
f"mcp-http-stateful-sync:{name}",
f"http server '{name}' 声明 stateful=true(需 await connect),同步 build_toolkit 路径无法承载"
" → 跳过;改 stateful=false 或走 async register_mcp_tools 路径。",
)
return None
else:
_warn_once(
f"mcp-badtransport:{name}",
f"server '{name}' transport='{transport}' 非法(应为 http | stdio)→ 跳过。",
)
return None
# 可选白名单 / 黑名单 / 执行超时(类型不对就忽略,best-effort)。
enable_tools = spec.get("enable_tools")
disable_tools = spec.get("disable_tools")
exec_timeout = spec.get("execution_timeout")
client = MCPClient(
name=name,
is_stateful=is_stateful,
mcp_config=cfg,
enable_tools=enable_tools if isinstance(enable_tools, list) else None,
disable_tools=disable_tools if isinstance(disable_tools, list) else None,
execution_timeout=float(exec_timeout) if isinstance(exec_timeout, (int, float)) else None,
)
return client
except Exception as exc: # noqa: BLE001 —— 单个 client 构造失败不连累其它 / 不阻断主链
_warn_once(
f"mcp-build:{name}",
f"构造 MCPClient '{name}' 失败({type(exc).__name__}: {exc})→ 跳过该项。",
)
return None
def build_mcp_clients(configs: dict[str, Any] | None = None) -> list:
"""【同步】据配置构造可在同步路径(build_toolkit→Toolkit(mcps=...))承载的 MCPClient 列表。
build_toolkit 的门控挂点调用:只构造 connect list_tools(stateless HTTP 客户端无需 connect,
toolkit 在运行时惰性枚举其工具)stateful / stdio 客户端在此被跳过( async connect, register_mcp_tools)
Args:
configs: 直接传入的 mcp 配置段(测试 / 外部注入用); None 则运行时从 generation.yaml mcp 段读
Returns:
list[MCPClient]:可直接喂给 Toolkit(mcps=...) 的客户端列表
默认关闭 / 未配置 / 不可用 / 任意异常 返回 [](此时 toolkit 行为与现在字节级一致)绝不抛
"""
try:
section = configs if isinstance(configs, dict) else _read_mcp_section()
if not _enabled(section):
return [] # 默认关闭:no-op(九工具行为零影响)
servers = _server_list(section)
if not servers:
_warn_once("mcp-noservers", "mcp.enabled=true 但 servers 为空/非列表 → 无 MCP 工具可挂(no-op)。")
return []
clients = []
for spec in servers:
c = _build_one_client(spec, allow_stateful=False) # 同步路径:只收 stateless
if c is not None:
clients.append(c)
if clients:
print(f"[tier2-genconfig] [mcp] 已构造 {len(clients)} 个 stateless MCP 客户端,挂入 toolkit basic 组。",
flush=True)
return clients
except Exception as exc: # noqa: BLE001 —— 整体 best-effort:任何意外都退化为 no-op
_warn_once("mcp-buildall", f"构造 MCP 客户端整体失败({type(exc).__name__}: {exc})→ no-op(九工具不受影响)。")
return []
async def register_mcp_tools(toolkit, *, configs: dict[str, Any] | None = None) -> int:
"""【异步】把配置里的 MCP server 工具注册进【已构造好的】toolkit(覆盖 stateful / stdio 客户端)。
用于已在事件循环内的调用点:构造 client stateful 的先 await connect() append toolkit
"basic" mcps(2.0.2 唯一合法的后置注册路径:toolkit.tool_groups[0].mcps, _tool_group.py:38)
toolkit 会在运行时惰性 list_tools() 枚举它们的工具stateless HTTP 客户端也一并挂入(无需 connect)
build_mcp_clients 的分工:build_mcp_clients 走同步 build_toolkit 路径(只能接 stateless);本函数走
async 路径,额外能连 stateful / stdio两者都默认关闭 + best-effort no-op,九工具行为永不受影响
Args:
toolkit: 已构造的 agentscope.tool.Toolkit 实例
configs: 直接传入的 mcp 配置段; None 则从 generation.yaml mcp 段读
Returns:
int:成功挂入的 MCP 客户端数(默认关闭 / 未配置 / 不可用 0)绝不抛
"""
try:
section = configs if isinstance(configs, dict) else _read_mcp_section()
if not _enabled(section):
return 0 # 默认关闭
servers = _server_list(section)
if not servers:
_warn_once("mcp-noservers", "mcp.enabled=true 但 servers 为空/非列表 → 无 MCP 工具可挂(no-op)。")
return 0
# 定位 toolkit 的 "basic" 组(后置注册的唯一落点);拿不到则放弃(不破坏 toolkit)。
basic_group = None
try:
for g in getattr(toolkit, "tool_groups", []) or []:
if getattr(g, "name", None) == "basic":
basic_group = g
break
except Exception as exc: # noqa: BLE001
_warn_once("mcp-nobasic", f"定位 toolkit basic 组失败({type(exc).__name__})→ 放弃 MCP 注册(no-op)。")
return 0
if basic_group is None or not isinstance(getattr(basic_group, "mcps", None), list):
_warn_once("mcp-nobasic2", "toolkit 无可写的 basic.mcps → 放弃 MCP 注册(no-op,九工具不受影响)。")
return 0
registered = 0
for spec in servers:
client = _build_one_client(spec, allow_stateful=True) # async 路径:可承载 stateful
if client is None:
continue
# stateful 客户端必须先连接(toolkit 惰性 list_tools 时要求已 connect)。连接失败 → 跳过该 client。
try:
if getattr(client, "is_stateful", False):
await client.connect()
except Exception as exc: # noqa: BLE001 —— 单个连接失败不连累其它 / 不阻断主链
_warn_once(
f"mcp-connect:{getattr(client, 'name', '?')}",
f"连接 MCP '{getattr(client, 'name', '?')}' 失败({type(exc).__name__}: {exc})→ 跳过该项。",
)
continue
basic_group.mcps.append(client)
registered += 1
if registered:
print(f"[tier2-genconfig] [mcp] async 注册了 {registered} 个 MCP 客户端进 toolkit basic 组。", flush=True)
return registered
except Exception as exc: # noqa: BLE001 —— 整体 best-effort
_warn_once("mcp-regall", f"register_mcp_tools 整体失败({type(exc).__name__}: {exc})→ no-op(九工具不受影响)。")
return 0

View File

@ -376,4 +376,21 @@ def build_toolkit(session: Tier2Session) -> Toolkit:
FunctionTool(query_asset),
FunctionTool(finish),
]
return Toolkit(tools=tools)
# ── 加性挂点(图说 E 族 · 可选 MCP 工具接入)── 默认关闭,九工具行为零改 ──
# 仅当 generation.yaml 的 mcp 段 enabled=true(或 env TIER2_GEN__MCP__ENABLED)且 servers 非空时,
# build_mcp_clients 才返回若干【stateless】MCP 客户端,经 Toolkit(mcps=...) 落进保留的 "basic" 组,
# 让 agent 在九工具之外多一组外部工具(2.0.2 MCP 接入的唯一构造姿势,见 mcp_tools.py 头部 API 核对)。
# 默认 / 未配置 / 任何异常 → build_mcp_clients 返回 [] → Toolkit(tools=tools, mcps=[]) 与原
# Toolkit(tools=tools) 字节级一致(构造器 mcps or [] → 同为空 list),九工具(spike 已验证主链)绝不受影响。
# stateful / stdio 客户端需 await connect(),不能在此同步路径承载——走 mcp_tools.register_mcp_tools(async)。
# 惰性 import:6c6g 无 agentscope.mcp,故在挂点处函数内 import(import 失败即 no-op,不阻断)。
mcp_clients: list = []
try:
from .mcp_tools import build_mcp_clients
mcp_clients = build_mcp_clients() # 默认关闭 → []
except Exception as exc: # noqa: BLE001 —— MCP 接入是加性增强,任何失败都退化为「不挂 MCP」,不连累九工具
print(f"[tier2-toolkit] MCP 接入挂点异常({type(exc).__name__}: {exc})→ 仅装九工具(无 MCP)。", flush=True)
mcp_clients = []
return Toolkit(tools=tools, mcps=mcp_clients)