Agent全栈工程：打通从“大脑”到“指尖”的实时数据动脉

# FastAPI 异步端点示例：Agent 流式执行引擎
from fastapi.responses import StreamingResponse
import json
import asyncio

async def agent_execution_flow(user_input: str):
    # 1. 推送“计划阶段”状态
    yield f"data: {json.dumps({'type': 'status', 'content': '🧠 正在拆解任务...'})}\n\n"
    await asyncio.sleep(0.1)  # 模拟逻辑耗时
    
    # 2. 模拟工具调用（如查询数据库），并推送进度
    yield f"data: {json.dumps({'type': 'tool_call', 'tool': 'query_db', 'params': 'user_id=123'})}\n\n"
    
    # 3. 模拟流式文本生成（逐字吐出）
    result_text = "根据分析，2026年Q3的销售峰值出现在8月中旬..."
    for chunk in result_text.split(" "):
        yield f"data: {json.dumps({'type': 'text_delta', 'content': chunk + ' '})}\n\n"
        await asyncio.sleep(0.05)  # 模拟大模型生成速度
    
    # 4. 推送最终完成信号
    yield f"data: {json.dumps({'type': 'done'})}\n\n"

@app.get("/agent/stream")
async def stream_agent(query: str):
    return StreamingResponse(agent_execution_flow(query), media_type="text/event-stream")

// Vue 3 / React 中的核心流式消费逻辑
const agentState = reactive({
    status: 'idle',
    messages: [],      // 最终展示区
    toolCalls: [],     // 工具调用卡片区
    currentText: ''    // 正在流式追加的缓冲区
});

const connectAgent = async (query) => {
    const eventSource = new EventSource(`/agent/stream?query=${encodeURI(query)}`);
    
    eventSource.onmessage = (e) => {
        const payload = JSON.parse(e.data);
        
        // 工程化的全栈关键：前端路由分流
        switch (payload.type) {
            case 'status':
                agentState.status = payload.content;
                break;
            case 'tool_call':
                // 动态渲染一个带“确认/取消”按钮的交互卡片
                agentState.toolCalls.push({ 
                    id: Date.now(), 
                    ...payload,
                    humanConfirmed: false 
                });
                break;
            case 'text_delta':
                // 打字机效果追加
                agentState.currentText += payload.content;
                break;
            case 'done':
                agentState.messages.push(agentState.currentText);
                agentState.currentText = '';
                eventSource.close();
                break;
        }
    };
};

# backend/models.py - 后端定义（唯一真相源）
from enum import Enum

class AgentEventType(str, Enum):
    STATUS = "status"
    TOOL_CALL = "tool_call"
    TEXT_DELTA = "text_delta"
    ERROR = "error"
    DONE = "done"

// frontend/types/agent.ts - 前端手动同步（或通过CI自动生成）
// 务必与后端严格对应！
export type AgentEventType = "status" | "tool_call" | "text_delta" | "error" | "done";

from fastapi import BackgroundTasks, Request
import asyncio

# 全局存储正在运行的任务Future对象（生产环境需用Redis）
running_tasks = {}

@app.post("/agent/cancel/{task_id}")
async def cancel_agent_task(task_id: str):
    if task_id in running_tasks:
        running_tasks[task_id].cancel()  # 抛出 CancelledError
        return {"status": "cancelled"}
    return {"status": "not_found"}

# 在流式生成函数中捕获取消信号
async def agent_execution_flow(task_id: str):
    try:
        # ... 执行推理
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        yield f"data: {json.dumps({'type': 'error', 'content': '任务已被用户手动终止'})}\n\n"
        raise