lambdagent/agentexample/agent67v2/run.py

#!/usr/bin/env python3
"""
agent67v2/run.py — Lambda v2 多智能体个人助理 🐑
=================================================

v2 架构: 协调者 + 5个专业微代理
  协调者: 轻量级 ReAct, 只做任务分析和分派
  微代理: code / shell / web / memory / system

三大改进:
  1. 拆分: 42工具 → 5个微代理, 每个代理工具少 → 响应快
  2. 复用: 每个微代理注册为 Skill, 可被任何编排器调用
  3. 懒加载: ToolSearch 元工具按需发现, 首次响应快 2-3x

用法:
  python run.py              # 使用 Claude Code (默认)
  python run.py --api        # 使用 API Key
  python run.py --model opus # 指定模型
  python run.py --verbose    # 详细模式
"""
from __future__ import annotations

import argparse
import os
import sys
from pathlib import Path

# 路径设置
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
AGENT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(PROJECT_ROOT))
sys.path.insert(0, str(AGENT_DIR.parent))

BANNER = """
╔═══════════════════════════════════════════════════╗
║  🐑 lambda v2 — 多智能体协作版个人助手            ║
║                                                   ║
║  架构: Coordinator + 5 Micro-Agents               ║
║  ┌──────────┐                                     ║
║  │Coordinator│─→ code-agent   (文件/代码/Git)     ║
║  │ (分析+    │─→ shell-agent  (终端命令)          ║
║  │  分派)    │─→ web-agent    (搜索/知识库)       ║
║  │          │─→ memory-agent (记忆/任务)          ║
║  │          │─→ system-agent (系统控制)           ║
║  └──────────┘                                     ║
║                                                   ║
║  命令: trace | stats | history | exit              ║
╚═══════════════════════════════════════════════════╝
"""


def main():
    parser = argparse.ArgumentParser(description="🐑 lambda v2 — 多智能体个人助理")
    parser.add_argument("--api", action="store_true",
                        help="使用 API Key 模式")
    parser.add_argument("--ollama", action="store_true",
                        help="使用 Ollama 本地模型")
    parser.add_argument("--model", type=str, default=None,
                        help="指定模型 (默认: 自动检测)")
    parser.add_argument("--verbose", action="store_true",
                        help="详细模式")
    args = parser.parse_args()

    print(BANNER)

    # 检测后端 (与 agent67 一致的优先级)
    try:
        from agent67.core.config import detect_backend, BACKEND_OLLAMA
        model, use_api, backend = detect_backend()
    except ImportError:
        import shutil
        if shutil.which("claude"):
            model, use_api, backend = "sonnet", False, "claude_code"
            print("✅ 使用 Claude Code Max Plan (无需 API Key)")
        elif os.environ.get("ANTHROPIC_API_KEY"):
            model, use_api, backend = "claude-sonnet-4-20250514", True, "api"
            print("✅ 使用 Anthropic API")
        else:
            model, use_api, backend = args.model or "sonnet", args.api, ""
            print("⚠️  未检测到 LLM 后端，将使用标准 Lam")

    if args.ollama:
        backend = "ollama"
        model = args.model or "qwen2.5:7b"
        use_api = False
    elif args.api:
        use_api = True
        backend = "api"
    if args.model:
        model = args.model

    # 创建协调者
    from agent67v2.core.coordinator import CoordinatorAssistant

    print("  🔄 初始化微代理注册表...")
    assistant = CoordinatorAssistant(
        model=model or "sonnet",
        use_api=use_api,
        backend=backend,
        verbose=args.verbose,
    )
    print("  ✅ 5 个微代理已就绪\n")

    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n👋 lambda v2 下线了，再见！")
            break

        if not user_input:
            continue

        if user_input.lower() in ("exit", "quit", "bye", "再见"):
            print("👋 lambda v2 下线了，祝你有美好的一天！")
            break

        if user_input.lower() == "trace":
            assistant.print_trace()
            continue

        if user_input.lower() == "stats":
            assistant.print_stats()
            continue

        if user_input.lower() == "history":
            print("\n📜 对话历史:")
            for entry in assistant.conversation_history:
                print(f"  [{entry['role']}] {entry['content'][:100]}")
            print()
            continue

        if user_input.lower() == "skills":
            from lambdagent.skills import SkillRegistry
            registry = SkillRegistry()
            print(f"\n📦 已注册技能 ({len(registry)} 个):")
            for name in registry.list_all():
                skill = registry.get(name)
                print(f"  • {name}: {skill.description}")
                print(f"    tags: {skill.tags}")
                print(f"    stats: {skill.stats}")
            print()
            continue

        print()
        response = assistant.chat(user_input)
        is_streaming = hasattr(assistant.brain, 'stream') and assistant.brain.stream
        if not is_streaming:
            print(f"🐑 lambda v2: {response}")
        print()


if __name__ == "__main__":
    main()