AI Agent多工具调用框架设计与Function Calling协议实现

AI Agent工具调用架构设计核心思路

AI Agent的核心能力在于根据用户意图自主选择并调用外部工具，完成单一模型无法直接处理的任务。Function Calling协议将工具描述、参数Schema、调用结果统一结构化，使大语言模型能够以JSON格式输出工具调用请求，由执行引擎完成实际调用并将结果回传模型，形成完整的思考-行动-观察循环。

在设计多工具调用框架时，需要解决三个关键问题：工具注册与发现机制、并行调用与依赖调度策略、调用失败的容错与重试逻辑。一个成熟的Agent框架通常包含Tool Registry、Planner、Executor和Observer四个核心模块。

Function Calling协议规范与参数Schema定义

OpenAI提出的Function Calling协议已成为行业事实标准。其核心流程是：在Chat Completion请求中通过tools参数传入可用工具列表，模型根据上下文判断是否需要调用工具，若需要则输出一个tool_calls对象，包含工具名和参数JSON。调用方执行后将结果以tool角色消息回传，模型继续推理。

工具描述的关键在于参数JSON Schema的精确度。以下是一个工具注册的典型Schema定义：

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "在指定数据库中执行SQL查询，返回结果集。仅支持SELECT语句，禁止DDL和DML操作。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL SELECT查询语句，表名需带schema前缀"
                    },
                    "database": {
                        "type": "string",
                        "enum": ["production", "staging", "analytics"],
                        "description": "目标数据库环境"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回结果行数上限，默认100",
                        "default": 100
                    }
                },
                "required": ["query", "database"]
            }
        }
    }
]

description字段的质量直接影响模型选择工具的准确率。模糊的描述会导致模型在相似工具间误选，建议在描述中明确工具的使用场景、输入约束和输出格式。

多工具并行调用与依赖图调度实现

当Agent面对复杂任务时，往往需要连续调用多个工具。如果工具之间没有数据依赖，并行调用能显著减少整体延迟。实现并行调度的核心是构建工具调用的DAG（有向无环图），识别可并行的节点。

import asyncio
from typing import Dict, List, Any

class ToolExecutor:
    def __init__(self, registry: Dict[str, callable]):
        self.registry = registry

    async def execute_single(self, tool_name: str, args: dict) -> Any:
        tool_fn = self.registry[tool_name]
        try:
            result = await tool_fn(**args)
            return {"tool": tool_name, "status": "success", "result": result}
        except Exception as e:
            return {"tool": tool_name, "status": "error", "error": str(e)}

    async def execute_parallel(self, calls: List[dict]) -> List[Any]:
        tasks = [
            self.execute_single(call["name"], call["arguments"])
            for call in calls
        ]
        return await asyncio.gather(*tasks)

    async def execute_with_retry(
        self, tool_name: str, args: dict,
        max_retries: int = 3, backoff: float = 1.0
    ) -> Any:
        for attempt in range(max_retries):
            result = await self.execute_single(tool_name, args)
            if result["status"] == "success":
                return result
            if attempt < max_retries - 1:
                await asyncio.sleep(backoff * (2 ** attempt))
        return result

依赖图调度器需要解析模型输出的多个tool_calls，将无依赖的调用归入同一批次并行执行，有依赖的调用按序执行。典型场景：先调用search_database获取数据，再调用generate_chart基于结果绘图——这两步存在数据依赖，必须串行；而同时搜索两个不同表的操作则可以并行。

工具调用结果回传与上下文窗口管理

每次工具调用完成后，结果需要以tool角色消息回传给模型。但频繁的工具调用会快速消耗上下文窗口token额度。生产环境中需要设计结果摘要策略：

def format_tool_result(call_id: str, result: dict, max_length: int = 2000) -> dict:
    content = json.dumps(result, ensure_ascii=False)
    if len(content) > max_length:
        content = content[:max_length] + "...[truncated, full result stored in cache]"
    return {
        "role": "tool",
        "tool_call_id": call_id,
        "content": content
    }

对于返回大量数据的工具（如数据库查询），建议在工具执行层做结果裁剪：只保留前N行数据、只返回关键字段、将大结果集写入缓存并返回摘要引用。模型拿到摘要后如果需要详细信息，可以通过另一个工具调用从缓存中读取。

ReAct推理循环与自我纠错机制

ReAct（Reasoning + Acting）模式是Agent框架的主流推理范式。每次循环包含三个阶段：Thought（分析当前状态和下一步动作）、Action（调用工具）、Observation（观察工具返回结果）。当工具调用失败时，Agent需要具备自我纠错能力：

class ReActAgent:
    def __init__(self, llm_client, tool_executor, max_iterations=10):
        self.llm = llm_client
        self.executor = tool_executor
        self.max_iterations = max_iterations

    async def run(self, user_query: str) -> str:
        messages = [{"role": "user", "content": user_query}]
        for i in range(self.max_iterations):
            response = await self.llm.chat(messages=messages, tools=tools)
            if not response.tool_calls:
                return response.content
            for call in response.tool_calls:
                result = await self.executor.execute_with_retry(
                    call.function.name,
                    json.loads(call.function.arguments)
                )
                messages.append(format_tool_result(call.id, result))
        return "Agent reached maximum iterations without completing the task."

自我纠错的关键在于：当工具返回错误信息时，错误描述需要足够详细，让模型理解失败原因并调整参数重试。比如数据库查询返回表不存在，模型应能切换到正确的表名或调用schema查询工具获取可用表列表。

生产环境安全防护与权限控制

工具调用框架在生产部署时必须考虑安全边界。核心原则是最小权限：每个工具只暴露必要的能力，参数严格校验，执行结果脱敏。具体措施包括：

工具执行沙箱化——文件操作限制在指定目录、网络请求限制目标域名白名单、代码执行使用容器隔离。参数注入防护——对所有字符串参数做SQL注入和命令注入检测。调用频率限制——单次Agent会话中同一工具调用次数上限，防止模型陷入死循环反复调用。人工确认门槛——高风险操作（如删除数据、发送邮件）在执行前暂停等待人工审批。

这些安全机制在框架层面统一实现，而不是在每个工具函数中重复编码，确保新增工具时默认获得安全保护。