AI Agent 编排实战：从工具调用到多 Agent 协作的任务拆解与执行框架

一、单次对话搞不定复杂任务——Agent 编排的工程化刚需

让 AI 一次性完成"分析竞品定价策略、生成报告、发送邮件"这种多步骤任务，结果：要么步骤遗漏、要么上下文丢失、要么中间步骤出错后无法恢复。更常见的情况是，AI 自信地给出了一个看似完整但逻辑断裂的答案。

问题根源：大语言模型是单轮预测引擎，不是任务执行器。它擅长生成文本，不擅长规划步骤、调用工具、处理异常、维护状态。Agent 编排的本质，就是把 LLM 的生成能力包装成一个可控的任务执行框架。

Agent 编排的核心挑战：

• 任务拆解：复杂任务如何分解为可执行的原子步骤
• 工具调用：如何让 LLM 安全、可靠地调用外部 API
• 状态管理：多步骤执行过程中如何维护上下文和中间结果
• 异常恢复：某步骤失败时如何重试、回退或降级
• 多 Agent 协作：不同角色的 Agent 如何分工与通信

二、AI Agent 编排的架构与执行机制

Agent 编排的核心模式：ReAct 循环（Reasoning + Acting）——思考下一步做什么，执行动作，观察结果，再思考。

graph TB
    subgraph "Agent 执行循环"
        A[接收任务] --> B[推理: 分析当前状态]
        B --> C[决策: 选择下一步动作]
        C --> D{动作类型?}
        D -->|工具调用| E[执行工具]
        D -->|子任务委派| F[委派给子 Agent]
        D -->|直接回答| G[生成最终结果]
        E --> H[观察: 获取执行结果]
        F --> H
        H --> I{任务完成?}
        I -->|否| B
        I -->|是| G
    end
    subgraph "工具注册表"
        J[搜索工具]
        K[数据库查询]
        L[文件读写]
        M[API 调用]
    end
    subgraph "状态管理"
        N[对话历史]
        O[中间结果]
        P[执行日志]
    end
    E --> J & K & L & M
    B --> N & O & P

编排模式对比：

三、生产级 Agent 编排框架实现

3.1 工具定义与注册

from dataclasses import dataclass, field
from typing import Callable, Any
import json

@dataclass
class ToolDefinition:
    """工具定义：描述工具的输入输出规范，供 LLM 理解和调用"""

    name: str
    description: str                    # 工具功能描述，LLM 据此判断何时调用
    parameters: dict                    # JSON Schema 格式的参数定义
    required: list[str]                 # 必填参数列表
    executor: Callable                  # 实际执行函数
    timeout_seconds: int = 30           # 执行超时时间
    max_retries: int = 2                # 最大重试次数

class ToolRegistry:
    """工具注册表：集中管理所有可用工具"""

    def __init__(self):
        self._tools: dict[str, ToolDefinition] = {}

    def register(self, tool: ToolDefinition):
        """注册工具"""
        self._tools[tool.name] = tool

    def get(self, name: str) -> ToolDefinition | None:
        return self._tools.get(name)

    def get_all_schemas(self) -> list[dict]:
        """获取所有工具的 JSON Schema 描述，供 LLM 函数调用使用"""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters,
                    "required": tool.required,
                },
            }
            for tool in self._tools.values()
        ]

    async def execute(self, name: str, arguments: dict) -> Any:
        """执行工具调用，带超时和重试"""
        tool = self._tools.get(name)
        if not tool:
            raise ValueError(f"未知工具: {name}")

        # 参数校验：检查必填参数是否存在
        for param in tool.required:
            if param not in arguments:
                raise ValueError(f"工具 {name} 缺少必填参数: {param}")

        last_error = None
        for attempt in range(tool.max_retries + 1):
            try:
                import asyncio
                result = await asyncio.wait_for(
                    tool.executor(**arguments),
                    timeout=tool.timeout_seconds,
                )
                return result
            except asyncio.TimeoutError:
                last_error = f"工具 {name} 执行超时 ({tool.timeout_seconds}s)"
            except Exception as e:
                last_error = f"工具 {name} 执行失败: {str(e)}"

        raise RuntimeError(f"工具 {name} 重试耗尽: {last_error}")

3.2 ReAct Agent 核心

import asyncio
from enum import Enum
from dataclasses import dataclass, field

class ActionKind(Enum):
    TOOL_CALL = "tool_call"
    FINAL_ANSWER = "final_answer"

@dataclass
class AgentAction:
    """Agent 的动作决策"""
    kind: ActionKind
    tool_name: str | None = None
    tool_args: dict | None = None
    answer: str | None = None
    reasoning: str = ""                # 推理过程，用于可解释性

@dataclass
class AgentState:
    """Agent 执行状态：维护对话历史和中间结果"""
    task: str
    messages: list[dict] = field(default_factory=list)
    tool_results: list[dict] = field(default_factory=list)
    step_count: int = 0
    max_steps: int = 15                # 防止无限循环

class ReActAgent:
    """ReAct 模式 Agent：推理-行动-观察循环"""

    def __init__(self, llm_client, tool_registry: ToolRegistry):
        self.llm = llm_client
        self.tools = tool_registry

    async def run(self, task: str) -> str:
        """执行任务，返回最终结果"""
        state = AgentState(task=task)
        # 初始化系统提示词
        state.messages.append({
            "role": "system",
            "content": self._build_system_prompt(),
        })
        state.messages.append({
            "role": "user",
            "content": task,
        })

        while state.step_count < state.max_steps:
            state.step_count += 1

            # 推理：让 LLM 决定下一步动作
            action = await self._decide_action(state)

            if action.kind == ActionKind.FINAL_ANSWER:
                return action.answer or ""

            if action.kind == ActionKind.TOOL_CALL:
                # 行动：执行工具调用
                result = await self._execute_tool(action)

                # 观察：将工具结果加入上下文
                state.tool_results.append({
                    "step": state.step_count,
                    "tool": action.tool_name,
                    "args": action.tool_args,
                    "result": result,
                })
                state.messages.append({
                    "role": "assistant",
                    "content": f"调用工具 {action.tool_name}，参数: {json.dumps(action.tool_args, ensure_ascii=False)}",
                })
                state.messages.append({
                    "role": "user",
                    "content": f"工具返回结果: {json.dumps(result, ensure_ascii=False)}",
                })

        return "任务执行超时：达到最大步骤数限制"

    async def _decide_action(self, state: AgentState) -> AgentAction:
        """让 LLM 决定下一步动作"""
        response = await self.llm.chat(
            messages=state.messages,
            tools=self.tools.get_all_schemas(),
            tool_choice="auto",
        )

        # 检查是否有工具调用
        if response.tool_calls:
            call = response.tool_calls[0]
            return AgentAction(
                kind=ActionKind.TOOL_CALL,
                tool_name=call.function.name,
                tool_args=json.loads(call.function.arguments),
                reasoning=response.content or "",
            )

        # 无工具调用，视为最终回答
        return AgentAction(
            kind=ActionKind.FINAL_ANSWER,
            answer=response.content or "",
            reasoning="",
        )

    async def _execute_tool(self, action: AgentAction) -> Any:
        """执行工具调用，捕获异常"""
        try:
            return await self.tools.execute(action.tool_name, action.tool_args or {})
        except Exception as e:
            # 工具执行失败，返回错误信息而非抛异常
            return {"error": str(e)}

    def _build_system_prompt(self) -> str:
        return """你是一个任务执行 Agent。根据用户任务，逐步推理并调用工具完成。

规则：
1. 每次只调用一个工具
2. 根据工具返回结果决定下一步
3. 如果工具返回错误，分析原因并尝试其他方案
4. 任务完成后给出最终答案
5. 不要编造工具返回结果"""

3.3 多 Agent 协作：层级编排

@dataclass
class SubTask:
    """子任务定义"""
    name: str
    description: str
    agent_role: str                    # 执行该任务的 Agent 角色
    dependencies: list[str] = field(default_factory=list)  # 依赖的子任务
    input_from: dict[str, str] = field(default_factory=dict)  # 从上游任务获取输入

class MultiAgentOrchestrator:
    """多 Agent 编排器：支持串行、并行和层级任务编排"""

    def __init__(self, agents: dict[str, ReActAgent]):
        self.agents = agents

    async def execute_pipeline(
        self,
        tasks: list[SubTask],
        initial_context: dict,
    ) -> dict:
        """执行任务流水线，自动处理依赖关系"""

        completed: dict[str, Any] = {}
        results: dict[str, Any] = {}

        # 拓扑排序：按依赖关系确定执行顺序
        execution_order = self._topological_sort(tasks)

        for task_name in execution_order:
            task = next(t for t in tasks if t.name == task_name)
            agent = self.agents.get(task.agent_role)
            if not agent:
                raise ValueError(f"未注册的 Agent 角色: {task.agent_role}")

            # 构建子任务输入：从上游结果中提取
            task_input = self._build_task_input(task, initial_context, results)

            # 执行子任务
            result = await agent.run(task_input)
            results[task.name] = result
            completed[task.name] = True

        return results

    def _topological_sort(self, tasks: list[SubTask]) -> list[str]:
        """拓扑排序：确保被依赖的任务先执行"""
        in_degree: dict[str, int] = {t.name: 0 for t in tasks}
        graph: dict[str, list[str]] = {t.name: [] for t in tasks}

        for task in tasks:
            for dep in task.dependencies:
                graph[dep].append(task.name)
                in_degree[task.name] += 1

        queue = [name for name, deg in in_degree.items() if deg == 0]
        order = []

        while queue:
            current = queue.pop(0)
            order.append(current)
            for neighbor in graph[current]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    queue.append(neighbor)

        if len(order) != len(tasks):
            raise ValueError("任务依赖关系存在环，无法排序")

        return order

    def _build_task_input(
        self,
        task: SubTask,
        initial_context: dict,
        completed_results: dict,
    ) -> str:
        """构建子任务输入，从上游结果中提取所需数据"""
        parts = [f"任务: {task.description}"]

        for param_name, source_path in task.input_from.items():
            # source_path 格式: "task_name.field_name"
            source_task, field = source_path.split(".", 1)
            if source_task in completed_results:
                parts.append(f"输入参数 {param_name}: {completed_results[source_task]}")

        if initial_context:
            parts.append(f"初始上下文: {json.dumps(initial_context, ensure_ascii=False)}")

        return "\n".join(parts)

四、Agent 编排的架构权衡与边界

ReAct 循环的可靠性问题：

• 无限循环：Agent 可能反复调用同一工具而不收敛。必须设置 max_steps 硬限制
• 幻觉工具调用：LLM 可能生成不存在的工具名或参数。需要在执行前校验
• 上下文窗口溢出：多轮工具调用后，对话历史超出模型上下文窗口。需要摘要压缩

多 Agent 协作的通信开销：

• 串行编排延迟叠加：3 个子任务各需 10 秒，总耗时 30 秒
• 并行编排需要无依赖前提：有依赖关系的子任务无法并行
• Agent 间信息传递有损：子 Agent 的输出被压缩为文本传给下游，结构化信息可能丢失

工具调用的安全边界：

• LLM 生成的参数可能包含注入攻击（如 SQL 注入、命令注入）
• 工具执行必须有权限隔离：只允许访问必要的资源
• 敏感操作（删除、支付）需要人工确认环节

禁用场景：

• 确定性任务：规则明确、逻辑固定的任务，用传统代码实现更可靠
• 实时性要求高的任务：LLM 推理延迟 1-5 秒，不适合毫秒级响应
• 安全敏感任务：金融交易、医疗诊断等场景，Agent 的不确定性不可接受

五、总结

AI Agent 编排的核心框架：ReAct 循环（推理-行动-观察）解决单 Agent 的任务执行问题，层级编排（分解-委派-汇总）解决多 Agent 的协作问题。工具注册表提供标准化的工具定义和执行接口，AgentState 维护多步骤执行的状态和上下文。ReAct 循环必须设置最大步骤数防止无限循环，工具调用需要参数校验和权限隔离。多 Agent 编排通过拓扑排序自动处理任务依赖，但串行编排的延迟叠加和 Agent 间信息传递的有损性是架构瓶颈。Agent 编排适用于步骤不确定、需要工具调用的复杂任务，确定性和实时性要求高的场景应使用传统代码实现。