Agent/knowhub/skill/cli.py

#!/usr/bin/env python3
"""
KnowHub CLI - 知识管理命令行工具

使用方法:
    python -m knowhub.skill.cli search "查询内容"
    python -m knowhub.skill.cli save --task "任务" --content "内容" --types strategy
    python -m knowhub.skill.cli list --limit 10
"""

import os
import sys
import json
import argparse
from pathlib import Path

try:
    import httpx
except ImportError:
    print("错误: 需要安装 httpx 库")
    print("运行: pip install httpx")
    sys.exit(1)

# KnowHub API 默认地址（CLI 需要在开头声明，方便用户查看和修改）
DEFAULT_API_URL = "http://43.106.118.91:9999"


def get_api_base() -> str:
    """获取 API 地址"""
    return os.getenv("KNOWHUB_API", DEFAULT_API_URL)


def search_knowledge(args):
    """搜索知识"""
    url = f"{get_api_base()}/api/knowledge/search"
    params = {
        "q": args.query,
        "top_k": args.top_k,
        "min_score": args.min_score,
    }
    if args.types:
        params["types"] = args.types

    try:
        response = httpx.get(url, params=params, timeout=30.0)
        response.raise_for_status()
        data = response.json()

        if data["count"] == 0:
            print("未找到相关知识")
            return

        print(f"找到 {data['count']} 条知识:\n")
        for i, item in enumerate(data["results"], 1):
            print(f"[{i}] {item['task']}")
            print(f"    ID: {item['id']}")
            eval_data = item.get("eval", {})
            print(f"    评分: {eval_data.get('score', 3)} | 质量分: {item.get('quality_score', 'N/A')}")
            print(f"    类型: {', '.join(item.get('types', []))}")
            print(f"    内容: {item['content'][:100]}...")
            print()

    except httpx.HTTPError as e:
        print(f"请求失败: {e}")
        sys.exit(1)


def save_knowledge(args):
    """保存知识"""
    url = f"{get_api_base()}/api/knowledge"

    data = {
        "message_id": args.message_id or f"cli-{os.getpid()}",
        "types": args.types.split(",") if args.types else ["strategy"],
        "task": args.task,
        "tags": json.loads(args.tags) if args.tags else {},
        "scopes": args.scopes.split(",") if args.scopes else ["org:cybertogether"],
        "owner": args.owner or "agent:cli",
        "content": args.content,
        "source": {
            "name": args.source_name or "cli",
            "category": args.source_category or "exp",
            "urls": args.urls.split(",") if args.urls else [],
            "agent_id": args.agent_id or "cli",
            "submitted_by": args.submitted_by or "cli-user",
        },
        "eval": {
            "score": args.score,
            "helpful": 1,
            "harmful": 0,
            "confidence": 0.5,
        }
    }

    try:
        response = httpx.post(url, json=data, timeout=30.0)
        response.raise_for_status()
        result = response.json()
        print(f"✅ 知识已保存: {result['knowledge_id']}")

    except httpx.HTTPError as e:
        print(f"保存失败: {e}")
        sys.exit(1)


def update_knowledge(args):
    """更新知识"""
    url = f"{get_api_base()}/api/knowledge/{args.id}"

    data = {}
    if args.score:
        data["update_score"] = args.score
    if args.helpful_case:
        data["add_helpful_case"] = args.helpful_case
    if args.harmful_case:
        data["add_harmful_case"] = args.harmful_case
    if args.evolve_feedback:
        data["evolve_feedback"] = args.evolve_feedback

    if not data:
        print("错误: 至少需要提供一个更新参数")
        sys.exit(1)

    try:
        response = httpx.put(url, json=data, timeout=30.0)
        response.raise_for_status()
        print(f"✅ 知识已更新: {args.id}")

    except httpx.HTTPError as e:
        print(f"更新失败: {e}")
        sys.exit(1)


def batch_update_knowledge(args):
    """批量更新知识"""
    url = f"{get_api_base()}/api/knowledge/batch_update"

    # 从文件读取反馈列表
    if args.file:
        with open(args.file, 'r') as f:
            feedback_list = json.load(f)
    else:
        print("错误: 需要提供 --file 参数")
        sys.exit(1)

    data = {"feedback_list": feedback_list}

    try:
        response = httpx.post(url, json=data, timeout=60.0)
        response.raise_for_status()
        result = response.json()
        print(f"✅ 批量更新完成: {result['updated']} 条知识")

    except httpx.HTTPError as e:
        print(f"批量更新失败: {e}")
        sys.exit(1)


def list_knowledge(args):
    """列出知识"""
    url = f"{get_api_base()}/api/knowledge"
    params = {"limit": args.limit}
    if args.types:
        params["types"] = args.types
    if args.scopes:
        params["scopes"] = args.scopes

    try:
        response = httpx.get(url, params=params, timeout=30.0)
        response.raise_for_status()
        data = response.json()

        if data["count"] == 0:
            print("知识库为空")
            return

        print(f"共 {data['count']} 条知识:\n")
        for i, item in enumerate(data["results"], 1):
            print(f"[{i}] {item['task']}")
            print(f"    ID: {item['id']}")
            eval_data = item.get("eval", {})
            print(f"    评分: {eval_data.get('score', 3)} | Helpful: {eval_data.get('helpful', 0)} | Harmful: {eval_data.get('harmful', 0)}")
            print(f"    类型: {', '.join(item.get('types', []))}")
            print(f"    所有者: {item.get('owner', 'N/A')}")
            print()

    except httpx.HTTPError as e:
        print(f"请求失败: {e}")
        sys.exit(1)


def slim_knowledge(args):
    """知识瘦身"""
    url = f"{get_api_base()}/api/knowledge/slim"
    params = {"model": args.model}

    try:
        print("正在执行知识瘦身，这可能需要一些时间...")
        response = httpx.post(url, params=params, timeout=120.0)
        response.raise_for_status()
        result = response.json()
        print(f"✅ 瘦身完成: {result['before']} → {result['after']} 条知识")
        if result.get("report"):
            print(f"   {result['report']}")

    except httpx.HTTPError as e:
        print(f"瘦身失败: {e}")
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description="KnowHub CLI - 知识管理工具")
    subparsers = parser.add_subparsers(dest="command", help="可用命令")

    # search 命令
    search_parser = subparsers.add_parser("search", help="搜索知识")
    search_parser.add_argument("query", help="查询文本")
    search_parser.add_argument("--top-k", type=int, default=5, help="返回结果数量")
    search_parser.add_argument("--min-score", type=int, default=3, help="最低评分")
    search_parser.add_argument("--types", help="类型过滤（逗号分隔）")

    # save 命令
    save_parser = subparsers.add_parser("save", help="保存知识")
    save_parser.add_argument("--task", required=True, help="任务描述")
    save_parser.add_argument("--content", required=True, help="知识内容")
    save_parser.add_argument("--types", default="strategy", help="类型（逗号分隔）")
    save_parser.add_argument("--tags", help="标签（JSON 格式）")
    save_parser.add_argument("--scopes", help="可见范围（逗号分隔）")
    save_parser.add_argument("--owner", help="所有者")
    save_parser.add_argument("--source-name", help="来源名称")
    save_parser.add_argument("--source-category", help="来源类别")
    save_parser.add_argument("--urls", help="相关 URL（逗号分隔）")
    save_parser.add_argument("--agent-id", help="Agent ID")
    save_parser.add_argument("--submitted-by", help="提交者")
    save_parser.add_argument("--message-id", help="消息 ID")
    save_parser.add_argument("--score", type=int, default=3, help="评分 (1-5)")

    # update 命令
    update_parser = subparsers.add_parser("update", help="更新知识")
    update_parser.add_argument("id", help="知识 ID")
    update_parser.add_argument("--score", type=int, help="更新评分")
    update_parser.add_argument("--helpful-case", help="添加有效案例")
    update_parser.add_argument("--harmful-case", help="添加有害案例")
    update_parser.add_argument("--evolve-feedback", help="知识进化反馈")

    # batch-update 命令
    batch_parser = subparsers.add_parser("batch-update", help="批量更新知识")
    batch_parser.add_argument("--file", required=True, help="反馈列表 JSON 文件")

    # list 命令
    list_parser = subparsers.add_parser("list", help="列出知识")
    list_parser.add_argument("--limit", type=int, default=10, help="返回数量")
    list_parser.add_argument("--types", help="类型过滤")
    list_parser.add_argument("--scopes", help="范围过滤")

    # slim 命令
    slim_parser = subparsers.add_parser("slim", help="知识瘦身")
    slim_parser.add_argument("--model", default="google/gemini-2.0-flash-001", help="使用的模型")

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    # 执行命令
    if args.command == "search":
        search_knowledge(args)
    elif args.command == "save":
        save_knowledge(args)
    elif args.command == "update":
        update_knowledge(args)
    elif args.command == "batch-update":
        batch_update_knowledge(args)
    elif args.command == "list":
        list_knowledge(args)
    elif args.command == "slim":
        slim_knowledge(args)


if __name__ == "__main__":
    main()