解决Agency Swarm在FastAPI应用中事件循环冲突的技术方案

事件循环冲突问题背景

在将Agency Swarm框架集成到FastAPI应用时，开发者经常会遇到"RuntimeError: This event loop is already running"的错误。这个问题的根源在于FastAPI和Agency Swarm对异步事件循环的不同处理方式。

问题本质分析

FastAPI作为现代异步Web框架，本身已经运行在一个事件循环中。当Agency Swarm尝试在其内部方法get_completion()中创建新的事件循环时，就会与现有的事件循环产生冲突。特别是在多代理通信场景下，这种冲突会频繁出现。

典型错误场景

在典型的Agency Swarm多代理架构中，比如：

1. 用户请求通过AgreementManager代理处理
2. AgreementManager调用PlanfixAgent创建子任务
3. PlanfixAgent完成任务后返回结果
4. AgreementManager再调用CommunicationAgent生成通信内容
5. CommunicationAgent需要与PlanfixAgent交互获取任务详情

在第五步的跨代理通信中，最容易触发事件循环冲突，因为此时FastAPI的事件循环仍在运行，而Agency Swarm试图创建新的事件循环来处理异步操作。

解决方案实现

方案一：使用nest_asyncio补丁

最直接的解决方案是使用nest_asyncio库来修补事件循环，使其支持嵌套运行：

import nest_asyncio
from fastapi import FastAPI
from agency_swarm import Agency

# 初始化Agency Swarm
agency = Agency([...])  # 代理配置

# 创建FastAPI应用
app = FastAPI()

# 应用nest_asyncio补丁
nest_asyncio.apply()

@app.post("/completion")
async def get_completion(request_data: dict):
    response = agency.get_completion(request_data["message"])
    return {"response": response}

关键点：

1. 必须在创建FastAPI应用前应用补丁
2. 补丁只需要应用一次，通常放在主入口文件
3. 确保所有异步操作都在修补后的事件循环中运行

方案二：重构异步调用链

更优雅的解决方案是重构代码，避免在已有事件循环中创建新循环：

from agency_swarm.threads import Thread

@app.post("/completion")
async def get_completion(request_data: dict):
    thread = Thread(agency)
    async for message in thread.get_completion_async(request_data["message"]):
        # 处理流式响应
        pass
    return {"response": message}

这种方法利用了Agency Swarm原生的异步生成器接口，完全避免了事件循环冲突。

最佳实践建议

1. 统一事件循环管理：在整个应用中保持单一事件循环，避免混合使用同步和异步调用

2. 代理通信优化：对于复杂的多代理交互，考虑使用消息队列作为中间件，减少直接的事件循环依赖

3. 错误处理增强：在关键位置添加事件循环状态检查，提前预防冲突

4. 性能监控：在使用nest_asyncio时，注意监控性能指标，确保没有引入不必要的开销

结论

在FastAPI中集成Agency Swarm框架时，正确处理事件循环是关键。通过nest_asyncio补丁或重构异步调用链，可以有效解决事件循环冲突问题。选择哪种方案取决于具体应用场景和性能要求。对于大多数情况，nest_asyncio提供了快速简单的解决方案，而对于高性能要求的场景，重构异步调用链可能是更好的选择。