graph TB
    subgraph "工具获得六层栈"
        L1["L1 静态注册<br/>开发者硬编码工具列表"]
        L2["L2 协议发现<br/>MCP Registry查询"]
        L3["L3 动态选择<br/>AutoTool图匹配"]
        L4["L4 经验学习<br/>ToolSandbox交互反馈"]
        L5["L5 文档生造<br/>ToolMaker自动生成"]
        L6["L6 自主进化<br/>Agent0零数据自造"]
    end

    L1 -->|"工具不够用"| L2
    L2 -->|"发现但不精准"| L3
    L3 -->|"选对但不会用"| L4
    L4 -->|"学会了但没有"| L5
    L5 -->|"生造依赖文档"| L6

    style L1 fill:#e8f5e9
    style L2 fill:#e3f2fd
    style L3 fill:#fff3e0
    style L4 fill:#fce4ec
    style L5 fill:#f3e5f5
    style L6 fill:#fff8e1

# LangChain静态工具注册 - 典型L1模式
from langchain_core.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search the product database for matching items.

    Args:
        query: Natural language search query
        limit: Maximum number of results to return
    """
    results = db.execute(f"SELECT * FROM products WHERE name LIKE '%{query}%' LIMIT {limit}")
    return json.dumps(results)

@tool
def calculate_price(base_price: float, discount: float) -> str:
    """Calculate final price after applying discount."""
    return str(base_price * (1 - discount))

# Agent只能使用这两个工具，无法扩展
agent = create_react_agent(model, [search_database, calculate_price])

# MCP工具发现协议 - L2模式
import httpx

async def discover_tools_from_registry(
    registry_url: str,
    capability_filter: str = None
) -> list:
    """从MCP注册表发现可用工具"""
    async with httpx.AsyncClient() as client:
        # 查询注册表
        response = await client.get(
            f"{registry_url}/api/servers",
            params={"category": capability_filter} if capability_filter else {}
        )
        servers = response.json()

        tools = []
        for server in servers:
            # 获取每个server的工具列表
            tool_response = await client.get(
                f"{server['endpoint']}/tools"
            )
            tools.extend(tool_response.json().get("tools", []))

        return tools

# 问题：33个注册表，12,000+ servers，质量参差不齐

# AutoTool核心架构 - 图结构动态工具选择
class AutoToolSelector:
    def __init__(self):
        self.tool_graph = ToolGraph()  # 工具能力知识图谱
        self.embedding_cache = {}

    def register_tool(self, tool_def: dict):
        """将工具注册到图中，按能力节点连接"""
        capabilities = self._extract_capabilities(tool_def)
        tool_node = ToolNode(
            id=tool_def["name"],
            description=tool_def["description"],
            params=tool_def["parameters"]
        )
        for cap in capabilities:
            self.tool_graph.add_edge(
                CapabilityNode(cap), tool_node
            )

    def select_tools(self, task_description: str, top_k: int = 5) -> list:
        """动态选择：任务→能力节点→工具节点"""
        task_caps = self._parse_task_requirements(task_description)
        candidates = []
        for cap in task_caps:
            matched = self.tool_graph.traverse(
                start=cap, 
                strategy="shortest_path"
            )
            candidates.extend(matched)

        # 轻量级排序，无需LLM推理
        return self._rank_by_relevance(candidates, task_description)[:top_k]

    def update_graph(self, new_tool: dict, removed_tool: str = None):
        """运行时热更新工具图"""
        if removed_tool:
            self.tool_graph.remove_node(removed_tool)
        if new_tool:
            self.register_tool(new_tool)

# ToolSandbox经验学习架构 - L4模式
class ToolExperienceLearner:
    def __init__(self, llm, experience_db):
        self.llm = llm
        self.experience_db = experience_db  # 工具使用经验库

    def call_with_experience(self, tool_name: str, params: dict, context: str):
        """带经验学习的工具调用"""
        # 1. 检索历史经验
        past_experiences = self.experience_db.search(
            tool=tool_name,
            scenario_similarity=context,
            top_k=3
        )

        # 2. 从失败经验中提取约束
        constraints = []
        for exp in past_experiences:
            if exp["outcome"] == "failure":
                constraints.append({
                    "avoid": exp["params"],
                    "reason": exp["error_message"],
                    "correction": exp.get("retry_params")
                })

        # 3. 注入经验到调用prompt
        enhanced_prompt = self._build_prompt_with_experience(
            tool_name, params, constraints
        )

        return self.llm.generate(enhanced_prompt)

    def record_experience(self, tool_name, params, result, success):
        """记录每次调用的结果到经验库"""
        self.experience_db.store({
            "tool": tool_name,
            "params": params,
            "result": result,
            "outcome": "success" if success else "failure",
            "timestamp": datetime.now()
        })

# ToolMaker自动化工具生成流程 - L5模式
class ToolMaker:
    def __init__(self, llm, code_executor):
        self.llm = llm
        self.executor = code_executor  # 沙箱执行器

    async def create_tool_from_repo(self, repo_url: str) -> dict:
        """从GitHub仓库自动生成工具"""
        # Step 1: 分析仓库结构和接口
        repo_analysis = await self._analyze_repository(repo_url)

        # Step 2: 生成工具封装代码
        tool_code = self.llm.generate(f"""
        Given this repository analysis:
        {repo_analysis}

        Generate a tool wrapper with:
        1. Clear function signature with type hints
        2. Input validation
        3. Error handling
        4. Docstring with usage examples
        """)

        # Step 3: 自动测试验证
        test_results = await self._self_test(tool_code, repo_analysis)

        if test_results["pass_rate"] < 0.7:
            # 自我修复循环
            tool_code = await self._self_repair(
                tool_code, test_results["failures"]
            )

        return {
            "tool_definition": self._extract_schema(tool_code),
            "implementation": tool_code,
            "test_coverage": test_results["pass_rate"]
        }

    async def _self_test(self, tool_code: str, analysis: dict) -> dict:
        """自动生成测试用例并执行"""
        test_cases = self.llm.generate(
            f"Generate test cases for: {tool_code}"
        )
        results = await self.executor.run(test_cases)
        return {
            "pass_rate": results["passed"] / results["total"],
            "failures": results.get("failures", [])
        }

# Agent0协同进化架构 - L6模式
class Agent0Framework:
    def __init__(self, base_llm):
        self.curriculum_agent = CurriculumAgent(base_llm)
        self.tool_agent = ToolAgent(base_llm)
        self.task_agent = TaskAgent(base_llm)
        self.tool_pool = ToolPool()

    def evolve(self, iterations: int = 100):
        """零数据协同进化主循环"""
        for i in range(iterations):
            # 1. Curriculum Agent：生成有挑战性的任务
            #    奖励机制鼓励生成"需要工具使用"的任务
            task = self.curriculum_agent.generate_task(
                difficulty="adaptive",
                tool_requirement="must_require_new_tools",
                uncertainty_bonus=True  # 不确定性越高奖励越大
            )

            # 2. Task Agent：尝试完成任务
            result = self.task_agent.execute(task, self.tool_pool)

            # 3. Tool Agent：根据失败分析创造新工具
            if not result.success:
                new_tool = self.tool_agent.create_tool(
                    task=task,
                    failure_analysis=result.error_analysis,
                    existing_tools=self.tool_pool.list()
                )
                self.tool_pool.add(new_tool)

            # 4. 协同进化奖励
            self.curriculum_agent.update(
                task_difficulty=task.uncertainty_score,
                tool_novelty=result.required_new_tools
            )

# 关键洞察：三个Agent互相制造压力
# Curriculum Agent故意制造需要新工具的任务
# Tool Agent被迫创造新工具
# Task Agent的使用反馈驱动工具改进

graph LR
    subgraph "工具发现四层架构"
        A["Agent工具请求"] --> B["发现层<br/>Registry查询"]
        B --> C["验证层<br/>安全+质量检查"]
        C --> D["适配层<br/>Schema转换"]
        D --> E["执行层<br/>沙箱运行"]
    end

    B -->|"碎片化"| B1["官方Registry"]
    B -->|"碎片化"| B2["GitHub Registry"]
    B -->|"碎片化"| B3["聚合索引"]
    B -->|"碎片化"| B4["私有Registry"]

    B1 -->|"元数据不一致"| C
    B2 -->|"审核标准不一"| C
    B3 -->|"无质量保障"| C
    B4 -->|"覆盖范围小"| C

    style B fill:#ffcdd2
    style C fill:#fff9c4

# LLM Compiler模式 - DAG工具编排
class ToolDAGCompiler:
    def __init__(self, llm):
        self.llm = llm

    def compile_tool_chain(self, task: str, available_tools: list) -> 'DAG':
        """将任务编译为工具执行DAG"""
        # Step 1: LLM规划工具依赖关系
        plan = self.llm.generate(f"""
        Task: {task}
        Available tools: {json.dumps(available_tools)}

        Generate a DAG execution plan:
        - Identify tool dependencies (output→input)
        - Mark independent tools for parallel execution
        - Specify data flow between tools
        """)

        # Step 2: 构建DAG
        dag = DAG()
        for step in plan.steps:
            dag.add_node(
                id=step.id,
                tool=step.tool,
                params=step.params,
                depends_on=step.depends_on  # 依赖的工具节点
            )

        # Step 3: 拓扑排序 + 并行分组
        execution_layers = dag.topological_sort_parallel()

        return execution_layers

    async def execute_dag(self, dag_layers):
        """按DAG层级并行执行"""
        results = {}
        for layer in dag_layers:
            # 同一层的工具可并行执行
            tasks = [
                self._execute_tool(node, results)
                for node in layer
            ]
            layer_results = await asyncio.gather(*tasks)
            results.update(zip([n.id for n in layer], layer_results))

        return results

graph TD
    subgraph "生产级工具获得架构推荐"
        A["任务输入"] --> B{工具库规模}
        B -->|<20| C["L1 静态注册<br/>直接调用"]
        B -->|20-100| D["L3 AutoTool<br/>图结构动态选择"]
        B -->|>100| E["L2 MCP Registry<br/>+ L3 AutoTool"]

        D --> F{是否长期运行}
        F -->|否| G["无状态执行"]
        F -->|是| H["+ L4 经验学习<br/>积累调用经验"]

        E --> I{现有工具是否够用}
        I -->|够| H
        I -->|不够| J["+ L5 ToolMaker<br/>动态生成工具"]

        J --> K{是否需要自主扩展}
        K -->|否| H
        K -->|是| L["+ L6 Agent0<br/>自主进化（实验性）"]
    end

    style C fill:#c8e6c9
    style H fill:#bbdefb
    style L fill:#fff9c4