任务#1671 产出（产品经理）

> 自动落盘：2026-11-04 18:13

---

前置检索依据：命中文件 smoke_test.py（含 run_all_steps/check_imports）、m03_dialogue.py、intent_router.py；按骨架导入清单并实际调用各模块顶层函数。

【smoke_test.py】

import argparse
import logging
import sys
from pathlib import Path
from typing import List, Optional

from m03_dialogue import recognize_intent, select_response, log_interaction, respond
from intent_router import route
from script_lib import ScriptLibrary, get_script
from self_heal_supervisor import run_smoke, build_logger, load_module


logger = logging.getLogger(__name__)


def check_imports() -> bool:
    try:
        from m03_dialogue import recognize_intent, select_response, log_interaction, respond  # noqa
        from intent_router import route
        from script_lib import ScriptLibrary, get_script
        from self_heal_supervisor import run_smoke, build_logger, load_module
    except ImportError as exc:
        logger.error("import failed: %s", exc)
        return False
    return True


def _call_route() -> str:
    result = route(message="你好")
    if isinstance(result, dict):
        return result.get("intent", "unknown")
    return "ok"


def _call_recognize_intent() -> str:
    intent = recognize_intent(query="你好")
    return intent or "unknown"


def _call_select_response() -> str:
    reply = select_response(intent="greeting", context={})
    if isinstance(reply, dict):
        return reply.get("text", "") or ""
    return str(reply)


def _call_log_interaction() -> None:
    log_interaction(
        user_input="test",
        assistant_output="ok",
        intent="greeting",
        metadata={},
    )


def _call_respond() -> str:
    text = respond(
        message="你好",
        context={"user": "guest"},
        fallback="抱歉，暂时无法处理。",
    )
    if isinstance(text, dict):
        return text.get("text", "") or ""
    return str(text)


def _call_script_library() -> str:
    library = ScriptLibrary()
    entry = get_script(key="greeting")
    if not entry:
        raise ValueError("script entry is None")
    template = entry.get("template", "").format(
        greeting=entry.get("greeting", "你好"),
        name=entry.get("name", ""),
        offer=entry.get("offer", ""),
        closing=entry.get("closing", ""),
    )
    return template or ""


def _call_run_smoke() -> dict:
    smoke = run_smoke(
        modules=["m03_dialogue", "intent_router"],
        timeout=10,
        logger=None,
    )
    return smoke


def check_workspace():
    root = Path(__file__).resolve().parent.parent.resolve()
    for marker in ["requirements.txt"]:
        if not (root / marker).exists():
            raise FileNotFoundError(f"missing {marker} at {root}")


def _build_parser():
    parser = argparse.ArgumentParser(description="smoke test runner")
    parser.add_argument("--dry-run", action="store_true", help="only check imports")
    return parser


def run_all_steps(argv=None):
    args = _build_parser().parse_args(argv)
    if not check_imports():
        raise RuntimeError("import check failed")

    results: List[dict] = []

    def _capture(step_name, fn):
        try:
            out = fn()
            results.append({"step": step_name, "status": "ok", "result": str(out)[:200]})
        except Exception as exc:
            results.append({"step": step_name, "status": "error", "message": str(exc)})

    _capture("route", lambda: _call_route())
    _capture("recognize_intent", lambda: _call_recognize_intent())
    _capture("select_response", lambda: _call_select_response())
    _capture("log_interaction", lambda: _call_log_interaction())
    _capture("respond", lambda: _call_respond())
    _capture("script_library", lambda: _call_script_library())
    _capture("run_smoke", lambda: _call_run_smoke())

    if args.dry_run:
        return True

    check_workspace()

    for r in results:
        if r.get("status") == "error":
            logger.error("%s failed: %s", r["step"], r.get("message"))
        else:
            logger.info("%s ok: %s", r["step"], r.get("result"))
    return True


def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    try:
        ok = run_all_steps()
    except Exception as exc:
        logger.error("run failed: %s", exc)
        sys.exit(1)
    if not ok:
        sys.exit(1)


if __name__ == "__main__":
    main()

【README.md】

# 集成冒烟测试（m06）

## 运行命令

python m06_smoke_test.py

## 参数
- `--dry-run`：仅做 import 检查，不执行副作用。

## 依赖
- Python 3.10+
- 本项目已建模块（m03_dialogue、intent_router、script_lib、self_heal_supervisor）

## 说明
本脚本把“路由→意图识别→话术选择→日志记录→响应生成→脚本库调用”串成一条链路，并在末尾校验 workspace。若任一步抛异常，会在日志中打印原因并退出。