GenAI长任务处理：ContinuationToken原理与实践

1. 项目概述

在开发基于GenAI的应用时，我们经常遇到一个棘手的工程问题：如何处理长时间运行的AI任务？想象一下，当用户请求"写一本关于火星殖民的长篇小说"或"分析50份PDF文档并总结结论"时，传统的Web请求模型很快就会遇到瓶颈。

1.1 核心痛点分析

Web服务通常设计为无状态和短生命周期的，这与GenAI任务的长耗时特性形成了根本性冲突。具体表现为：

• HTTP超时：大多数Web服务器和负载均衡器默认的超时设置在30-90秒，而复杂AI任务可能需要几分钟甚至更长时间
• 连接中断风险：用户可能随时关闭浏览器或移动应用，导致任务中断
• 状态丢失：服务重启或扩展时，正在执行的任务状态难以保持
• 资源占用：长时间保持连接会消耗宝贵的服务器资源

1.2 解决方案概览

微软的Agent Framework提供了一种优雅的解决思路：通过ContinuationToken机制将长任务拆分为多个可恢复的短任务。这种方案的核心优势在于：

• 保持Web服务的无状态特性
• 允许任务在后台持续执行
• 支持从任意断点恢复
• 与现有基础设施兼容

2. 技术实现详解

2.1 环境准备与初始化

首先需要安装必要的NuGet包：

bash复制dotnet add package Azure.AI.OpenAI --version 2.8.0-beta.1
dotnet add package Azure.Identity --version 1.17.1
dotnet add package Microsoft.Agents.AI.Hosting.OpenAI --version 1.0.0-alpha.251219.1
dotnet add package Microsoft.Agents.AI.OpenAI --version 1.0.0-preview.251219.1

初始化Agent时，关键是要使用Responses API而非传统的Chat API：

csharp复制AIAgent agent = new AzureOpenAIClient(
    new Uri(endpoint),
    new AzureCliCredential())
     .GetResponsesClient(deploymentName)
     .CreateAIAgent(
        name: "SpaceNovelWriter",
        instructions: @"你是一名太空题材小说作家...",
        tools: [
            AIFunctionFactory.Create(ResearchSpaceFactsAsync),
            AIFunctionFactory.Create(GenerateCharacterProfilesAsync)
        ]);

AgentRunOptions options = new()
{
    AllowBackgroundResponses = true
};

2.2 任务执行循环设计

核心的执行循环实现了"存盘-读盘"机制：

csharp复制AgentRunResponse response = await agent.RunAsync("写一本超长的太空小说...", thread, options);

while (response.ContinuationToken is not null)
{
    // 1. 保存当前状态
    PersistAgentState(thread, response.ContinuationToken);
    
    // 2. 模拟断开连接
    await Task.Delay(TimeSpan.FromSeconds(10));
    
    // 3. 恢复状态
    RestoreAgentState(agent, out thread, out ResponseContinuationToken? continuationToken);
    
    // 4. 继续执行
    options.ContinuationToken = continuationToken;
    response = await agent.RunAsync(thread, options);
}

2.3 状态持久化实现

在实际应用中，状态持久化通常使用数据库或Redis：

csharp复制// 示例使用Entity Framework Core实现
public async Task PersistAgentState(AgentThread thread, ResponseContinuationToken token)
{
    using var db = new AgentDbContext();
    var state = new AgentState
    {
        ThreadId = thread.Id,
        ContinuationToken = JsonSerializer.Serialize(token),
        LastUpdated = DateTime.UtcNow
    };
    
    await db.AgentStates.Upsert(state)
        .On(a => a.ThreadId)
        .RunAsync();
}

public void RestoreAgentState(AIAgent agent, out AgentThread thread, out ResponseContinuationToken? token)
{
    // 从数据库恢复实现...
}

3. 架构深度解析

3.1 Responses API vs Chat Completions API

两种API的关键区别：

3.2 框架调用链分析

Responses API的调用流程：

1. GetResponsesClient()返回AzureResponsesClient
2. 通过扩展方法CreateAIAgent()创建Agent
3. 内部转换为IChatClient接口
4. 最终创建ChatClientAgent实例

关键源码片段：

csharp复制// AzureOpenAIClient.cs
public override ResponsesClient GetResponsesClient(string deploymentName)
{
    return new AzureResponsesClient(Pipeline, deploymentName, _endpoint, _options);
}

// OpenAIResponseClientExtensions.cs
public static ChatClientAgent CreateAIAgent(
    this ResponsesClient client, ...)
{
    var chatClient = client.AsIChatClient();
    return new ChatClientAgent(chatClient, options, loggerFactory, services);
}

4. 生产环境实践要点

4.1 性能优化建议

• 批处理大小：根据任务复杂度调整每次恢复后处理的数据量
• 心跳机制：定期保存进度，即使没有显式中断
• 资源监控：跟踪长时间运行任务的内存和CPU使用情况
• 超时设置：合理配置各环节超时，平衡响应性和资源利用率

4.2 错误处理策略

建议实现以下错误处理机制：

csharp复制try
{
    while (response.ContinuationToken is not null)
    {
        // ...执行循环
        
        // 添加重试逻辑
        var retryPolicy = Policy
            .Handle<HttpRequestException>()
            .WaitAndRetryAsync(3, retryAttempt => 
                TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)));
            
        await retryPolicy.ExecuteAsync(async () => 
        {
            response = await agent.RunAsync(thread, options);
        });
    }
}
catch (OperationCanceledException)
{
    // 处理取消逻辑
    await SaveCheckpointForRecovery();
}

4.3 安全考量

• 令牌安全：ContinuationToken应加密存储
• 访问控制：验证恢复请求的权限
• 数据隔离：确保不同租户/用户的状态隔离
• 清理机制：对僵尸任务实现自动回收

5. 扩展应用场景

5.1 文档处理流水线

csharp复制AIAgent docAgent = new AzureOpenAIClient(...)
    .GetResponsesClient("gpt-4-doc")
    .CreateAIAgent(
        name: "DocumentProcessor",
        instructions: "分析上传的文档，提取关键信息...",
        tools: [DocumentAnalysisTool, SummaryGenerationTool]);

5.2 多步骤工作流

csharp复制// 定义工作流步骤
var workflow = new WorkflowBuilder()
    .AddStep("research", ResearchStepAsync)
    .AddStep("draft", DraftingStepAsync)
    .AddStep("review", ReviewStepAsync)
    .Build();

// 执行并持久化每个步骤的状态
foreach (var step in workflow.Steps)
{
    var result = await step.ExecuteAsync(context);
    await PersistWorkflowState(context);
}

5.3 分布式任务协调

结合Azure Durable Functions实现：

csharp复制[FunctionName("OrchestrateAgent")]
public static async Task RunOrchestrator(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var state = context.GetInput<AgentState>();
    
    while (!context.IsReplaying)
    {
        state = await context.CallActivityAsync<AgentState>(
            "RunAgentStep", state);
            
        await context.CreateTimer(
            context.CurrentUtcDateTime.AddSeconds(10),
            CancellationToken.None);
    }
}

6. 性能对比数据

我们在测试环境中对比了不同实现方式的性能：

7. 常见问题排查

7.1 ContinuationToken失效

症状：恢复任务时收到无效令牌错误
可能原因：

• 令牌过期（默认24小时）
• 底层模型变更
• 序列化/反序列化问题
  解决方案：

csharp复制// 检查令牌有效期
if (token.ExpiresAt < DateTime.UtcNow)
{
    // 重新初始化任务
    return await StartNewTaskAsync();
}

7.2 工具调用失败

症状：Agent卡在工具调用阶段
排查步骤：

1. 检查工具方法的可访问性
2. 验证输入参数格式
3. 查看执行日志中的错误详情
   示例修复：

csharp复制// 确保工具方法正确处理异步
[AITool]
public static async Task<ResearchResult> ResearchSpaceFactsAsync(
    [AIParam]string topic)
{
    // 添加超时处理
    using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    return await ExternalService.ResearchAsync(topic, cts.Token);
}

7.3 状态恢复不一致

症状：恢复后Agent行为异常
解决方案：

csharp复制// 在持久化时保存完整上下文
public async Task PersistFullState(AgentThread thread)
{
    var state = new {
        Thread = thread,
        Tools = GetCurrentToolsState(),
        LLMContext = CaptureModelState()
    };
    
    await distributedCache.SetAsync(
        $"agent:{thread.Id}", 
        state, 
        new DistributedCacheEntryOptions {
            SlidingExpiration = TimeSpan.FromHours(6)
        });
}

8. 进阶优化技巧

8.1 动态指令调整

根据执行阶段修改Agent行为：

csharp复制var progress = GetTaskProgress();
if (progress > 0.5)
{
    agent.UpdateInstructions(@"你现在进入精修阶段，请专注于...");
}

8.2 混合持久化策略

csharp复制// 根据数据特点选择存储
public async Task PersistSmartState(AgentState state)
{
    if (state.Size < 10_000)
    {
        await redis.StringSetAsync(state.Key, state.Data);
    }
    else
    {
        await blobStorage.UploadAsync(state.Key, state.Data);
    }
}

8.3 可视化监控

实现监控仪表板：

csharp复制// 示例使用Application Insights
public void LogAgentTelemetry(AgentRunResponse response)
{
    var telemetry = new Dictionary<string, string>
    {
        ["AgentName"] = response.AgentName,
        ["Step"] = response.CurrentStep,
        ["TokensUsed"] = response.Usage.TotalTokens.ToString()
    };
    
    telemetryClient.TrackEvent("AgentStepCompleted", telemetry);
}

9. 架构演进建议

随着业务规模扩大，考虑以下演进路径：

1. 水平扩展：使用Azure Load Balancer分发Agent请求
2. 垂直分层：
   • 接入层：处理即时响应
   • 工作层：执行长时间任务
   • 存储层：分布式状态存储
3. 混合部署：关键Agent本地部署，普通任务云端运行

示例架构：

mermaid复制graph TD
    A[客户端] --> B[API网关]
    B --> C{请求类型}
    C -->|即时| D[快速响应集群]
    C -->|长任务| E[Agent工作集群]
    E --> F[分布式存储]
    F --> G[Redis缓存]
    F --> H[Cosmos DB]

10. 成本优化方案

10.1 计算资源

• 冷热分离：频繁访问的状态存内存，历史数据存磁盘
• 自动缩放：基于队列深度动态调整Worker数量
• Spot实例：对中断不敏感的任务使用低成本VM

10.2 LLM调用

csharp复制// 根据任务复杂度选择模型
public string SelectModelForTask(string taskComplexity)
{
    return taskComplexity switch
    {
        "high" => "gpt-4-32k",
        "medium" => "gpt-4",
        "low" => "gpt-3.5-turbo",
        _ => throw new ArgumentOutOfRangeException()
    };
}

10.3 存储优化

• 压缩：对LLM上下文使用GZIP压缩
• 分片：大状态对象分割存储
• 生命周期：自动清理已完成任务状态

11. 替代方案对比

当Agent Framework不适用时，考虑这些替代方案：

12. 迁移指南

从传统实现迁移到Agent Framework的步骤：

1. 分析现有任务：

   • 识别长时间运行的操作
   • 标记状态保存点

2. 重构代码：

csharp复制// 之前
public async Task<Novel> WriteNovel(string prompt)
{
    // 一次性执行
    var novel = await llmService.GenerateLongContent(prompt);
    return novel;
}

// 之后
public async Task<AgentRunResponse> StartWritingNovel(string prompt)
{
    var response = await agent.RunAsync(prompt, thread, options);
    if (response.ContinuationToken != null)
    {
        await PersistState(thread, response);
    }
    return response;
}

3. 数据迁移：

   • 设计状态转换器
   • 批量转换现有任务

4. 监控调整：

   • 更新指标收集
   • 调整告警阈值

13. 调试技巧

13.1 本地调试配置

json复制// launchSettings.json
{
  "profiles": {
    "AgentDebug": {
      "commandName": "Project",
      "environmentVariables": {
        "ASPNETCORE_ENVIRONMENT": "Development",
        "AGENT_DEBUG_MODE": "true",
        "PERSISTENCE_INTERVAL": "5" // 秒
      }
    }
  }
}

13.2 状态检查端点

csharp复制[HttpGet("state/{threadId}")]
public async Task<IActionResult> GetAgentState(string threadId)
{
    var state = await stateStore.GetAsync(threadId);
    if (state == null) return NotFound();
    
    return Ok(new {
        threadId,
        state.ContinuationToken,
        lastActivity = state.LastUpdated,
        progress = CalculateProgress(state)
    });
}

13.3 诊断工具

推荐工具组合：

• Application Insights：端到端追踪
• Azure Monitor：性能指标
• Seq：结构化日志查询
• Postman：API测试

14. 测试策略

14.1 单元测试示例

csharp复制[Fact]
public async Task ShouldResumeFromBreakpoint()
{
    // 准备
    var mockAgent = new Mock<AIAgent>();
    var testToken = new ResponseContinuationToken("test");
    var testThread = new AgentThread("thread1");
    
    // 设置mock行为
    mockAgent.SetupSequence(a => a.RunAsync(It.IsAny<AgentThread>(), It.IsAny<AgentRunOptions>()))
        .ReturnsAsync(new AgentRunResponse { ContinuationToken = testToken })
        .ReturnsAsync(new AgentRunResponse { Content = "Completed" });
    
    // 执行
    var service = new AgentService(mockAgent.Object);
    var result = await service.ExecuteWithPersistenceAsync(testThread);
    
    // 验证
    Assert.Equal("Completed", result.Content);
    mockAgent.Verify(a => a.RunAsync(testThread, It.Is<AgentRunOptions>(o => o.ContinuationToken == testToken)), Times.Once);
}

14.2 集成测试要点

1. 测试状态持久化完整周期
2. 模拟网络中断恢复
3. 验证工具调用一致性
4. 压力测试长时间运行场景

14.3 混沌工程实验

建议注入的故障：

• 随机杀死进程
• 模拟存储延迟
• 断开网络连接
• 填充磁盘空间
• 随机拒绝服务

15. 性能调优

15.1 基准测试结果

在4核8G VM上的测试数据：

15.2 优化参数

关键配置建议：

csharp复制services.AddAgentFramework(options =>
{
    options.MaxConcurrentRequests = Environment.ProcessorCount * 2;
    options.RequestQueueLimit = 1000;
    options.DefaultCompletionOptions = new()
    {
        Temperature = 0.7,
        MaxTokens = 2048,
        FrequencyPenalty = 0.5
    };
});

15.3 缓存策略

实现响应缓存：

csharp复制public class CachedAgentService : IAgentService
{
    private readonly IAgentService _inner;
    private readonly IDistributedCache _cache;
    
    public async Task<AgentRunResponse> RunAsync(string input)
    {
        var cacheKey = $"agent:response:{input.MD5()}";
        var cached = await _cache.GetAsync(cacheKey);
        if (cached != null) return Deserialize(cached);
        
        var response = await _inner.RunAsync(input);
        await _cache.SetAsync(cacheKey, Serialize(response), new()
        {
            SlidingExpiration = TimeSpan.FromMinutes(30)
        });
        
        return response;
    }
}

16. 安全加固

16.1 认证授权

csharp复制[Authorize(Policy = "AgentAccess")]
[HttpPost("run")]
public async Task<IActionResult> RunAgent([FromBody] AgentRequest request)
{
    var userId = User.FindFirstValue(ClaimTypes.NameIdentifier);
    var thread = await _threadStore.GetOrCreateAsync(userId, request.ThreadId);
    
    // ...执行逻辑
}

16.2 输入验证

csharp复制public class AgentRequestValidator : AbstractValidator<AgentRequest>
{
    public AgentRequestValidator()
    {
        RuleFor(x => x.Input).NotEmpty().MaximumLength(1000);
        RuleFor(x => x.ThreadId).MaximumLength(50).Matches(@"^[a-zA-Z0-9_-]+$");
        RuleFor(x => x.Parameters).Must(BeValidJson);
    }
}

16.3 审计日志

csharp复制public async Task LogAgentActivity(AgentRunResponse response)
{
    var auditEntry = new {
        Timestamp = DateTime.UtcNow,
        User = GetCurrentUser(),
        Agent = response.AgentName,
        Input = SanitizeInput(response.Input),
        OutputMetadata = new {
            TokenUsage = response.Usage,
            Duration = response.Duration
        }
    };
    
    await _auditStore.AppendAsync(auditEntry);
}

17. 部署模式

17.1 容器化部署

示例Dockerfile：

dockerfile复制FROM mcr.microsoft.com/dotnet/aspnet:8.0 AS base
WORKDIR /app
EXPOSE 80

FROM mcr.microsoft.com/dotnet/sdk:8.0 AS build
WORKDIR /src
COPY ["AgentService.csproj", "."]
RUN dotnet restore "AgentService.csproj"
COPY . .
RUN dotnet build "AgentService.csproj" -c Release -o /app/build

FROM build AS publish
RUN dotnet publish "AgentService.csproj" -c Release -o /app/publish

FROM base AS final
WORKDIR /app
COPY --from=publish /app/publish .
ENTRYPOINT ["dotnet", "AgentService.dll"]

17.2 Kubernetes配置

示例Deployment：

yaml复制apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-service
  template:
    metadata:
      labels:
        app: agent-service
    spec:
      containers:
      - name: agent
        image: yourregistry/agent-service:latest
        ports:
        - containerPort: 80
        resources:
          limits:
            cpu: "2"
            memory: "4Gi"
          requests:
            cpu: "500m"
            memory: "1Gi"
        env:
        - name: ASPNETCORE_ENVIRONMENT
          value: Production
        - name: STATE_STORE__ENDPOINT
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: state-store-endpoint

18. 监控与告警

18.1 关键指标

建议监控的指标：

1. 执行时间：p50/p95/p99
2. 恢复成功率：按任务类型细分
3. 令牌使用：每请求平均消耗
4. 队列深度：等待恢复的任务数
5. 错误率：按错误类型分类

18.2 Grafana仪表板

示例查询：

sql复制SELECT
  floor($__time(time_utc)) as time,
  avg(duration_ms) as avg_duration,
  percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) as p95
FROM agent_runs
WHERE $__timeFilter(time_utc)
GROUP BY 1
ORDER BY 1

18.3 告警规则

推荐设置：

• 连续5分钟恢复失败率 > 1%
• 平均响应时间 > 5秒持续10分钟
• 内存使用 > 80%持续15分钟
• 连续错误 > 100次/分钟

19. 成本监控

19.1 成本分解

典型成本构成：

1. 计算资源：40-60%
2. LLM调用：30-50%
3. 存储：5-10%
4. 网络：<5%

19.2 优化建议

• 分时调度：非高峰时段缩减规模
• 模型选择：根据SLA要求动态调整
• 缓存策略：复用相似请求结果
• 资源预留：对稳定负载使用预留实例

19.3 预算控制

实现示例：

csharp复制public class CostAwareAgentMiddleware : IAgentMiddleware
{
    private readonly ICostTracker _costTracker;
    
    public async Task<AgentRunResponse> InvokeAsync(AgentRunContext context, NextMiddleware next)
    {
        if (_costTracker.MonthlyCost > context.BudgetLimit)
        {
            throw new BudgetExceededException();
        }
        
        var response = await next(context);
        
        _costTracker.TrackRequest(
            context.Model,
            response.Usage.TotalTokens,
            response.Duration);
            
        return response;
    }
}

20. 演进路线图

20.1 短期优化

1. 性能提升：

   • 优化状态序列化
   • 实现增量更新
   • 引入更高效的存储格式

2. 开发者体验：

   • 完善调试工具
   • 增强文档
   • 提供更多示例

20.2 中期规划

1. 多语言支持：

   • Python SDK
   • Java客户端
   • TypeScript集成

2. 生态系统：

   • 可视化编排工具
   • 模板市场
   • 插件系统

20.3 长期愿景

1. 智能调度：

   • 自动模型选择
   • 动态资源分配
   • 预测性扩展

2. 自适应Agent：

   • 运行时自我优化
   • 个性化调整
   • 自动问题修复

在实际项目中采用这种模式后，我们发现最大的收获不仅是解决了技术问题，更重要的是改变了团队对GenAI任务处理的思维方式。从"如何让请求不超时"转变为"如何设计可恢复的任务流"，这种范式转变带来了更健壮、更可扩展的架构。