【LLM应用开发原理】Agent Skill原理与基础

# 工具时代：手写每一步
search_result = search_api("量子计算最新进展")
summary = llm.summarize(search_result)
translation = llm.translate(summary, "zh")
save_to_notion(translation)

# 技能时代：封装为可复用单元
@skill
def research_and_translate(topic: str, lang: str):
    """研究某个主题并翻译为指定语言"""
    search_result = search_api(topic)
    summary = llm.summarize(search_result)
    translation = llm.translate(summary, lang)
    save_to_notion(translation)
    return translation

class SkillTreeNode:
    def __init__(self, name: str, skills: List[Skill] = None, children: List['SkillTreeNode'] = None):
        self.name = name
        self.skills = skills or []
        self.children = children or []

# 构建技能树
root = SkillTreeNode("根节点", skills=[web_search, read_file])
root.children = [
    SkillTreeNode("内容创作", skills=[write_blog, edit_text]),
    SkillTreeNode("数据分析", skills=[query_db, generate_chart]),
    SkillTreeNode("系统操作", skills=[write_file, run_command], is_sensitive=True)
]

# 渐进式披露
async def progressive_reveal(task: str, node: SkillTreeNode) -> List[Skill]:
    """根据任务逐步披露技能"""
    revealed = node.skills.copy()

    # 检查是否需要展开子节点
    for child in node.children:
        if child.is_sensitive and not await user_confirm(f"需要{child.name}权限，是否继续？"):
            continue

        if is_relevant(task, child.name):
            revealed.extend(await progressive_reveal(task, child))

    return revealed

class IntentRouter:
    def __init__(self):
        self.domain_skills = {
            "research": [web_search, summarize, extract_info],
            "writing": [outline, draft, revise],
            "coding": [read_code, write_code, run_tests],
            "analysis": [load_data, visualize, stats_test]
        }

    async def route(self, user_input: str) -> Tuple[str, List[Skill]]:
        """分析意图并返回对应技能"""
        # 1. 轻量级意图分类（使用小模型或快速提示）
        intent = await self._classify_intent(user_input)

        # 2. 返回对应领域的技能
        return intent, self.domain_skills.get(intent, [])

    async def _classify_intent(self, text: str) -> str:
        """使用小型模型进行快速分类"""
        prompt = f"""
        将以下用户请求分类到一个领域:
        {list(self.domain_skills.keys())}

        用户请求: {text}

        只返回领域名称:
        """
        return await small_llm.ask(prompt)

class DynamicSkillRegistry:
    def __init__(self):
        self.core_skills = load_core_skills()
        self.all_skills_index = build_skill_index()  # 所有技能的元数据索引
        self.loaded_skills = {s.name: s for s in self.core_skills}

    async def get_relevant_skills(self, context: List[Message]) -> List[Skill]:
        """根据上下文动态加载相关技能"""
        # 1. 分析当前上下文需要什么能力
        needed = await self._analyze_needs(context)

        # 2. 查找相关但未加载的技能
        to_load = [
            skill_id for skill_id in needed
            if skill_id not in self.loaded_skills
        ]

        # 3. 动态加载
        for skill_id in to_load:
            skill = await self._load_skill(skill_id)
            self.loaded_skills[skill.name] = skill

        # 4. 返回所有相关的已加载技能
        return [
            self.loaded_skills[skill_id]
            for skill_id in needed
        ]

    async def _analyze_needs(self, context: List[Message]) -> List[str]:
        """分析上下文需要的技能"""
        # 使用LLM分析需要什么能力
        prompt = f"""
        基于以下对话历史，分析完成任务可能需要哪些技能能力。

        对话历史:
        {context}

        可用技能目录:
        {self.all_skills_index}

        返回技能ID列表（JSON格式）:
        """
        return parse_json(await llm.ask(prompt))

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable
from enum import Enum

class SkillType(Enum):
    ATOMIC = "atomic"        # 原子技能
    COMPOSITE = "composite"  # 组合技能
    META = "meta"            # 元技能

@dataclass
class SkillParameter:
    name: str
    type: str
    description: str
    required: bool = True
    default: Any = None
    enum_values: List[Any] = field(default_factory=list)

@dataclass
class Skill:
    name: str                          # 技能名称（唯一标识）
    description: str                   # 技能描述（供LLM理解何时使用）
    version: str = "1.0.0"            # 版本号，用于升级和兼容性
    skill_type: SkillType = SkillType.ATOMIC
    input_schema: List[SkillParameter] = field(default_factory=list)
    output_schema: Dict[str, str] = field(default_factory=dict)
    dependencies: List[str] = field(default_factory=list)  # 依赖的其他技能
    categories: List[str] = field(default_factory=list)    # 用于分类和发现
    tags: List[str] = field(default_factory=list)          # 用于搜索和过滤
    is_sensitive: bool = False        # 是否需要权限确认
    timeout: int = 300                  # 超时时间（秒）
    max_retries: int = 2               # 最大重试次数
    _execute: Optional[Callable] = field(default=None, repr=False)

    def execute(self, **kwargs) -> Any:
        """执行技能"""
        if self._execute:
            return self._execute(**kwargs)
        raise NotImplementedError("Execute method not implemented")

    def to_openai_function(self) -> Dict:
        """转换为OpenAI Function Calling格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        p.name: {
                            "type": p.type,
                            "description": p.description,
                        }
                        for p in self.input_schema
                    },
                    "required": [p.name for p in self.input_schema if p.required]
                }
            }
        }

@atomic_skill(
    name="web_search",
    description="使用搜索引擎查询网络信息",
    categories=["information", "search"],
    timeout=30
)
def web_search(query: str, num_results: int = 5, time_range: str = "1y") -> List[dict]:
    """
    网络搜索

    Args:
        query: 搜索关键词
        num_results: 返回结果数量 (1-10)
        time_range: 时间范围 ("1d", "1w", "1m", "1y", "all")

    Returns:
        搜索结果列表，每个结果包含title, url, snippet
    """
    client = get_search_client()
    return client.search(
        query=query,
        num_results=num_results,
        time_range=time_range
    )

@atomic_skill
def read_file(path: str, encoding: str = "utf-8") -> str:
    """读取文件"""
    with open(path, encoding=encoding) as f:
        return f.read()

@atomic_skill(is_sensitive=True)
def write_file(path: str, content: str, encoding: str = "utf-8") -> None:
    """写入文件（敏感操作）"""
    with open(path, 'w', encoding=encoding) as f:
        f.write(content)

@composite_skill(
    name="research_topic",
    description="深度研究某个主题，生成综合报告",
    categories=["research", "analysis"],
    dependencies=["web_search", "fetch_content", "extract_key_points", "generate_report"]
)
def research_topic(
    topic: str,
    depth: str = "medium",  # "shallow", "medium", "deep"
    output_format: str = "markdown"
) -> Report:
    """
    深度研究某个主题

    执行流程:
    1. 搜索相关信息
    2. 并行抓取内容
    3. 提取关键信息
    4. 生成最终报告
    """
    # 配置参数
    num_sources = {
        "shallow": 3,
        "medium": 7,
        "deep": 15
    }[depth]

    # 步骤1：搜索
    logger.info(f"搜索关于 '{topic}' 的信息...")
    results = web_search(topic, num_results=num_sources * 2)

    # 步骤2：并行抓取内容
    logger.info("抓取网页内容...")
    contents = parallel_execute(
        [fetch_content(r['url']) for r in results[:num_sources]],
        max_workers=5
    )
    # 过滤掉抓取失败的内容
    valid_contents = [c for c in contents if c is not None]

    # 步骤3：提取关键信息
    logger.info("提取关键信息...")
    key_points = extract_key_points(
        valid_contents,
        focus_topics=[topic]
    )

    # 步骤4：生成报告
    logger.info("生成报告...")
    report = generate_report(
        key_points,
        topic=topic,
        depth=depth,
        format=output_format
    )

    return report

@meta_skill(
    name="skill_router",
    description="根据任务描述选择最合适的技能",
    categories=["meta", "orchestration"]
)
def skill_router(
    task: str,
    available_skills: List[Skill],
    context: List[Message] = None
) -> SkillSelection:
    """
    根据任务选择合适的技能

    使用嵌入向量相似度 + LLM判断的混合方式
    """
    # 1. 语义搜索初选
    query_embedding = embed(task)
    scored_skills = []
    for skill in available_skills:
        # 组合技能描述和标签
        skill_text = f"{skill.name} {skill.description} {' '.join(skill.tags)}"
        skill_embedding = embed(skill_text)
        similarity = cosine_similarity(query_embedding, skill_embedding)
        scored_skills.append((similarity, skill))

    # 2. 取Top 5
    scored_skills.sort(reverse=True, key=lambda x: x[0])
    candidates = [s for _, s in scored_skills[:5]]

    # 3. LLM做最终判断
    prompt = f"""
    任务: {task}

    候选技能:
    {[(i, s.name, s.description) for i, s in enumerate(candidates)]}

    请选择最适合完成此任务的技能，返回JSON格式:
    {{
        "selected_skill_index": 0-based index,
        "confidence": 0-1,
        "reasoning": "简短说明"
    }}
    """
    llm_decision = parse_json(llm.ask(prompt))

    return SkillSelection(
        skill=candidates[llm_decision['selected_skill_index']],
        confidence=llm_decision['confidence'],
        reasoning=llm_decision['reasoning']
    )

@declarative_skill(
    name="write_blog",
    description="根据主题撰写一篇完整的博客文章",
    input_schema=[
        SkillParameter("topic", "string", "博客主题"),
        SkillParameter("style", "string", "文章风格", required=False, default="tech",
                      enum_values=["tech", "tutorial", "opinion", "news"])
    ],
    output_schema={"markdown": "string"}
)
def write_blog(topic: str, style: str = "tech"):
    """
    撰写博客（声明式实现）

    LLM会自动规划执行路径，例如:
    1. 搜索相关信息
    2. 撰写大纲
    3. 逐段撰写
    4. 润色优化
    """
    return llm.plan_and_execute(f"""
    请撰写一篇关于"{topic}"的博客文章，风格：{style}

    要求:
    - 结构清晰，包含引言、正文、总结
    - 有实际案例和代码示例（如适用）
    - 字数在1500-2000字
    """,
    available_tools=[web_search, read_file, outline, draft, revise]
)

@imperative_skill
def write_blog(topic: str, style: str = "tech") -> str:
    """
    撰写博客（命令式实现）

    每一步都明确定义，执行路径确定
    """
    # 步骤1：调研
    logger.info("步骤1: 调研主题...")
    research = research_topic(topic, depth="deep")

    # 步骤2：写大纲
    logger.info("步骤2: 生成大纲...")
    outline = generate_outline(research, style=style)

    # 步骤3：逐段撰写
    logger.info("步骤3: 撰写内容...")
    sections = []
    for i, section in enumerate(outline.sections):
        logger.info(f"  撰写章节 {i+1}/{len(outline.sections)}: {section.title}")
        content = write_section(
            section=section,
            research=research,
            style=style,
            outline=outline
        )
        sections.append(content)

    # 步骤4：合并润色
    logger.info("步骤4: 合并润色...")
    full_text = "\n\n".join(sections)
    final_article = polish(
        text=full_text,
        style=style,
        topic=topic
    )

    return final_article

from dataclasses import dataclass

@dataclass
class StateContext:
    conversation_history: List[Message]
    user_profile: Dict
    current_order: Optional[Order] = None
    pending_actions: List = field(default_factory=list)

@state_machine_skill(initial_state="greeting")
class CustomerServiceSkill:
    """
    客服技能（状态机实现）

    状态流转:
    greeting → listening → {troubleshooting, faq, escalation} → confirm_solved → end
    """

    def on_greeting(self, ctx: StateContext) -> Transition:
        """问候状态 - 对话开始"""
        greeting = "您好！欢迎联系客服，请问有什么可以帮助您的？"
        return Transition(
            next_state="listening",
            response=greeting
        )

    def on_listening(self, ctx: StateContext) -> Transition:
        """聆听状态 - 分析用户意图"""
        user_message = ctx.conversation_history[-1].content
        intent = classify_intent(user_message)

        if intent == "problem":
            return Transition(
                next_state="troubleshooting",
                context_update={"problem_type": extract_problem_type(user_message)}
            )
        elif intent == "complaint":
            return Transition(next_state="escalation")
        elif intent == "question":
            return Transition(next_state="faq")
        else:
            return Transition(
                next_state="listening",
                response="抱歉，我没有完全理解。可以请您再详细说明一下吗？"
            )

    def on_troubleshooting(self, ctx: StateContext) -> Transition:
        """问题排查状态 - 尝试解决问题"""
        problem = ctx.problem_type
        solution = find_solution(problem)

        if solution:
            return Transition(
                next_state="confirm_solved",
                response=f"请尝试以下解决方案：\n{solution}\n\n这解决了您的问题吗？"
            )
        else:
            return Transition(next_state="escalation")

    def on_confirm_solved(self, ctx: StateContext) -> Transition:
        """确认解决状态"""
        user_response = ctx.conversation_history[-1].content

        if is_affirmative(user_response):
            return Transition(
                next_state="end",
                response="太好了！很高兴能帮到您。如有其他问题，随时联系我们。"
            )
        else:
            return Transition(
                next_state="escalation",
                response="抱歉没能解决您的问题，我将为您转接人工客服。"
            )

    def on_faq(self, ctx: StateContext) -> Transition:
        """常见问题状态"""
        question = ctx.conversation_history[-1].content
        answer = find_faq_answer(question)

        return Transition(
            next_state="listening",
            response=f"{answer}\n\n还有其他问题吗？"
        )

    def on_escalation(self, ctx: StateContext) -> Transition:
        """升级状态 - 转接人工"""
        return Transition(
            next_state="end",
            response="正在为您转接人工客服，请稍候..."
        )

    def on_end(self, ctx: StateContext) -> Transition:
        """结束状态"""
        return Transition(
            next_state="end",  # 保持在结束状态
            response=None,
            is_terminal=True
        )

import os
import importlib.util
from typing import Dict, List, Optional, Set
from collections import defaultdict
from dataclasses import dataclass

@dataclass
class SkillIndex:
    name: str
    description: str
    categories: List[str]
    tags: List[str]
    embedding: List[float]
    skill_path: str
    is_loaded: bool = False
    skill: Optional[Skill] = None

class SkillRegistry:
    def __init__(self, embedding_model=None):
        self._skills: Dict[str, Skill] = {}
        self._categories: Dict[str, List[str]] = defaultdict(list)
        self._index: Dict[str, SkillIndex] = {}
        self._embedding_model = embedding_model or get_default_embedder()
        self._skill_dirs: List[str] = []

    def register(self, skill: Skill, categories: List[str] = None):
        """注册单个技能"""
        self._skills[skill.name] = skill

        # 分类索引
        all_categories = (skill.categories or []) + (categories or [])
        for cat in all_categories:
            if skill.name not in self._categories[cat]:
                self._categories[cat].append(skill.name)

        # 构建搜索索引
        self._build_index(skill)

        logger.info(f"Registered skill: {skill.name}")

    def register_directory(self, directory: str, recursive: bool = True):
        """从目录批量注册技能"""
        self._skill_dirs.append(directory)

        for root, dirs, files in os.walk(directory):
            for file in files:
                if file.endswith('_skill.py') or file.endswith('skills.py'):
                    self._register_from_file(os.path.join(root, file))

            if not recursive:
                break

    def _register_from_file(self, filepath: str):
        """从Python文件注册技能"""
        spec = importlib.util.spec_from_file_location("skill_module", filepath)
        module = importlib.util.module_from_spec(spec)

        try:
            spec.loader.exec_module(module)

            # 查找所有Skill实例或@skill装饰的函数
            for name, obj in module.__dict__.items():
                if isinstance(obj, Skill):
                    self.register(obj)
                elif hasattr(obj, '_is_skill'):
                    self.register(obj)

        except Exception as e:
            logger.error(f"Failed to load skills from {filepath}: {e}")

    def _build_index(self, skill: Skill):
        """构建技能的语义搜索索引"""
        # 组合用于嵌入的文本
        index_text = "\n".join([
            f"Name: {skill.name}",
            f"Description: {skill.description}",
            f"Categories: {', '.join(skill.categories or [])}",
            f"Tags: {', '.join(skill.tags or [])}",
        ])

        # 生成嵌入向量
        embedding = self._embedding_model.embed(index_text)

        self._index[skill.name] = SkillIndex(
            name=skill.name,
            description=skill.description,
            categories=skill.categories or [],
            tags=skill.tags or [],
            embedding=embedding,
            skill_path=f"registry://{skill.name}",
            is_loaded=True,
            skill=skill
        )

    def find_by_name(self, name: str) -> Optional[Skill]:
        """按名称查找"""
        return self._skills.get(name)

    def find_by_category(self, category: str) -> List[Skill]:
        """按分类查找"""
        return [
            self._skills[name]
            for name in self._categories.get(category, [])
            if name in self._skills
        ]

    def find_by_description(
        self,
        query: str,
        top_k: int = 5,
        threshold: float = 0.5
    ) -> List[Skill]:
        """
        语义搜索（供LLM使用）

        使用嵌入向量做相似度匹配，这比纯文本搜索更能理解语义
        """
        if not self._embedding_model:
            # 降级到关键词匹配
            return self._keyword_search(query, top_k)

        # 生成查询嵌入
        query_embedding = self._embedding_model.embed(query)

        # 计算相似度
        scores = []
        for skill_name, index in self._index.items():
            if not index.is_loaded:
                continue

            similarity = cosine_similarity(query_embedding, index.embedding)
            if similarity >= threshold:
                scores.append((similarity, self._skills[skill_name]))

        # 排序返回
        scores.sort(reverse=True, key=lambda x: x[0])
        return [skill for _, skill in scores[:top_k]]

    def _keyword_search(self, query: str, top_k: int) -> List[Skill]:
        """降级的关键词搜索"""
        query_words = set(query.lower().split())
        scores = []

        for skill in self._skills.values():
            text = f"{skill.name} {skill.description}".lower()
            matches = sum(1 for word in query_words if word in text)
            if matches > 0:
                scores.append((matches, skill))

        scores.sort(reverse=True, key=lambda x: x[0])
        return [skill for _, skill in scores[:top_k]]

    def list_all(self) -> List[Skill]:
        """列出所有已注册的技能"""
        return list(self._skills.values())

    def to_openai_functions(self, skills: List[Skill] = None) -> List[Dict]:
        """导出为OpenAI Function Calling格式"""
        skills = skills or self.list_all()
        return [skill.to_openai_function() for skill in skills]

    def get_system_prompt_addition(self, skills: List[Skill] = None) -> str:
        """生成技能描述的系统提示词"""
        skills = skills or self.list_all()

        prompt = "\n你可以使用以下技能:\n"
        for skill in skills:
            prompt += f"- {skill.name}: {skill.description}\n"

        return prompt

# 扩展SkillRegistry，增加懒加载能力
class LazySkillRegistry(SkillRegistry):
    def __init__(self, embedding_model=None):
        super().__init__(embedding_model)
        self._skill_packages: Dict[str, str] = {}  # name → package_path

    def register_package_index(self, index_path: str):
        """注册技能包索引（只加载元数据）"""
        with open(index_path) as f:
            index = json.load(f)

        for package in index['packages']:
            self._register_package_metadata(package)

    def _register_package_metadata(self, package: Dict):
        """注册技能包的元数据（不实际加载代码）"""
        for skill_meta in package['skills']:
            name = skill_meta['name']
            self._index[name] = SkillIndex(
                name=name,
                description=skill_meta['description'],
                categories=skill_meta.get('categories', []),
                tags=skill_meta.get('tags', []),
                embedding=self._embedding_model.embed(
                    f"{name} {skill_meta['description']}"
                ),
                skill_path=package['path'],
                is_loaded=False
            )
            self._skill_packages[name] = package['path']

    def find_by_description(self, query: str, top_k: int = 5) -> List[Skill]:
        """语义搜索，按需加载"""
        # 先搜索所有索引（包括未加载的）
        query_embedding = self._embedding_model.embed(query)
        scored = []

        for name, index in self._index.items():
            sim = cosine_similarity(query_embedding, index.embedding)
            scored.append((sim, index))

        scored.sort(reverse=True, key=lambda x: x[0])

        # 按需加载Top K
        result = []
        for sim, index in scored[:top_k]:
            if not index.is_loaded:
                self._load_skill_package(index.name)
            if index.name in self._skills:
                result.append(self._skills[index.name])

        return result

    def _load_skill_package(self, skill_name: str):
        """按需加载技能包"""
        if skill_name not in self._skill_packages:
            return

        package_path = self._skill_packages[skill_name]
        logger.info(f"Lazy loading skill package: {package_path}")

        self.register_directory(package_path)

class DirectInvocationAgent:
    def __init__(self):
        self.skills = SkillRegistry()
        self._register_core_skills()

    def _register_core_skills(self):
        """注册核心技能"""
        self.skills.register(web_search)
        self.skills.register(write_blog)
        self.skills.register(code_review)

    def handle(self, user_input: str) -> str:
        """根据用户输入选择技能"""
        if "写博客" in user_input or "blog" in user_input.lower():
            topic = extract_topic(user_input)
            result = self.skills.find_by_name("write_blog").execute(topic=topic)
            return f"博客已生成：\n\n{result}"

        elif "代码审查" in user_input or "review" in user_input.lower():
            code = extract_code(user_input)
            result = self.skills.find_by_name("code_review").execute(code=code)
            return format_code_review(result)

        elif "搜索" in user_input or "search" in user_input.lower():
            query = extract_query(user_input)
            results = self.skills.find_by_name("web_search").execute(query=query)
            return format_search_results(results)

        else:
            return llm.chat(user_input)

class LLMSelectionAgent:
    def __init__(self):
        self.skills = SkillRegistry()
        self.skills.register_directory("./skills")
        self.message_history = []

    def react(self, task: str) -> str:
        """
        ReAct范式 + LLM技能选择

        这一模式结合了ReAct论文（Yao et al., 2022）的思想
        """
        self.message_history.append({"role": "user", "content": task})

        max_iterations = 10
        for i in range(max_iterations):
            # 1. 渐进式披露：获取当前相关的技能
            relevant_skills = self._get_relevant_skills()

            # 2. 让LLM选择和调用技能（或直接回答）
            response = llm.ask_with_functions(
                messages=self.message_history,
                functions=self.skills.to_openai_functions(relevant_skills)
            )

            if not response.tool_calls:
                # 没有工具调用，说明这是最终回答
                self.message_history.append({
                    "role": "assistant",
                    "content": response.content
                })
                return response.content

            # 3. 解析并执行工具调用
            self.message_history.append({
                "role": "assistant",
                "content": response.content,
                "tool_calls": response.tool_calls
            })

            for tool_call in response.tool_calls:
                skill = self.skills.find_by_name(tool_call.function.name)
                if skill:
                    try:
                        result = skill.execute(**parse_json(tool_call.function.arguments))
                        result_str = str(result)
                    except Exception as e:
                        result_str = f"Error: {str(e)}"

                    # 4. 将结果返回给LLM
                    self.message_history.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result_str
                    })

        return "已达到最大迭代次数"

    def _get_relevant_skills(self) -> List[Skill]:
        """渐进式披露：获取与当前上下文相关的技能"""
        # 结合最近的消息来搜索相关技能
        context = "\n".join([
            m.get("content", "") for m in self.message_history[-5:]
        ])
        return self.skills.find_by_description(context, top_k=5)

from typing import Dict, List, Optional, Set
from dataclasses import dataclass
from enum import Enum

class NodeStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class DAGNode:
    id: str
    skill: Skill
    params: Dict
    dependencies: List[str] = field(default_factory=list)
    status: NodeStatus = NodeStatus.PENDING
    result: Any = None
    error: Optional[Exception] = None

class OrchestratorAgent:
    def __init__(self):
        self.skills = SkillRegistry()

    def plan_and_execute(self, task: str) -> Any:
        """
        规划与执行

        参考了Google的[Plan-and-Solve论文](https://arxiv.org/abs/2305.04091)[^20]（Wang et al., 2023）
        和OpenAI的[Planning Agents最佳实践](https://platform.openai.com/docs/guides/function-calling#planning)[^7]
        """
        # 1. 生成执行计划
        logger.info("阶段1: 生成执行计划")
        plan = self._generate_plan(task)

        # 2. 构建DAG
        logger.info("阶段2: 构建执行图")
        dag = self._build_dag(plan)

        # 3. 拓扑排序执行
        logger.info("阶段3: 执行任务")
        results = {}
        execution_order = topological_sort(dag)

        for node in execution_order:
            if node.status == NodeStatus.COMPLETED:
                continue

            logger.info(f"执行节点: {node.id} ({node.skill.name})")
            node.status = NodeStatus.RUNNING

            try:
                # 收集依赖
                deps = {k: results[k] for k in node.dependencies}
                # 合并参数
                params = {**node.params, **deps}
                # 执行技能
                node.result = node.skill.execute(**params)
                node.status = NodeStatus.COMPLETED
                results[node.id] = node.result
                logger.info(f"节点完成: {node.id}")

            except Exception as e:
                node.status = NodeStatus.FAILED
                node.error = e
                logger.error(f"节点失败: {node.id}, 错误: {e}")

                # 尝试容错策略
                if not self._handle_failure(node, dag):
                    raise  # 无法恢复，抛出异常

        # 4. 返回最终结果
        return results[dag.final_node]

    def _generate_plan(self, task: str) -> Plan:
        """让LLM生成执行计划"""
        prompt = f"""
        分析以下任务，将其分解为可执行的步骤。

        任务: {task}

        可用技能:
        {self._skill_list_for_prompt()}

        请返回JSON格式的计划:
        {{
            "steps": [
                {{
                    "id": "step_1",
                    "skill": "技能名称",
                    "params": {{参数}},
                    "dependencies": ["依赖的step_id"]
                }}
            ],
            "final_output": "最后一步的step_id"
        }}
        """
        plan_json = llm.ask(prompt, response_format={"type": "json_object"})
        return parse_plan(plan_json)

    def _build_dag(self, plan: Plan) -> DAG:
        """从计划构建DAG"""
        nodes = {}
        for step in plan.steps:
            skill = self.skills.find_by_name(step.skill)
            nodes[step.id] = DAGNode(
                id=step.id,
                skill=skill,
                params=step.params,
                dependencies=step.dependencies
            )
        return DAG(nodes=nodes, final_node=plan.final_output)

    def _handle_failure(self, node: DAGNode, dag: DAG) -> bool:
        """处理节点失败"""
        # 策略1: 重试（如适用）
        if node.skill.max_retries > 0:
            retry_count = getattr(node, '_retry_count', 0)
            if retry_count < node.skill.max_retries:
                node._retry_count = retry_count + 1
                node.status = NodeStatus.PENDING
                logger.info(f"重试节点 {node.id} ({retry_count + 1}/{node.skill.max_retries})")
                return True

        # 策略2: 使用备选技能
        if hasattr(node.skill, 'fallback_skills'):
            for fallback_name in node.skill.fallback_skills:
                fallback = self.skills.find_by_name(fallback_name)
                if fallback:
                    logger.info(f"使用备选技能: {fallback_name}")
                    node.skill = fallback
                    node.status = NodeStatus.PENDING
                    return True

        # 策略3: 询问用户
        # ...

        return False

from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum

class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    STYLE = "style"

@dataclass
class Issue:
    line: int
    column: int
    severity: Severity
    message: str
    rule_id: str
    suggestion: Optional[str] = None

@dataclass
class ReviewReport:
    issues: List[Issue]
    overall_score: float
    summary: str
    suggested_fixes: List[str]

@composite_skill(
    name="code_review",
    description="对代码进行全面审查，包括风格、安全性、性能等",
    categories=["development", "code_quality"],
    dependencies=[
        "lint_check", "security_scan", "performance_analyze",
        "complexity_measure", "generate_fix"
    ]
)
def code_review(code: str, lang: str = "python") -> ReviewReport:
    """
    代码审查技能

    这是一个组合技能，并行执行多个检查后综合分析
    设计参考了:
    - [GitHub's Code Scanning](https://docs.github.com/en/code-security/code-scanning)[^25]
    - [SonarQube's Quality Gates](https://docs.sonarqube.com/latest/quality-gates/)[^26]
    - [Google's Code Review Guidelines](https://google.github.io/eng-practices/)[^27]
    """
    # 步骤1: 并行执行多个检查
    logger.info("启动代码审查...")

    checks = parallel_run([
        ("lint", lambda: lint_check(code, lang)),           # 风格检查
        ("security", lambda: security_scan(code, lang)),    # 安全扫描
        ("performance", lambda: performance_analyze(code, lang)),  # 性能分析
        ("complexity", lambda: complexity_measure(code, lang))    # 复杂度度量
    ], max_workers=4)

    # 步骤2: 收集所有问题
    all_issues = []
    for check_name, check_result in checks.items():
        if check_result and check_result.issues:
            for issue in check_result.issues:
                issue.source = check_name
                all_issues.append(issue)

    # 步骤3: 按严重程度排序和去重
    all_issues.sort(key=lambda x: (
        -_severity_weight(x.severity),  # 严重程度降序
        x.line, x.column                 # 位置升序
    ))
    all_issues = _deduplicate_issues(all_issues)

    # 步骤4: 为最严重的问题生成修复建议
    suggested_fixes = []
    for issue in all_issues[:5]:  # 重点修复前5个
        try:
            fix = generate_fix(code, issue, lang)
            if fix:
                suggested_fixes.append(fix)
        except Exception as e:
            logger.warning(f"生成修复失败: {e}")

    # 步骤5: 计算总体评分
    overall_score = _calculate_score(all_issues, checks)

    # 步骤6: 生成摘要
    summary = _generate_summary(all_issues, overall_score, checks)

    logger.info(f"代码审查完成: {len(all_issues)} 个问题, 评分 {overall_score:.1f}/10")

    return ReviewReport(
        issues=all_issues,
        overall_score=overall_score,
        summary=summary,
        suggested_fixes=suggested_fixes
    )

def _severity_weight(severity: Severity) -> int:
    weights = {
        Severity.CRITICAL: 5,
        Severity.HIGH: 4,
        Severity.MEDIUM: 3,
        Severity.LOW: 2,
        Severity.STYLE: 1
    }
    return weights.get(severity, 0)

def _deduplicate_issues(issues: List[Issue]) -> List[Issue]:
    """去重：同一位置的相似问题只保留最严重的"""
    seen = set()
    result = []
    for issue in issues:
        key = (issue.line, issue.column, issue.rule_id)
        if key not in seen:
            seen.add(key)
            result.append(issue)
    return result

def _calculate_score(issues: List[Issue], checks: Dict) -> float:
    """
    计算代码质量评分 (0-10)

    参考[SonarQube的评分算法](https://docs.sonarqube.com/latest/quality-gates/)[^26]
    """
    base_score = 10.0

    # 根据问题扣分
    penalties = {
        Severity.CRITICAL: 2.0,
        Severity.HIGH: 1.0,
        Severity.MEDIUM: 0.3,
        Severity.LOW: 0.1,
        Severity.STYLE: 0.05
    }

    for issue in issues:
        base_score -= penalties.get(issue.severity, 0)

    # 复杂度加分/减分
    if 'complexity' in checks:
        complexity = checks['complexity']
        if complexity.cognitive_complexity < 10:
            base_score += 0.5
        elif complexity.cognitive_complexity > 30:
            base_score -= 1.0

    return max(0.0, min(10.0, base_score))

def _generate_summary(issues: List[Issue], score: float, checks: Dict) -> str:
    """生成自然语言摘要"""
    severity_counts = defaultdict(int)
    for issue in issues:
        severity_counts[issue.severity] += 1

    score_desc = {
        (9, 10): "优秀",
        (7, 9): "良好",
        (5, 7): "一般",
        (3, 5): "较差",
        (0, 3): "很差"
    }
    desc = next(v for (low, high), v in score_desc.items() if low <= score < high)

    summary_parts = [
        f"代码质量评分: {score:.1f}/10 ({desc})",
        f"发现问题: {len(issues)} 个",
    ]

    if severity_counts[Severity.CRITICAL]:
        summary_parts.append(f"🔴 严重问题: {severity_counts[Severity.CRITICAL]}")
    if severity_counts[Severity.HIGH]:
        summary_parts.append(f"🟠 重要问题: {severity_counts[Severity.HIGH]}")

    return "\n".join(summary_parts)