SSE 事件参考

// 每个事件由 event 行 + data 行组成，以空行结束
event:response.created
data:{"type":"response.created","response":{"id":"resp_abc123","status":"in_progress",...}}

event:response.output_item.added
data:{"type":"response.output_item.added","output_index":0,"item":{"id":"lc_run--019e0001-abcd-1234-ef00-000000000001","type":"message","status":"in_progress"}}

event:response.content_part.added
data:{"type":"response.content_part.added","output_index":0,"item":{"id":"lc_run--019e0001-abcd-1234-ef00-000000000001","type":"message"}}

event:response.output_text.delta
data:{"type":"response.output_text.delta","output_index":0,"content_index":0,"item_id":"lc_run--019e0001-abcd-1234-ef00-000000000001","delta":"你好"}

event:response.output_text.delta
data:{"type":"response.output_text.delta","output_index":0,"content_index":0,"item_id":"lc_run--019e0001-abcd-1234-ef00-000000000001","delta":"，今天"}

event:response.completed
data:{"type":"response.completed","response":{"id":"resp_abc123","status":"completed","output":[...],"usage":{...}}}

// 发送请求并读取 SSE 流
const response = await fetch('/api/ai/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ agent: 'my-agent', input: '你好' }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
let currentEvent = '';

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split('\n');
  buffer = lines.pop() || '';

  for (const line of lines) {
    if (line.startsWith('event:')) {
      currentEvent = line.slice(7);
    } else if (line.startsWith('data:')) {
      const data = JSON.parse(line.slice(6));
      handleEvent(currentEvent, data);  // 你的事件处理函数
    }
  }
}

response.output_item.added      ← 创建气泡占位（item.id 即后续 item_id）
response.content_part.added     ← content part 开始（富文本场景多个 part）
response.output_text.delta × N  ← 逐字追加，item_id 不变
response.output_text.done       ← text 字段含完整文本，可做最终校正
response.content_part.done      ← content part 结束
response.output_item.done       ← 气泡渲染完成，item.item() 含完整内容

event:custom
data:{"type":"task_running","task_id":"uuid-xxx","message":{...}}

// 或
event:custom
data:{"type":"llm_retry","attempt":2,"max_attempts":3,"message":"..."}

// thinking 也走 custom 通道，但被 SDK 拦截转换
event:custom
data:{"type":"thinking","data":"让我想一想..."}

SSE event="custom" 到达
  ↓
normalize() → 匹配 case "custom" → handleCustom(data)
  ↓
解析 JSON，取 root.get("type")
  ├── type="thinking"
  │     ↓ 拦截，提取 data 字段作为推理文本
  │     ↓ appendReasoningDelta(text)
  │     ↓ return REASONING_TEXT_DELTA  ← 不生成 CustomItem，走推理流程
  │
  └── type="task_running"
        ↓ 累加子代理 token（从 message.response_metadata.token_usage 提取）
        ↓ 不 return，继续往下

  其余任意 type 值（包括 task_running / llm_retry / 未来的新类型）
        ↓
        ↓ new CustomItem(null, "custom")
        ↓ 将整个 JSON 展开写入 raw（所有的 key-value 对）
        ↓ emit OUTPUT_ITEM_ADDED  ← CustomItem，不进入 Response.output

{
  "type": "task_running",
  "task_id": "uuid-of-subagent-task",   // 子代理任务 ID，前端按此去重/分组
  "message": {                          // 完整的 LangChain AIMessage 对象
    "type": "ai",                       // LangChain 消息类型
    "content": "正在查询猪场数据...",
    "id": "lc_run--019e0001-...",
    "response_metadata": {
      "model_name": "deepseek-v3",
      "finish_reason": "stop",
      "token_usage": {                  // ← SDK 从此处提取 token 累计到全局 usage
        "prompt_tokens": 150,
        "completion_tokens": 35,
        "total_tokens": 185
      }
    }
  },
  "message_index": 1,                  // 第几条消息（1-based），前端可展示序号
  "total_messages": 3                  // 当前已累积的消息总数
}

case 'response.output_item.added': {
  const item = data.item;
  if (item.type !== 'custom') break;

  // method 1: raw 通过 SSE data 中的 item.raw 字段取得
  // method 2: 将 data.item 作为 CustomItem，取 getRaw()["type"]
  const rawType = data.item.raw?.type;  // 或 item.raw["type"]

  if (rawType === 'task_running') {
    const { task_id, message, message_index } = data.item.raw;
    // 更新子任务进度卡片
    updateSubtask(task_id, {
      latestMessage: message,
      messageIndex: message_index,
    });
  }
  break;
}

{
  "type": "llm_retry",
  "attempt": 2,               // 当前是第几次重试（从 1 开始）
  "max_attempts": 3,          // 最多重试次数
  "wait_ms": 2000,            // 距离上一次失败等待的毫秒数
  "reason": "rate_limit",     // 重试原因（rate_limit / timeout / server_error）
  "message": "LLM 调用失败（限流），正在第 2/3 次重试（等待 2.0s）..."  // 人类可读提示
}

const rawType = data.item.raw?.type;

if (rawType === 'llm_retry') {
  const { attempt, max_attempts, message } = data.item.raw;
  toast.warning(message, { duration: 3000 });  // 弹出提示但不要阻断用户操作
}

// 通用 SSE 流解析器
async function streamChat(
  body: { agent: string; input: string; previous_response_id?: string },
  handlers: {
    onDelta?: (delta: string) => void;
    onReasoning?: (delta: string) => void;
    onToolCall?: (name: string, args: string) => void;
    onCompleted?: (response: any) => void;
    onError?: (message: string) => void;
  },
  signal?: AbortSignal
): Promise<void> {
  const response = await fetch('/api/ai/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(body),
    signal,
  });

  if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
  }

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  let currentEvent = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n');
    buffer = lines.pop() || '';

    for (const line of lines) {
      if (line.startsWith('event:')) {
        currentEvent = line.slice(7);
      } else if (line.startsWith('data:')) {
        const data = JSON.parse(line.slice(6));

        switch (currentEvent) {
          case 'response.output_text.delta':
            handlers.onDelta?.(data.delta);
            break;
          case 'response.output_text.done':
            if (data.text) handlers.onDelta?.(data.text);  // 完整文本替换增量拼接结果
            break;
          case 'response.reasoning_summary_text.delta':
            handlers.onReasoning?.(data.delta);
            break;
          case 'response.function_call_arguments.done':
            // item_id 对应 output_item.added 中的 item.id，name 需从彼处缓存
            handlers.onToolCall?.(data.item_id, data.arguments);
            break;
          case 'response.completed':
            handlers.onCompleted?.(data.response);
            break;
          case 'response.failed':
            handlers.onError?.(data.response?.error?.message || 'AI 服务异常');
            break;
          case 'error':
            handlers.onError?.(data.error?.message || '连接异常');
            break;
          default:
            // 未知事件类型，记录日志后忽略，不抛异常
            console.debug('未处理的事件:', currentEvent, data);
        }
      }
    }
  }
}

const params = new URLSearchParams({ agent: 'my-agent', input: '你好' });
const source = new EventSource(`/api/ai/chat/stream?${params}`);

let fullText = '';

source.addEventListener('response.output_text.delta', (e) => {
  const { delta } = JSON.parse(e.data);
  fullText += delta;
  renderText(fullText);  // 追加渲染
});

source.addEventListener('response.output_text.done', (e) => {
  const { text } = JSON.parse(e.data);
  if (text) renderText(text);  // text 字段含完整文本，以服务端结果为准
});

source.addEventListener('response.completed', (e) => {
  const { response } = JSON.parse(e.data);
  console.log('完成，tokens:', response.usage?.total_tokens);
  source.close();
});

source.addEventListener('response.failed', (e) => {
  const { response } = JSON.parse(e.data);
  showError(response.error?.message || 'AI 服务异常');
  source.close();
});

source.onerror = () => {
  showError('连接异常，请重试');
  source.close();
};

import { useState, useRef, useCallback } from 'react';

// ── 类型定义 ──
type Status = 'idle' | 'streaming' | 'completed' | 'failed' | 'incomplete';

interface ToolCallState {
  id: string; name: string; arguments: string; done: boolean;
}

interface BuiltinStatus {
  webSearch?: 'searching' | 'completed' | 'failed';
  fileSearch?: 'searching' | 'completed';
  codeInterpreter?: 'in_progress' | 'coding' | 'interpreting' | 'completed';
}

interface AiChatState {
  status: Status;
  answerText: string;           // 当前活跃 item 的累积文本
  currentItemId: string | null;  // 正在流式输出的 item_id
  reasoningText: string;
  refusalText: string;
  toolCalls: ToolCallState[];
  builtinStatus: BuiltinStatus;
  responseId: string | null;
  usage: { inputTokens: number; outputTokens: number } | null;
  error: string | null;
}

// ── Hook ──
export function useAiChat(agent: string) {
  const [state, setState] = useState<AiChatState>({
    status: 'idle', answerText: '', currentItemId: null, reasoningText: '', refusalText: '',
    toolCalls: [], builtinStatus: {}, responseId: null, usage: null, error: null,
  });

  const abortRef = useRef<AbortController | null>(null);
  // 用 ref 持有最新 responseId，避免 send 闭包因 React batching 读到旧 state
  const responseIdRef = useRef<string | null>(null);

  const cancel = useCallback(() => {
    abortRef.current?.abort();
    setState(prev => ({ ...prev, status: 'idle' }));
  }, []);

  const send = useCallback(async (input: string) => {
    cancel();
    setState(prev => ({ ...prev,
      status: 'streaming', answerText: '', currentItemId: null, reasoningText: '',
      refusalText: '', toolCalls: [], builtinStatus: {}, usage: null, error: null,
    }));

    const abortController = new AbortController();
    abortRef.current = abortController;

    try {
      const resp = await fetch('/api/ai/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        // previous_response_id 为业务 DTO 字段名，可自定义；SDK Builder 对应 previousResponseId()
        // 读 ref 而非 state，确保拿到最新值（绕过 React 18 batching 闭包陈旧问题）
        body: JSON.stringify({ agent, input, previous_response_id: responseIdRef.current }),
        signal: abortController.signal,
      });
      if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

      const reader = resp.body!.getReader();
      const decoder = new TextDecoder();
      let buffer = '', currentEvent = '';

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
          if (line.startsWith('event:')) {
            currentEvent = line.slice(7);
          } else if (line.startsWith('data:')) {
            const data = JSON.parse(line.slice(6));
            // ref 在 setState 外更新（updater 内不应有副作用）
            if (currentEvent === 'response.created' && data.response?.id) {
              responseIdRef.current = data.response.id;
            }
            setState(prev => {
              const s = { ...prev };

              switch (currentEvent) {
                // ── 生命周期 ──
                case 'response.created':
                  s.responseId = data.response?.id; // 同步到 state 供 UI 展示
                  break;
                case 'response.in_progress':
                  s.status = 'streaming';
                  break;

                // ── 文本增量 ──
                // item_id 串行出现（DeerFlow 顺序执行），新 item 直接覆盖旧文本
                case 'response.output_text.delta': {
                  const id = data.item_id;
                  if (id && id !== s.currentItemId) {
                    s.currentItemId = id; // 新 item：切换并重置文本
                    s.answerText = data.delta || '';
                  } else {
                    s.answerText += data.delta || '';
                  }
                  break;
                }
                case 'response.output_text.done':
                  // 用服务端校正后的完整文本覆盖增量累积结果（可选）
                  if (data.text) s.answerText = data.text;
                  break;
                case 'response.output_text.annotation.added':
                  // 可在此收集引用/脚注列表，与文本位置关联渲染
                  break;

                // ── 拒绝 ──
                case 'response.refusal.delta':
                  s.refusalText += data.delta || '';
                  break;
                case 'response.refusal.done':
                  if (data.delta) s.refusalText = data.delta;
                  break;

                // ── 推理 ──
                case 'response.reasoning_text.delta':
                  // 完整思维链（调试用），生产环境可跳过渲染
                  break;
                case 'response.reasoning_summary_text.delta':
                  s.reasoningText += data.delta || '';
                  break;
                case 'response.reasoning_summary_text.done':
                  s.reasoningText += '\n\n——思考完毕——';
                  break;

                // ── 输出项 ──
                case 'response.output_item.added':
                  if (data.item?.type === 'function_call') {
                    s.toolCalls = [...s.toolCalls, {
                      id: data.item.id, name: data.item.name || '...',
                      arguments: '', done: false,
                    }];
                  }
                  break;
                case 'response.output_item.done':
                  if (data.item?.type === 'function_call') {
                    s.toolCalls = s.toolCalls.map(tc =>
                      tc.id === data.item.id ? { ...tc, done: true } : tc
                    );
                  }
                  break;
                case 'response.content_part.added':
                case 'response.content_part.done':
                  // 富文本场景可按需处理
                  break;

                // ── 工具调用 ──
                // item_id = 工具调用 id（与 output_item.added 的 item.id 对应）
                // name 在 output_item.added 时已写入 toolCalls，此处只追加 arguments
                case 'response.function_call_arguments.delta':
                  s.toolCalls = s.toolCalls.map(tc =>
                    tc.id === data.item_id
                      ? { ...tc, arguments: tc.arguments + (data.delta || '') }
                      : tc
                  );
                  break;
                case 'response.function_call_arguments.done':
                  s.toolCalls = s.toolCalls.map(tc =>
                    tc.id === data.item_id
                      ? { ...tc, arguments: data.arguments, done: true }
                      : tc
                  );
                  break;

                // ── 内置工具状态 ──
                case 'response.web_search_call.searching':   s.builtinStatus = { ...s.builtinStatus, webSearch: 'searching' }; break;
                case 'response.web_search_call.completed':  s.builtinStatus = { ...s.builtinStatus, webSearch: 'completed' }; break;
                case 'response.web_search_call.failed':      s.builtinStatus = { ...s.builtinStatus, webSearch: 'failed' }; break;
                case 'response.file_search_call.searching':  s.builtinStatus = { ...s.builtinStatus, fileSearch: 'searching' }; break;
                case 'response.file_search_call.completed': s.builtinStatus = { ...s.builtinStatus, fileSearch: 'completed' }; break;
                case 'response.code_interpreter_call.in_progress':   s.builtinStatus = { ...s.builtinStatus, codeInterpreter: 'in_progress' }; break;
                case 'response.code_interpreter_call_code.delta':    s.builtinStatus = { ...s.builtinStatus, codeInterpreter: 'coding' }; break;
                case 'response.code_interpreter_call_code.done':     // 代码完整，即将执行 break;
                case 'response.code_interpreter_call.interpreting':  s.builtinStatus = { ...s.builtinStatus, codeInterpreter: 'interpreting' }; break;
                case 'response.code_interpreter_call.completed':     s.builtinStatus = { ...s.builtinStatus, codeInterpreter: 'completed' }; break;

                // ── 终止 ──
                case 'response.completed':
                  s.status = 'completed';
                  s.usage = {
                    inputTokens: data.response?.usage?.input_tokens || 0,
                    outputTokens: data.response?.usage?.output_tokens || 0,
                  };
                  break;
                case 'response.failed':
                  s.status = 'failed';
                  s.error = data.response?.error?.message || 'AI 服务异常';
                  break;
                case 'response.incomplete':
                  s.status = 'incomplete';
                  break;
                case 'error':
                  s.status = 'failed';
                  s.error = data.error?.message || '连接异常';
                  break;

                default:
                  console.debug('未处理事件:', currentEvent, data);
              }
              return s;
            });
          }
        }
      }
    } catch (err: any) {
      if (err.name !== 'AbortError') {
        setState(prev => ({ ...prev, status: 'failed', error: err.message }));
      }
    }
  }, [agent, cancel]);  // responseIdRef 不需要列入 deps（ref 变化不触发重渲染）

  return { ...state, send, cancel };
}

function ChatPanel() {
  const {
    status, answerText, reasoningText, refusalText,
    toolCalls, builtinStatus, usage, error, send, cancel,
  } = useAiChat('pig-farm-agent');

  const isStreaming = status === 'streaming';

  return (
    <div className="chat-panel">
      {/* 错误提示 */}
      {error && <div className="error-banner">{error}</div>}

      {/* 内置工具状态 */}
      {builtinStatus.webSearch === 'searching' && <Spinner label="搜索中..." />}
      {builtinStatus.webSearch === 'failed'    && <Hint>搜索失败，基于已有知识回答</Hint>}
      {builtinStatus.fileSearch === 'searching' && <Spinner label="检索文档..." />}
      {builtinStatus.codeInterpreter && <CodeStatusBadge status={builtinStatus.codeInterpreter} />}

      {/* 拒绝回答 */}
      {refusalText && <div className="refusal-box">{refusalText}</div>}

      {/* 推理面板 */}
      {reasoningText && (
        <details open>
          <summary>AI 思考中...</summary>
          <p>{reasoningText}</p>
        </details>
      )}

      {/* 回答区 */}
      <Markdown content={answerText} />

      {/* 工具调用 */}
      {toolCalls.map(tc => (
        <ToolCard key={tc.id} name={tc.name} args={tc.arguments} loading={!tc.done} />
      ))}

      {/* 底部状态 */}
      {isStreaming && <Spinner />}
      {status === 'incomplete' && <Hint>回答因长度限制被截断</Hint>}
      {usage && <div className="usage">消耗 {usage.inputTokens + usage.outputTokens} tokens</div>}

      {/* 操作 */}
      <InputBox onSend={send} disabled={isStreaming} />
      <button onClick={cancel} disabled={!isStreaming}>取消</button>
    </div>
  );
}

// 首轮对话
let lastResponseId = null;

async function chat(userInput) {
  const body = {
    agent: 'my-agent',
    input: userInput,
    // 非首轮时传入上一轮 responseId
    ...(lastResponseId && { previous_response_id: lastResponseId }),
  };

  const response = await fetch('/api/ai/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(body),
  });

  // ... 解析 SSE 流 ...

  // 在 response.created 事件中更新：
  // lastResponseId = data.response.id;
}

try {
  await streamChat(body, {
    onCompleted(response) {
      if (response.status === 'incomplete') {
        showWarning('回答因长度限制被截断，建议精简提问');
      }
    },
    onError(message) {
      // 区分是否可重试
      if (message.includes('超时') || message.includes('连接')) {
        showRetryable('网络异常，请重试');
      } else {
        showError(message);
      }
    },
  }, abortSignal);
} catch (err) {
  if (err.name === 'AbortError') {
    // 用户取消，不做任何处理
  } else {
    showError('请求失败: ' + err.message);
  }
}

function debounce(fn, ms) {
  let timer;
  return (...args) => { clearTimeout(timer); timer = setTimeout(() => fn(...args), ms); };
}

const renderMarkdown = debounce((text) => {
  element.innerHTML = marked.parse(text);  // 或其他 Markdown 渲染库
}, 50);  // 50ms 防抖，平衡流畅度与性能

// ✅ 推荐：按事件名监听，自动兼容新事件
source.addEventListener(eventName, handler);

// ✅ 也可以：switch 带 default 分支
switch (currentEvent) {
  case 'response.output_text.delta': /* ... */ break;
  case 'response.completed':       /* ... */ break;
  default:
    console.debug('未知事件:', currentEvent);  // 记录日志后忽略，不抛异常
}

// ❌ 错误：switch 没有 default，未知事件导致静默丢弃
// ❌ 错误：default 分支 throw error，一条未知事件就中断整个流

async function chatWithRetry(agent, input, maxRetries = 3) {
  let lastResponseId = null;
  let collectedText = '';

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      await streamChat(
        {
          agent,
          input: attempt === 0 ? input : collectedText || input,
          previous_response_id: lastResponseId,  // 续接上下文
        },
        {
          onCreated(id) { lastResponseId = id; },
          onDelta(delta) { collectedText += delta; render(collectedText); },
          onCompleted() { return; },  // 成功，退出重试循环
        }
      );
      break;  // 正常完成，退出重试
    } catch (err) {
      if (attempt === maxRetries - 1) {
        showError('多次重试失败，请稍后再试');
      } else {
        // 指数退避：1s, 2s, 4s...
        await new Promise(r => setTimeout(r, Math.pow(2, attempt) * 1000));
      }
    }
  }
}

async function ask(message, file, previousResponseId, ui) {
  const form = new FormData();
  form.append('message', message);
  if (file) form.append('file', file);                 // <input type=file> 选中的 File
  if (previousResponseId) form.append('previousResponseId', previousResponseId);

  const res = await fetch('/ai/chat', { method: 'POST', body: form });
  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buf = '';

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });

    // SSE 以空行分隔事件块
    let idx;
    while ((idx = buf.indexOf('\n\n')) >= 0) {
      const block = buf.slice(0, idx); buf = buf.slice(idx + 2);
      const line = block.split('\n').find(l => l.startsWith('data:'));
      if (!line) continue;
      const data = JSON.parse(line.slice(5).trim());
      handle(data, ui);
    }
  }
}

function handle(data, ui) {
  switch (data.type) {
    case 'response.created':
      ui.previousResponseId = data.response?.id; break;       // 存起来供续话
    case 'response.output_text.delta':
      ui.appendText(data.delta); break;                       // 增量正文
    case 'response.output_text.annotation.added': {        // 附件信号！
      const a = data.annotation;
      if (a && a.file_id) ui.addAttachment(a.file_id, a.filename);
      break;
    }
    case 'response.completed':
      ui.done(); break;
    case 'error':
      ui.error(data.message); break;
    default: break;                                          // 未知事件忽略（前向兼容）
  }
}

function addAttachment(fileId, filename) {
  const q = 'fileId=' + encodeURIComponent(fileId);
  const previewUrl  = '/ai/file?' + q + '&mode=preview';
  const downloadUrl = '/ai/file?' + q + '&mode=download';

  const el = document.createElement('div');
  el.className = 'attachment';
  el.innerHTML =
    '📎 ' + escapeHtml(filename) +
    ' <a target="_blank" href="' + previewUrl  + '">预览</a>' +
    ' <a href="' + downloadUrl + '" download>下载</a>';
  document.getElementById('attachments').appendChild(el);
}