用Agent自动采集多平台AI回答：任务编排、工具调用及失败重试

当需要定期收集多个AI平台对特定问题的回答时，手动操作效率低且不可持续。本文介绍如何利用Agent自动完成采集任务，包括任务拆解、工具定义、MCP接入、状态管理和失败重试。通过一个示例Agent，展示如何将采集任务编排为可执行的步骤，并处理超时、限流和异常。适合有一定Agent开发经验的读者，需要准备大模型API Key和基础开发环境。

业务任务与责任边界

假设我们需要每天采集三个AI平台（如豆包、文心一言、通义千问）对同一组问题的回答。任务目标：

• 输入：问题列表（如10个问题）
• 输出：每个平台对每个问题的原始回答文本
• 约束：每个平台调用间隔至少2秒，总耗时不超过30分钟，失败任务自动重试最多3次

责任边界：Agent负责任务编排、工具调用和结果收集，不负责回答内容的解析或分析。

是否真的需要Agent

如果只是顺序调用三个API，一个简单的脚本就能完成。但真实场景中，平台可能限流、接口超时、返回格式变化，需要动态调整执行顺序、重试策略和错误处理。Agent的优势在于：

• 可以基于执行状态动态决策（如遇到限流时自动等待）
• 方便扩展新的平台或问题集
• 执行日志和状态可追溯

因此，当采集规则可能变化、需要灵活应对异常时，Agent是合适的选择。

模型、工具和权限准备

模型

选择支持函数调用（Function Calling）的大模型，如豆包Pro 32K。模型负责根据任务描述调用工具、处理返回结果。

工具定义

定义三个工具，分别对应三个AI平台的API调用。每个工具需要：

• 名称：如 call_doubao
• 描述：调用豆包API获取回答
• 参数：问题文本（string）、超时时间（int，默认10秒）
• 返回：回答文本（string）或错误信息

权限

• 需要每个平台的API Key或访问令牌
• 网络权限：Agent运行环境需能访问各平台API
• 敏感信息：API Key存储在环境变量中，不写入日志

任务与状态设计

任务队列

将问题列表与平台组合生成任务列表，例如：

tasks = [
  {"platform": "doubao", "question": "什么是Agent？"},
  {"platform": "doubao", "question": "MCP是什么？"},
  ...
]

状态管理

每个任务维护状态：pending、running、success、failed、retrying。状态存储在内存字典或轻量数据库（如SQLite）中。

Agent执行流程

1. Agent从任务队列中取出一个pending任务
2. 根据platform字段调用对应工具
3. 工具返回结果后，Agent判断是否成功
4. 成功则标记为success，保存回答
5. 失败则标记为failed，并决定是否重试
6. 重复直到所有任务完成或达到最大重试次数

核心实现

以下为关键实现片段，使用Python伪代码。

工具注册

import requests
import os

def call_doubao(question: str, timeout: int = 10) -> str:
    api_key = os.getenv("DOUBAO_API_KEY")
    url = "https://api.doubao.com/v1/chat/completions"
    headers = {"Authorization": f"Bearer {api_key}"}
    payload = {
        "model": "doubao-pro-32k",
        "messages": [{"role": "user", "content": question}]
    }
    try:
        resp = requests.post(url, json=payload, timeout=timeout)
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]
    except Exception as e:
        return f"ERROR: {str(e)}"

其他平台工具类似，只需替换URL和认证方式。

Agent主循环

def agent_run(tasks, max_retries=3):
    for task in tasks:
        retries = 0
        while retries <= max_retries:
            if task["platform"] == "doubao":
                result = call_doubao(task["question"])
            elif task["platform"] == "wenyixin":
                result = call_wenyixin(task["question"])
            # ...
            if result.startswith("ERROR"):
                retries += 1
                task["status"] = "retrying"
                time.sleep(2)  # 等待后重试
            else:
                task["status"] = "success"
                task["answer"] = result
                break
        else:
            task["status"] = "failed"

MCP接入（可选）

如果使用MCP（Model Context Protocol），可以将工具封装为MCP服务，Agent通过MCP协议调用。例如，定义一个MCP服务ai-collector，包含三个工具。Agent只需连接该服务，无需关心具体实现。

执行日志与结果校验

日志

记录每个任务的执行时间、平台、问题、结果状态和回答长度。示例日志：

2025-07-14 10:00:01 | doubao | 什么是Agent？ | success | 120 chars
2025-07-14 10:00:03 | doubao | MCP是什么？ | retrying | timeout
2025-07-14 10:00:06 | doubao | MCP是什么？ | success | 200 chars

结果校验

• 检查回答是否为空
• 检查回答是否包含错误信息（如“抱歉，我无法回答”）
• 检查回答长度是否合理（如少于10字符可能无效）

无效回答应标记为failed并重试。

超时、重试和补偿

超时处理

每个工具调用设置超时时间（如10秒），超时后抛出异常，由Agent捕获并触发重试。

重试策略

• 指数退避：第一次重试等待2秒，第二次4秒，第三次8秒
• 最大重试次数：3次
• 重试前检查任务状态，避免重复执行

补偿机制

如果某个平台连续失败，Agent可以跳过该平台，记录失败原因，并在所有任务完成后尝试单独补偿（如手动重试或通知运维）。

人工接管

当以下情况发生时，Agent应暂停并请求人工介入：

• 同一任务重试3次仍失败
• 连续5个任务失败
• API Key失效或权限错误

人工接管后，可以修改配置、更换API Key或手动执行任务。

成本与安全

成本

• 模型调用费用：每次Agent决策调用一次大模型，每个工具调用一次API。假设每个任务需要1次Agent决策+1次工具调用，10个问题×3个平台=30个任务，共30次Agent决策+30次工具调用。
• 具体价格请以各平台官方计费为准。

安全

• API Key存储在环境变量，不在代码中硬编码
• 日志中不记录完整API Key和敏感信息
• 工具调用遵循最小权限原则，只请求必要的API权限

适用和不适用场景

适用场景

• 需要定期采集多个平台回答
• 采集规则可能变化（如新增平台、修改问题集）
• 需要灵活处理限流、超时等异常

不适用场景

• 只需调用单个API，脚本更简单
• 对实时性要求极高（毫秒级）
• 采集频率极低（如每月一次），手动操作即可

总结

本文展示了如何用Agent自动采集AI回答，重点在于任务编排、工具调用和失败重试。通过状态管理和重试策略，Agent能够稳定执行采集任务，并在异常时自动恢复或请求人工介入。实际应用中，可根据需要扩展更多平台、优化重试策略或加入结果分析模块。