lambdagent/demo/demo1_type_check.py

#!/usr/bin/env python3
"""
Demo 1: 扫描鸿蒙 Agent DSL 管道配置，发现类型不兼容 bug
========================================================
流程:
  1. 加载鸿蒙旅行规划 YAML 配置（模拟 Agent DSL 导出）
  2. 从 chain.steps 提取每个 Agent 的 output_type / input_type
  3. 用 T-Compose 规则检查相邻 Agent 的类型兼容性
  4. 用 lint_config 扫描 26 条结构规则

核心发现:
  Planner 输出 Json(object{route, budget, days})
  Transport 期望 Str
  → 仓颉编译器看到的都是 String，不报错
  → lambdagent T-Compose: Json(object) ≮: Str → 编译时报错
"""

from __future__ import annotations
import sys, os, yaml
from typing import Optional
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from lambdagent.types import (
    AgentType, LamType, TypeTag,
    check_compose_types, AgentTypeError,
    is_subtype, T_ANY, T_STR,
)
from lambdagent.fromconfig.lint import lint_config


def json_schema_to_lam_type(schema: Optional[dict]) -> LamType:
    """把 YAML 中的 JSON Schema 类型标注转为 LamType"""
    if schema is None:
        return T_ANY
    t = schema.get("type", "any")
    if t == "string":
        return T_STR
    elif t in ("object", "array"):
        return LamType(TypeTag.JSON, schema=schema)
    elif t == "integer":
        return LamType(TypeTag.INT)
    elif t == "number":
        return LamType(TypeTag.FLOAT)
    elif t == "boolean":
        return LamType(TypeTag.BOOL)
    return T_ANY


def main():
    config_path = os.path.join(os.path.dirname(__file__), "harmony_travel.yaml")

    print("=" * 62)
    print("Demo 1: 扫描鸿蒙 Agent DSL 管道 — T-Compose 类型检查")
    print("=" * 62)
    print()

    # ── Step 1: 加载配置 ──
    print(f"[Step 1] 加载配置: {os.path.basename(config_path)}")
    with open(config_path, "r", encoding="utf-8") as f:
        cfg = yaml.safe_load(f)

    steps = cfg.get("chain", {}).get("steps", [])
    print(f"  Pipeline: {' → '.join(s['name'] for s in steps)}")
    print(f"  模型: {cfg.get('model', {}).get('name', 'unknown')}")
    print()

    # ── Step 2: 从配置提取 Agent 类型签名 ──
    print("[Step 2] 从 YAML 提取 Agent 类型签名")
    agent_types = []
    for step in steps:
        name = step["name"]
        in_t = json_schema_to_lam_type(step.get("input_type"))
        out_t = json_schema_to_lam_type(step.get("output_type"))
        at = AgentType(input_type=in_t, output_type=out_t, effect="llm")
        agent_types.append((name, at))
        print(f"  {name:12s}: {at}")
    print()

    # ── Step 3: CangjieMagic / 仓颉编译器视角 ──
    print("[Step 3] CangjieMagic / 仓颉编译器视角")
    print()
    print("  // CangjieMagic Agent DSL 代码:")
    print('  @agent[model: "pangu-lite", executor: "naive"]')
    print("  class Planner {")
    print('    @prompt[pattern: APE](')
    print('      action: "制定旅行计划",')
    print('      expectation: "JSON格式行程")')
    print("    func planTrip(req: String): String  // ← 仓颉类型")
    print("  }")
    print()
    print('  @agent[model: "pangu-lite"]')
    print("  class Transport {")
    print('    @prompt[pattern: APE](action: "预订交通")')
    print("    func book(plan: String): String      // ← 仓颉类型")
    print("  }")
    print()
    print("  // 仓颉编译器: String → String → String ✅ 全部通过")
    print("  // → 仓颉/CangjieMagic 只看到 String，不检查 JSON 结构")
    print()

    # ── Step 4: T-Compose 类型检查 ──
    print("[Step 4] lambdagent T-Compose 类型检查")
    print("  规则: f >> g 要求 output(f) <: input(g)")
    print()

    for i in range(len(agent_types) - 1):
        name_f, type_f = agent_types[i]
        name_g, type_g = agent_types[i + 1]
        out_f = type_f.output_type
        in_g = type_g.input_type
        ok = is_subtype(out_f, in_g)
        status = "✅" if ok else "❌"
        print(f"  {name_f} >> {name_g}:")
        print(f"    output({name_f}) = {out_f}")
        print(f"    input({name_g})  = {in_g}")
        print(f"    {out_f} <: {in_g} ?  → {status}")
        print()

    # 整体检查
    try:
        result = check_compose_types([at for _, at in agent_types])
        print(f"  ✅ Pipeline 类型安全: {result}")
    except AgentTypeError as e:
        print(f"  ❌ [T-Compose ERROR] {e}")
        print()
        print("  影响分析:")
        print("    → Transport 收到 JSON 而非城市名 → 解析失败")
        print("    → 端侧: 用户看到白屏 → 卸载 App")
        print("    → 服务端: 可以重试；手机上: 没有第二次机会")
    print()

    # ── Step 5: Lint 结构扫描 ──
    print("[Step 5] Lint 26 条规则扫描")
    results = lint_config(cfg)
    if results:
        for r in results:
            print(f"  [{r.level:5s}] {r.rule}: {r.message}")
    else:
        print("  ✅ No lint issues")
    print()

    print("=" * 62)
    print("结论:")
    print("  仓颉编译器: String → String → String ✅ (漏掉类型 bug)")
    print("  lambdagent:  Json(obj) → Str ❌ (编译时发现)")
    print("  → 合作方案: 把 T-Compose 集成到仓颉编译器")
    print("=" * 62)


if __name__ == "__main__":
    main()