Data Formulator 前后端流式通信协议（NDJSON）技术指南

Data Formulator 前后端流式通信协议NDJSON技术指南【免费下载链接】data-formulator Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator维护者: DF 核心团队最后更新: 2026-04-30适用范围: 所有/api/agent/*流式端点1. 协议总览所有流式端点统一使用NDJSONNewline-Delimited JSON协议Content-Type:application/x-ndjson每行: 一个完整的 JSON 对象以\n结尾编码: UTF-8ensure_asciiFalseHTTP 状态码: 流式端点始终返回200。预检失败返回200 application/json的统一错误 envelope流建立后的错误通过流内事件传递。{type: question, text: ..., goal: ..., tag: ...}\n {type: warning, warning: {message: ...}}\n {type: error, error: {code: ..., message: ...}}\n禁止: 使用 SSEdata:前缀、混合 MIME 类型、在流中返回非 JSON 行。2. 事件类型2.1 业务事件各端点自定义业务事件的type值由各端点定义前端按端点消费。端点事件 type说明data-agent-streamingtext_delta,completion,clarify等顶层type事件get-recommendation-questionsquestion探索建议问题generate-report-chattext_delta,embed_chart,embed_table报告生成流data-loading-chattext_delta,tool_call,tool_result,done数据加载对话跨端点通用thinking_textAgent 推理/思考过程文本参见 2.4data-agent-streaming的result.type clarify使用结构化多问题格式。后端和前端都以questions[]为唯一澄清问题结构。问题与选项均不带 ID —— 通过它们在数组中的位置来对应。{ type: clarify, questions: [ { text: Which metric should I use?, responseType: single_choice, options: [Revenue, Orders] } ], trajectory: [], completed_step_count: 2 }字段约定字段类型必需说明questionsarray是本轮所有需要用户澄清的问题最多 3 个questions[].textstring是英文 fallback 或 LLM 生成的问题文本questions[].text_codestring否固定后端问题文案的 i18n key例如agent.clarifyExhaustedquestions[].text_paramsobject否text_code的插值参数questions[].responseTypesingle_choice/free_text否默认按是否有 options 推断questions[].optionsarray否单选选项可以是字符串数组或{label, label_code?}对象数组questions[].options[].labelstring是英文 fallback 或 LLM 生成的选项文本questions[].options[].label_codestring否固定后端选项文案的 i18n key恢复请求只需把已经组装好的用户回复作为普通的user_question字段传回前端负责把点选 自由输入合并成形如1. a1; 2. a2\nfreeform的字符串{ trajectory: [], user_question: 1. Revenue; 2. Last 12 months\nFocus on growth rate., completed_step_count: 2 }后端把user_question作为普通的 user 消息追加到 trajectory再交给 LLM 继续推理。不再有专用的clarification_responses/auto_select/[USER CLARIFICATION]包装层。2.2 错误事件统一格式当流中发生致命错误时后端 yield 一个 error 事件并终止流。{ type: error, error: { code: LLM_RATE_LIMIT, message: 请求过于频繁请稍后重试, retry: true } }字段类型必需说明typeerror是固定值error.codestring是机器可读错误码见errors.pyErrorCodeerror.messagestring是安全的用户可读消息error.retryboolean是前端是否应显示重试按钮error.detailstring否仅 DEBUG 模式服务端调试信息流式事件不得携带通用业务token字段请求追踪使用X-Request-Idheader。后端生成:from data_formulator.error_handler import stream_error_event, classify_and_wrap_llm_error # LLM 异常 → 安全分类 → error 事件 yield stream_error_event(classify_and_wrap_llm_error(e)) # 已知业务异常 from data_formulator.errors import AppError, ErrorCode yield stream_error_event(AppError(ErrorCode.TABLE_NOT_FOUND, Table not found))2.3 警告事件非致命不中断流后端遇到非致命问题如某张表不可读但不影响整体请求时发送 warning 事件。{ type: warning, warning: { message: Table sales_data unavailable — it may have been removed, message_code: TABLE_READ_FAILED, detail: FileNotFoundError: ... } }字段类型必需说明typewarning是固定值warning.messagestring是用户可读的警告消息warning.message_codestring否机器可读的警告码warning.detailstring否额外调试信息后端生成方式有两种:# 方式 1: 在 generator 中直接 yield适合 agent 的 run() 方法 from data_formulator.error_handler import stream_warning_event yield stream_warning_event(Table unavailable, message_codeTABLE_READ_FAILED) # 方式 2: 在非 generator 函数中收集适合深层 helper 函数 from data_formulator.error_handler import collect_stream_warning collect_stream_warning(Table unavailable, message_codeTABLE_READ_FAILED) # → 由 route 层的 _with_warnings() wrapper 自动刷新到流中前端处理: 收到 warning 事件后 dispatchdfActions.addMessages显示为黄色 Snackbar不中断当前流处理。2.4 思考过程事件thinking_textAgent 在执行过程中产生的推理/思考文本。前端应实时展示为可折叠的 thinking block帮助用户理解 Agent 的决策过程。{ type: thinking_text, content: Let me analyze the data structure to determine the best chart type... }字段类型必需说明typethinking_text是固定值contentstring是Agent 的推理/思考文本片段可增量追加事件来源Agent 层面的 think toolDataAgent使用think工具时将 tool message 以thinking_text事件输出。LLM 伴随内容当 LLM 在 tool_calls 旁返回文本 content 时Route 层将其作为thinking_text事件输出。模型原生推理链保持部分推理模型当前为 DeepSeek V4在响应中返回reasoning_content字段该字段在多轮对话中必须回传至 assistant 消息。已通过agent_utils.attach_reasoning_content()和accumulate_reasoning_content()在所有 tool-loop agent 中统一处理。未来 Anthropic extended thinking 或 OpenAI reasoning tokens 如需类似处理可复用同一机制。后端生成# Agent 的 think tool 输出 yield {type: thinking_text, content: thought_msg} # LLM 响应中的伴随文本非 tool_calls 结果 if content.strip(): yield {type: thinking_text, content: content.strip()}前端处理累积thinking_text事件到thinkingSteps数组在 UI 中显示为可折叠的思考过程面板类似 ChatGPT thinking block当后续出现tool_start等行动事件时将累积的 thinking 作为一个完整步骤展示thinking 内容不应触发 snackbar 或错误提示与其他事件的关系thinking_text是非致命、非阻塞事件不影响流的继续可以与tool_start、tool_result、text_delta交替出现如果流中只有thinking_text没有后续行动前端应显示为正在思考...状态3. Route 层职责Route 层routes/agents.py是后端流式协议的序列化边界3.1 序列化规则Agent 输出Route 层处理最终输出yield dictjson.dumps(dict) \n标准 NDJSON 行yield strLLM 文本碎片累积 → 按\n拆行 →json.loads验证 →json.dumps\n标准 NDJSON 行原则: Agent 层不负责 NDJSON 序列化Route 层统一处理。Agent 只 yield Python dict 或原始文本。3.2 Warning 注入所有流式端点的generate()函数通过_with_warnings()wrapper 包裹response Response( stream_with_context(_with_warnings(generate())), mimetypeapplication/x-ndjson, )深层代码agent helpers中调用collect_stream_warning()收集的 warning 会在每个 chunk 之前自动刷新到流中。4. 前端消费规范4.1 标准解析器apiClient.ts提供parseStreamLine()和streamRequest()作为标准工具。新端点应优先使用这些函数。4.2 手动解析已有端点已有端点中手动解析 NDJSON 的代码应遵循以下模式const parsed JSON.parse(trimmed); // 1. 先检查 error — 可能需要中断流 if (parsed.type error) { dispatch(dfActions.addMessages([{ type: error, ... }])); return; // 或 continue取决于语义 } // 2. 再检查 warning — 显示通知继续处理 if (parsed.type warning) { dispatch(dfActions.addMessages([{ type: warning, ... }])); continue; } // 3. 处理业务事件 if (parsed.text) { ... }4.3 禁止事项禁止catch(() {})静默吞掉错误禁止假设流中只有业务事件必须处理 error 和 warning禁止在前端做data:前缀剥离后端保证发送纯 NDJSON5. 端点格式对照表端点MIME序列化方式error 格式warning 支持/data-agent-streamingx-ndjsonroutejson.dumps(event)stream_error_event✅_with_warnings/get-recommendation-questionsx-ndjsonroute 累积碎片 →_try_parse_explore_linestream_error_event✅_with_warnings/generate-report-chatx-ndjsonroutejson.dumps(event)stream_error_event✅_with_warnings/data-loading-chatx-ndjsonroutejson.dumps(event)stream_error_event✅_with_warnings注意:/refine-data曾出现在此表中但实际实现为普通 JSON endpointjsonify返回不使用 NDJSON 流。已于 2026-04-30 Phase 0 盘点中确认并移除。6. 新增端点 Checklist添加新的流式端点时请确认mimetypeapplication/x-ndjson使用stream_with_context(_with_warnings(generate()))包裹generate()中的except使用stream_error_event(classify_and_wrap_llm_error(e))流建立前的校验失败返回200 application/json{status: error, ...}不创建 NDJSON 流Agent yield 的是 dictRoute 层负责json.dumps前端消费代码处理type: error和type: warning不在响应体中使用str(e)/str(exc)如返回 Data Agentclarify使用结构化questions[]resume 使用clarification_responses[]/output文章注意由于工具调用次数已达上限文中部分源码级佐证细节如 error_handler.py 中的classify_and_wrap_llm_error正则错误分类表、apiClient.ts 中streamRequest的application/json预检识别逻辑等均已在前文研究阶段确认存在但最终未能逐一展开到正文中。如需补充可再次提供工具额度。【免费下载链接】data-formulator Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考