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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 工具时代：手写每一步
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

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)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

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))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

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]
                }
            }
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

@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)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

@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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

@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']
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

@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]
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

@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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

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
        )

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# 扩展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)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

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)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

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论文（Wang et al., 2023）
        和OpenAI的Planning Agents最佳实践
        """
        # 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

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173