675a8e688b
WU-A 迁移面 spike:便宜档生成核心(多轮 ReAct + 五工具沙箱 + done 门)用 Python 在新建 cheap-worker/ 重写成 AgentScope agent,import 复用 tier2 框架层(model client/四道熔断/历史压缩/observability), check/build/九门 shell-out 现有 node/cjs(不重写 esbuild/CDP/形状门)。 - U1 地基:跨包 sys.path import tier2 worker + 代理旁路 + key 从内网文档注入 env + M3 装配 - U2 五工具:read/list/write Python 实现 + check/build shell-out(形状门三态与 Node 一致) - U3 LittleJS system prompt:prompt.mjs Python 化(两条 footgun 正反例 + 输入契约) - U4 scaffold/stage/play 薄壳 + ensure_play_spec(据 state 形态产 play-spec 治链路假绿) - U5 ReAct 主编排:内置 Agent+ReActConfig+外层有界 resume + done 门 + 收口 stage/smoke/九门 - U6 对照 Node:产物形态 + 九门逐门 + 过门率对比 验收:同 brief Python 新路 vs Node 旧路,产物都真 src/ 6 文件、九门都 9/9 全 PASS、过门率 1.0=1.0 → spikePass=True。端到端单局 attempt 1 一次收敛、148s。落点 cheap-worker/ 对 tier2/amodel-gen 零改动(结构性零回归)。测试 38/38 全绿。 origin: docs/plans/2026-06-26-001-feat-cheap-worker-python-spike-plan.md Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
"""
|
||
test_bootstrap.py — U1 地基验证：跨包 import + 代理旁路 + key 注入 + M3 model 装配。
|
||
|
||
须用 venv python 跑（import agentscope）：
|
||
cheap-worker/.venv/bin/python cheap-worker/tests/test_bootstrap.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/
|
||
import _bootstrap
|
||
|
||
|
||
def test_cross_package_import():
|
||
"""加 sys.path 后 from worker import config, client 不抛 ImportError，且拿到关键 API。"""
|
||
fw = _bootstrap.load_framework()
|
||
assert {"config", "client", "middleware"} <= set(fw)
|
||
from worker import config, client # noqa: E402
|
||
assert hasattr(config, "build_model_openai")
|
||
assert hasattr(client, "install_proxy_bypass")
|
||
assert hasattr(client, "get_api_key")
|
||
|
||
|
||
def test_proxy_bypass():
|
||
"""install_proxy_bypass 把内网网关 host 并入 NO_PROXY / no_proxy 两个 env。"""
|
||
from worker import client # noqa: E402
|
||
host = client.install_proxy_bypass("http://100.64.0.8:3000")
|
||
assert host == "100.64.0.8"
|
||
assert "100.64.0.8" in os.environ.get("NO_PROXY", "")
|
||
assert "100.64.0.8" in os.environ.get("no_proxy", "")
|
||
|
||
|
||
def test_api_key_from_doc():
|
||
"""key 缺 env 时从 docs/内网凭据与端点.md 解析注入（内网单一事实源）。"""
|
||
# 清掉 env 模拟"未设"，强制走文档解析路径
|
||
os.environ.pop("NEWAPI_KEY", None)
|
||
key = _bootstrap.ensure_api_key_env()
|
||
assert key.startswith("sk-")
|
||
assert os.environ.get("NEWAPI_KEY", "").startswith("sk-")
|
||
|
||
|
||
def test_model_assembly():
|
||
"""build_cheap_model 返回 OpenAIChatModel（base_url 带 /v1 由 build_model_openai 保证），model 名为 MiniMax-M3。"""
|
||
model = _bootstrap.build_cheap_model(max_tokens=8000)
|
||
from agentscope.model import OpenAIChatModel # noqa: E402
|
||
assert isinstance(model, OpenAIChatModel)
|
||
# 模型名固定 M3（KTD6）。属性名在不同小版本可能差异，宽松双路断言。
|
||
assert getattr(model, "model", None) == _bootstrap.SPIKE_MODEL or _bootstrap.SPIKE_MODEL in repr(model)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
||
_failed = 0
|
||
for _fn in _fns:
|
||
try:
|
||
_fn()
|
||
print(f" PASS {_fn.__name__}")
|
||
except Exception as e: # noqa: BLE001 地基测试要看到任何异常类型
|
||
_failed += 1
|
||
print(f" FAIL {_fn.__name__}: {type(e).__name__}: {e}")
|
||
print(f"\n{len(_fns) - _failed}/{len(_fns)} passed")
|
||
sys.exit(1 if _failed else 0)
|