AI Agent 系统设计与多模态交互实验：接口怎么定才不返工

在开发 AI Agent 系统或多模态交互应用时，许多团队在前期的接口定义上踩过大坑。最常见的做法是按照传统 RESTful API 的思路，定义一个 POST /api/v1/agent/run 接口，入参传入用户 Prompt，响应体定义为一个简单的 {"result": "string"}。

这种“一次性同步返回结果”的接口设计，一旦遇到多轮 Tool Calling、长推理耗时或网络中途抖动，前端界面就会长时间陷入空白死等。更糟的是，当 Agent 中途需要前端用户二次确认（Human-in-the-loop）或上传图片补全上下文时，原有的接口结构瞬间崩塌，不得不推翻重来。

AI Agent 的接口设计，必须从“请求-响应”模式切换为**“流式事件驱动（Event-Driven Stream）”与状态机模式**。

一句“文本太长解析失败”，导致前后端联调折腾整整三天

某团队在开发一款代码生成 Agent 时，后端将 LLM 的思考过程、工具调用日志以及最终代码全压缩在一个 JSON 字段里返回给前端。

联调第一天，模型生成的代码里包含未转义的换行符与双引号，前端 JSON.parse() 抛出异常；第二天，Agent 执行了 3 次 SQL 查询，耗时 45 秒，浏览器触发 HTTP Timeout 导致连接断开；第三天，产品经理提出要求：“前端要实时看到 Agent 当前是在思考、还是在查数据库”。

原本以为半天就能搞定的 API 联调，因为接口定义缺少流式分块（Streaming Chunking）与事件类型（Event Type）划分，前后端整整折腾了三天。

一次性 JSON 响应 -> 大文本未转义解析崩溃 + 长耗时超时 -> 前端界面空白 -> 接口全盘废弃重构

Agent 接口必须在设计第一天就天生支持 SSE（Server-Sent Events）与颗粒化的 Event Protocol。

流式 SSE 传输下 Agent 状态事件（Event Pattern）的标准化划分

在一个标准规范的 Agent SSE 传输协议中，后端吐给前端的不能仅仅是 Token 文本，而必须是结构化的状态事件。

下表展示了生产环境 Agent 接口中标准的 Event 协议定义：

通过将 Agent 执行链路拆分为具体的 Event，前端可以精确获知 Agent 的每一步动作，彻底告别盲目等待。

状态机驱动的 Agent 接口交互时序图

在包含多模态与人工确认的 Agent 交互流程中，接口必须支持断线重连（Resume）与异步事件回调。

面向生产环境的 SSE Agent 交互接口与增量状态更新实现

下面使用 Python FastAPI 展示了一套符合上述规范的 Agent SSE 接口。它实现了流式 Generator、事件结构化打包以及异常中断捕获。

import asyncio
import json
import time
from typing import AsyncGenerator
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class AgentRequest(BaseModel):
    session_id: str
    user_prompt: str
    image_url: str = None

def format_sse_event(event_type: str, data: dict) -> str:
    """格式化符合 SSE 规范的文本块"""
    return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"

async def fake_agent_execution_stream(req: AgentRequest) -> AsyncGenerator[str, None]:
    """模拟 Agent 状态机流式推送"""
    start_time = time.time()
    
    # 1. 推送思考事件
    yield format_sse_event("agent_thought", {"delta": "收到请求，正在解析图片与文本..."})
    await asyncio.sleep(0.5)

    # 2. 推送工具调用开始事件
    yield format_sse_event("tool_start", {
        "tool_name": "visual_ocr_scanner",
        "input": {"image_url": req.image_url}
    })
    await asyncio.sleep(0.8) # 模拟 OCR 执行

    # 3. 推送工具调用结果
    yield format_sse_event("tool_result", {
        "tool_name": "visual_ocr_scanner",
        "status": "success",
        "extracted_text": "发票金额: ¥1280.00"
    })
    await asyncio.sleep(0.5)

    # 4. 模拟生成增量 Token
    final_text = "根据上传的发票，核销金额为 1280.00 元，已成功提交报销单。"
    for char in final_text:
        yield format_sse_event("agent_thought", {"delta": char})
        await asyncio.sleep(0.05)

    # 5. 推送完成事件
    yield format_sse_event("agent_finish", {
        "final_answer": final_text,
        "total_latency_seconds": round(time.time() - start_time, 2),
        "status": "completed"
    })

@app.post("/api/v1/agent/stream")
async def handle_agent_stream(req: AgentRequest, request: Request):
    """ Agent SSE 流式主接口 """
    async def event_generator():
        try:
            async for event in fake_agent_execution_stream(req):
                # 检查客户端是否中途断开连接
                if await request.is_disconnected():
                    print(f"[SSE Notice] 客户端 session {req.session_id} 主动断开连接，终止 Generator")
                    break
                yield event
        except Exception as e:
            # 捕获异常，推送 error 事件给前端，而不是直接断连
            yield format_sse_event("agent_error", {"code": 500, "message": str(e)})

    return StreamingResponse(event_generator(), media_type="text/event-stream")

这套代码的关键防线在于：通过 format_sse_event 约束事件格式，并使用 request.is_disconnected() 监控前端取消行为，防止前端关掉页面后后端 Agent 还在盲目调用 GPU 产生开销。

协议演进时的向下兼容策略与版本隔离

在 Agent 接口协议持续演进时，需要遵循以下 3 个向下兼容准则：

1. 增量事件扩展不破坏旧客户端：新增 event_type（如 agent_memory_retrieved）时，前端未识别的事件必须被默认 fallback 逻辑忽略，而不是抛异常崩溃。
2. 强制使用 Session ID 关联状态：任何 Agent 接口请求必须携带 session_id 字段，服务端将 Agent 的状态机上下文存入 Redis，确保在网络抖动重连时可以通过 GET /stream?session_id=xxx 恢复 Event 播放。
3. 敏感 Tool 必须独立为 ACK 接口：需要人工确认的 Tool 调用，不应在 SSE 长连接里直接阻塞等待，而要将其拆解为 human_approval_required 事件推送 + 独立 POST /action/approve HTTP 回调接口。