Agentic AI与生成式AI的协同架构设计与实战解析

1次阅读

共计 2382 个字符，预计需要花费 6 分钟才能阅读完成。

在智能客服系统中，Agentic AI（代理型 AI）负责意图识别和决策（比如判断用户想查询订单还是退货），而生成式 AI（Generative AI）则根据决策结果生成自然语言响应。这种协作模式也常见于智能写作助手（先确定写作风格再生成内容）、游戏 NPC 对话系统等场景。

两者的协同面临几个关键挑战：

延迟叠加 ：串行调用导致总响应时间 =Agentic AI 推理时间 + 生成式 AI 推理时间
状态同步困难 ：当生成式 AI 需要基于多轮对话历史生成内容时，如何保持上下文一致性
资源竞争 ：两类模型通常需要不同的硬件加速策略（如 Agentic AI 可能用 CPU 推理，生成式 AI 需要 GPU）

# 伪代码示例 - 不推荐的实际做法
response = agentic_ai(query)
generated_text = generative_ai(response["intent"])

优点：
– 实现简单
– 强一致性保证

缺点：
– 阻塞式调用放大延迟
– 错误传播风险（任一环节失败整个流程中断）
– 难以扩展

我们推荐使用事件驱动架构（EDA）解耦两类 AI 服务：

sequenceDiagram
    participant Client
    participant AgenticService
    participant RedisStream
    participant GenerativeService

    Client->>AgenticService: POST /query
    AgenticService->>RedisStream: 发布意图分析事件
    RedisStream->>GenerativeService: 推送事件
    GenerativeService-->>Client: 异步返回生成结果

核心组件：
1. 事件总线 ：使用 Redis Streams 作为持久化消息队列
2. 去中心化处理 ：各服务独立扩缩容
3. 异步响应 ：通过 WebSocket 或长轮询返回最终结果

# 使用 asyncio + Redis Streams 的完整示例
import asyncio
import redis.asyncio as redis

class AIOrchestrator:
    def __init__(self):
        self.redis = redis.Redis()
        self.stream_key = "ai_events"

    async def handle_query(self, query):
        # 1. 触发 Agentic AI 处理
        intent = await self._call_agentic_ai(query)

        # 2. 发布到事件流
        await self.redis.xadd(self.stream_key, {
            "session_id": query.session_id,
            "intent": intent,
            "original_query": query.text
        })

        # 3. 返回临时响应
        return {"status": "processing"}

    async def event_consumer(self):
        while True:
            # 从 Stream 读取新事件
            events = await self.redis.xread(streams={self.stream_key: "$"},
                block=1000
            )

            for event in events:
                await self._call_generative_ai(event["intent"])

    # 实际 AI 调用方法省略...

对于生成式 AI 的高延迟特性，建议实现请求聚合：

async def batch_generate(texts):
    # 将多个请求合并为 batch
    combined = "\n[SEP]\n".join(texts)
    results = await generative_ai(combined)
    return results.split("\n[SEP]\n")

效果对比 ：
– 单次处理：50 requests/s (延迟 200ms)
– 批次大小 32：600 requests/s (延迟 210ms)

# 服务启动时预加载模型
@app.on_event("startup")
async def warmup():
    dummy_input = "warmup"
    await agentic_ai(dummy_input)
    await generative_ai(dummy_input)

关键设计：
– 为每个会话分配唯一 UUID
– 使用 Redis 的 SETNX 实现互斥锁

async def get_session(session_id):
    lock_key = f"lock:{session_id}"
    if not await redis.setnx(lock_key, 1):
        raise ConcurrentUpdateError()

    try:
        data = await redis.get(f"session:{session_id}")
        return json.loads(data)
    finally:
        await redis.delete(lock_key)

建议方案：
1. 使用 Redis 的 ACK 机制确保消息至少处理一次
2. 检查点（Checkpoint）保存处理进度
3. 死信队列处理失败消息