Files
deerflow-code/frontend-web/src/roundtable-planning/api/intent.ts
T
2026-09-07 17:30:53 +08:00

637 lines
26 KiB
TypeScript

import { apiFetch } from "@/core/api/fetch-client";
import { getBackendBaseURL } from "@/core/config";
import type { ClarificationPayload } from "./multi-agent";
import { reportRoundtableError } from "./diagnostics-report";
import { resolveAllowMultiple } from "@/roundtable-planning/lib/clarification";
import { normalizeIntentReadySummary } from "@/roundtable-planning/lib/roundtable-constants";
export type { ClarificationPayload };
/** Stable built-in agent used by the multi-agent roundtable Step 1 flow. */
export const INTENT_AGENT_ID = "roundtable-intent";
/** Dedicated Step 1 agent for the position-roundtable workspace. */
export const POSITION_INTENT_AGENT_ID = "position-roundtable-intent";
/** Dedicated Step 1 agent for the 8BF roundtable workspace (same prompt family
* as `roundtable-intent`, independently tunable via the agent config UI). */
export const EIGHTBF_INTENT_AGENT_ID = "eightbf-roundtable-intent";
/**
* Task-intent agent (Step 1) API. Mirror of multi-agent.ts but with a single
* built-in agent and no per-call coordinator: one /init creates a thread
* tied to the built-in `roundtable-intent` agent, and each /stream POST is
* one user-turn of the clarification dialogue. The terminal status frame
* carries either `asking` (more turns needed) or `done` (structured summary
* available — hand off to Step 2).
*/
export interface IntentInitResponse {
status: "success";
agent_id: string;
thread_id: string;
}
/**
* Structured diagnostic payload surfaced by the backend whenever an intent
* route raises (init) or yields a final `status: "error"` frame (stream).
* Same shape across `init_intent` / `stream_intent`; renderer keys off `code`
* for quick visual triage, but also shows `hint` + `trace_tail` as a panel
* so deployment-time issues (config.yaml unreachable LLM, missing builtin
* agent dirs, loopback going through proxy) are debuggable from the UI alone
* without tail-ing backend logs.
*/
export interface IntentErrorDetail {
/** Stable short category — see backend `_diagnose_exception`. */
code: string;
/** One-line human-readable description (already includes exception type). */
message: string;
exception_type: string;
exception_repr: string;
/** Root of the `__cause__` chain when distinct from the outer exception. */
root_type?: string | null;
root_repr?: string | null;
/** Actionable hint in Chinese, keyed by `code`. May be null for unrecognized errors. */
hint?: string | null;
/** Last ~12 lines of the traceback for in-UI inspection. */
trace_tail?: string[];
/** Caller-supplied label (e.g. "intent_init"). */
context?: string | null;
}
/**
* Thrown by `initIntent` when the HTTP call fails. Carries the parsed
* `IntentErrorDetail` (if the backend emitted one) so the page can render
* a rich diagnostic panel instead of a one-line message.
*/
export class IntentInitError extends Error {
/** HTTP status (typically 500 for diagnosed failures, 401/403 for auth). */
readonly status: number;
/** Structured detail when the backend emitted `_diagnose_exception` output. */
readonly detail: IntentErrorDetail | null;
/** Raw body text when JSON parse failed (kept for completeness). */
readonly rawBody: string;
constructor(status: number, detail: IntentErrorDetail | null, rawBody: string) {
super(detail?.message || `init intent failed (${status}): ${rawBody}`);
this.name = "IntentInitError";
this.status = status;
this.detail = detail;
this.rawBody = rawBody;
}
}
function parseIntentErrorDetail(value: unknown): IntentErrorDetail | null {
if (!value || typeof value !== "object") return null;
const obj = value as Record<string, unknown>;
if (typeof obj.code !== "string" || typeof obj.message !== "string") return null;
return {
code: obj.code,
message: obj.message,
exception_type: typeof obj.exception_type === "string" ? obj.exception_type : "",
exception_repr: typeof obj.exception_repr === "string" ? obj.exception_repr : "",
root_type: typeof obj.root_type === "string" ? obj.root_type : null,
root_repr: typeof obj.root_repr === "string" ? obj.root_repr : null,
hint: typeof obj.hint === "string" ? obj.hint : null,
trace_tail: Array.isArray(obj.trace_tail) ? obj.trace_tail.filter((l): l is string => typeof l === "string") : [],
context: typeof obj.context === "string" ? obj.context : null,
};
}
/**
* Create a fresh thread for an intent-clarification session.
* The backend returns the built-in agent id so the frontend doesn't hardcode it.
*
* On failure throws `IntentInitError` whose `.detail` carries the backend's
* `_diagnose_exception` payload — the page should surface `detail.hint` /
* `detail.trace_tail` to help diagnose intranet deployment issues without
* shell access to the server.
*/
export async function initIntent(
model?: string,
taskContext?: Record<string, unknown> | null,
intentAgentId?: string,
): Promise<IntentInitResponse> {
const res = await apiFetch(`${getBackendBaseURL()}/api/intent/init`, {
method: "POST",
headers: { "Content-Type": "application/json" },
// 岗位会商在创建意图线程时就写入父任务标识。这样即使用户尚未确认
// 业务链,第一阶段的对话也已经可以按 taskId 恢复。
body: JSON.stringify({
model,
task_context: taskContext ?? undefined,
intent_agent_id: intentAgentId || undefined,
}),
});
if (!res.ok) {
const raw = await res.text().catch(() => "");
let detail: IntentErrorDetail | null = null;
try {
const parsed = JSON.parse(raw);
// FastAPI HTTPException(detail=dict) → { "detail": {...} }
if (parsed && typeof parsed === "object" && parsed !== null) {
detail = parseIntentErrorDetail((parsed as { detail?: unknown }).detail) || parseIntentErrorDetail(parsed);
}
} catch {
// body wasn't JSON — fall through with raw text only.
}
throw new IntentInitError(res.status, detail, raw);
}
return (await res.json()) as IntentInitResponse;
}
/**
* Structured task summary the intent agent emits when it judges the user's
* task is fully specified. The frontend feeds this into Step 2's initial
* leader prompt.
*/
export interface IntentSummary {
/** One-line statement of what the user wants accomplished. */
objective: string;
/** Hard constraints (budget / time / compliance / scope). At least 1 entry. */
constraints: string[];
/** Optional assumptions the agent is making on the user's behalf. */
assumptions?: string[];
/** Present only for the dedicated position-roundtable intent agent. */
coreGoals?: string[];
riskWarnings?: string[];
keyPoints?: string[];
strategicSignificance?: string[];
}
/** Final hand-off shape emitted only by the position-roundtable intent agent. */
export interface PositionIntentSummary {
coreGoals: string[];
riskWarnings: string[];
keyPoints: string[];
strategicSignificance: string[];
}
/**
* Same phases as streamMultiAgent so the UI can reuse one phase-label mapper.
* `connecting` ≈ HTTP sent, no frames yet; `connected` ≈ got `event: metadata`;
* `streaming` ≈ first AI text chunk; `tool_calling` ≈ ask_clarification or
* similar tool call (which usually means the run is about to interrupt and
* the user-visible text won't grow further this turn).
*/
export type IntentStreamPhase = "connecting" | "connected" | "streaming" | "tool_calling";
/**
* 用户上传文件的引用(已经过 `/api/threads/{thread_id}/uploads`)。透传给后端
* `additional_kwargs.files`,由 UploadsMiddleware 注入文件清单并解锁 read_file 等工具。
* 字段与普通聊天的 `FileInMessage` 对齐:`path` 用上传返回的 `virtual_path`。
*/
export interface IntentFileRef {
filename: string;
size: number;
path: string;
status?: "uploading" | "uploaded";
}
export interface StreamIntentRequest {
threadId: string;
message: string;
model?: string;
/** Explicit built-in Step-1 agent. Omit to retain the multi-agent default. */
intentAgentId?: string;
/**
* 本回合幂等键(Phase 4 单飞):同一 (threadId, clientTurnId) 在后端至多一个 run。
* 缺省由调用方(useStep1Intent)每条消息生成;同一回合的重试 / 双击复用它,后端据此
* 去重(在跑→丢弃重复请求 / 已完成→幂等回放),杜绝并发下双发意图 / 重复澄清卡。
*/
clientTurnId?: string;
/** Optional hidden business/task context injected into the model run. */
taskContext?: Record<string, unknown> | null;
/** Step 1 用户上传的参考文件(可选)。 */
files?: IntentFileRef[];
signal?: AbortSignal;
/** Fires for every AI text chunk parsed out of LangGraph messages-tuple frames. */
onTextDelta?: (text: string, messageId: string) => void;
/**
* Fires whenever the inferred SSE phase changes. The UI uses this to flip
* the placeholder bubble's caption from "调度中" → "已连接,等待响应" →
* "正在输出" so users see what's happening during the inevitable LLM
* first-token wait.
*/
onPhaseChange?: (
phase: IntentStreamPhase,
toolName?: string,
args?: Record<string, unknown>,
) => void;
/**
* Fires for every `ask_clarification` tool message parsed out of the live
* stream (the intent agent may emit several in a single turn). Lets the UI
* surface each assistance question in real-time instead of waiting for the
* terminal frame. `index` is the 0-based arrival order within this turn.
*/
onClarification?: (clarification: ClarificationPayload, index: number) => void;
}
export interface StreamIntentResult {
/** "asking" → keep the dialogue going; "clarification" → structured question card; "done" → summary is ready; "error" → backend failure. */
status: "asking" | "clarification" | "done" | "error";
/** Final assistant text for this turn (markdown, may contain the [INTENT_READY] block). */
content: string;
/** Populated only when status === "done". */
summary: IntentSummary | null;
/**
* Populated when the intent agent called ask_clarification — render option
* cards + custom input instead of treating the turn as plain text only.
* This is the *first* clarification (kept for backward compat); prefer
* `clarifications` which carries the full set.
*/
clarification: ClarificationPayload | null;
/**
* All `ask_clarification` questions the agent asked this turn, in arrival
* order. Collected live and reconciled against the terminal frame so a
* reconnect cannot lose cards. Empty unless status === "clarification".
*/
clarifications: ClarificationPayload[];
/**
* Populated only when status === "error" and the backend emitted a
* structured diagnostic (new in `_diagnose_exception`). UI uses this to show
* the same rich panel as init failures.
*/
errorDetail: IntentErrorDetail | null;
}
/**
* Recognise the terminal status frame the backend appends after the upstream
* LangGraph stream finishes. Must carry `status` and `content`; `summary`
* is optional.
*/
function isFinalIntentFrame(value: unknown): value is {
status: "asking" | "clarification" | "done" | "error";
content?: string;
summary?: IntentSummary | null;
question?: string;
clarification_type?: string;
clarification_context?: string;
options?: unknown[];
allow_custom?: boolean;
allow_multiple?: boolean;
clarifications?: unknown[];
error_detail?: unknown;
} {
if (!value || typeof value !== "object") return false;
if (!("status" in value) || !("content" in value)) return false;
const status = (value as { status: unknown }).status;
return (
status === "asking" ||
status === "clarification" ||
status === "done" ||
status === "error"
);
}
/**
* Reuse the same delta-extraction shape as multi-agent.ts so AI text streams
* smoothly into the frontend bubble. Returns null for tool / value / tool
* result frames.
*/
function extractAiTextDelta(parsed: unknown): { text: string; id: string } | null {
if (!Array.isArray(parsed) || parsed.length < 1) return null;
const chunk = parsed[0];
if (!chunk || typeof chunk !== "object") return null;
const obj = chunk as Record<string, unknown>;
const type = String(obj.type ?? "").toLowerCase();
if (type !== "ai" && type !== "aimessagechunk" && type !== "aimessage") return null;
const content = obj.content;
let text = "";
if (typeof content === "string") {
text = content;
} else if (Array.isArray(content)) {
for (const item of content) {
if (typeof item === "string") text += item;
else if (item && typeof item === "object") {
const block = item as Record<string, unknown>;
const blockText = block.text;
if (typeof blockText === "string") text += blockText;
}
}
}
if (!text) return null;
const id = typeof obj.id === "string" ? obj.id : "";
return { text, id };
}
/**
* Parse a live `type:"tool"` `ask_clarification` frame into a ClarificationPayload.
*
* The intent agent emits one such frame per question (ClarificationMiddleware
* formats them), each carrying a fully-formed clarification in
* `additional_kwargs.clarification` — options already normalized to
* `{id, label}` objects, snake_case keys. Returns null for any other frame
* (AI text chunks, tool_call_chunks, value snapshots).
*/
function extractClarificationToolMessage(
parsed: unknown,
): { payload: ClarificationPayload; toolCallId: string } | null {
if (!Array.isArray(parsed) || parsed.length < 1) return null;
const msg = parsed[0];
if (!msg || typeof msg !== "object") return null;
const obj = msg as Record<string, unknown>;
if (String(obj.type ?? "").toLowerCase() !== "tool") return null;
if (obj.name !== "ask_clarification") return null;
const kwargs = obj.additional_kwargs;
if (!kwargs || typeof kwargs !== "object") return null;
const clar = (kwargs as Record<string, unknown>).clarification;
if (!clar || typeof clar !== "object") return null;
const c = clar as Record<string, unknown>;
const toolCallId =
typeof obj.tool_call_id === "string" && obj.tool_call_id
? obj.tool_call_id
: typeof obj.id === "string"
? obj.id
: "";
const clarificationType =
typeof c.clarification_type === "string" ? c.clarification_type : "";
return {
toolCallId,
payload: {
id: toolCallId || undefined,
question: typeof c.question === "string" ? c.question : "",
clarificationType,
context: typeof c.context === "string" ? c.context : "",
options: Array.isArray(c.options) ? c.options : [],
allowCustom: c.allow_custom !== false,
allowMultiple: resolveAllowMultiple(
typeof c.allow_multiple === "boolean" ? c.allow_multiple : undefined,
clarificationType,
),
},
};
}
/**
* Normalize one clarification included in the terminal status frame. This is
* the reconnect / proxy fallback for cases where one or more live `tool`
* frames were not delivered to the browser.
*/
function parseTerminalClarification(
value: unknown,
): { payload: ClarificationPayload; toolCallId: string } | null {
if (!value || typeof value !== "object") return null;
const raw = value as Record<string, unknown>;
const clarificationType =
typeof raw.clarification_type === "string" ? raw.clarification_type : "";
const toolCallId = typeof raw.id === "string" ? raw.id : "";
return {
toolCallId,
payload: {
id: toolCallId || undefined,
question: typeof raw.question === "string" ? raw.question : "",
clarificationType,
context:
typeof raw.clarification_context === "string"
? raw.clarification_context
: typeof raw.context === "string"
? raw.context
: "",
options: Array.isArray(raw.options) ? raw.options : [],
allowCustom: raw.allow_custom !== false,
allowMultiple: resolveAllowMultiple(
typeof raw.allow_multiple === "boolean" ? raw.allow_multiple : undefined,
clarificationType,
),
},
};
}
/**
* Open an SSE connection to /api/intent/stream and resolve when the upstream
* stream completes. Each call is one user turn; reuse the same threadId
* across calls to keep the dialogue history.
*/
export async function streamIntent(req: StreamIntentRequest): Promise<StreamIntentResult> {
const res = await apiFetch(`${getBackendBaseURL()}/api/intent/stream`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
thread_id: req.threadId,
message: req.message,
model: req.model,
intent_agent_id: req.intentAgentId || undefined,
task_context: req.taskContext ?? undefined,
// Phase 4 单飞幂等键:缺省由后端兜底生成,但调用方应始终传(useStep1Intent 每条消息生成)。
client_turn_id: req.clientTurnId || undefined,
// 仅在有文件时带上,空数组后端也安全忽略。
files: req.files && req.files.length > 0 ? req.files : undefined,
}),
signal: req.signal,
});
if (!res.ok || !res.body) {
const text = await res.text().catch(() => "");
reportRoundtableError({
stage: "step1_intent", event: "intent_stream_http_error",
message: `第一步意图生成失败 (HTTP ${res.status}): ${text.slice(0, 500)}`,
httpStatus: res.status,
});
throw new Error(`intent/stream failed (${res.status}): ${text}`);
}
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let lastEvent = "";
// Single source of truth for the inferred SSE phase. The helper only fires
// the callback when the phase actually changes (or when a tool name is
// provided, since the same `tool_calling` phase may surface a new tool).
let currentPhase: IntentStreamPhase = "connecting";
req.onPhaseChange?.("connecting");
const setPhase = (
next: IntentStreamPhase,
toolName?: string,
args?: Record<string, unknown>,
) => {
if (next === currentPhase && !toolName) return;
currentPhase = next;
req.onPhaseChange?.(next, toolName, args);
};
const result: StreamIntentResult = {
status: "asking",
content: "",
summary: null,
clarification: null,
clarifications: [],
errorDetail: null,
};
// Clarifications collected live from `type:"tool"` frames, deduped by
// tool_call_id, then reconciled against the complete terminal list.
const seenClarificationIds = new Set<string>();
// A clarification tool call is a hard pause in the Step 1 conversation.
// Some model providers keep emitting text (and can even produce an
// INTENT_READY terminal frame) after the tool result. That text belongs to
// a continuation the user has not approved yet, so never let it replace the
// assistance card in the current turn.
let awaitingUserClarification = false;
const collectClarification = (payload: ClarificationPayload, toolCallId: string) => {
awaitingUserClarification = true;
const dedupeKey = toolCallId || `idx-${result.clarifications.length}`;
if (seenClarificationIds.has(dedupeKey)) return;
seenClarificationIds.add(dedupeKey);
const index = result.clarifications.length;
result.clarifications.push(payload);
if (!result.clarification) result.clarification = payload;
req.onClarification?.(payload, index);
};
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 rawLine of lines) {
const line = rawLine.trimEnd();
if (!line) continue;
if (line.startsWith("event:")) {
lastEvent = line.slice(6).trim();
// First `event: metadata` = the upstream LangGraph run was accepted.
// Flip out of "connecting" so the bubble label changes from
// "调度中" → "已连接,等待响应".
if (lastEvent === "metadata" && currentPhase === "connecting") {
setPhase("connected");
}
continue;
}
if (!line.startsWith("data:")) continue;
const payload = line.slice(5).trim();
if (!payload) continue;
let parsed: unknown;
try {
parsed = JSON.parse(payload);
} catch {
continue;
}
if (isFinalIntentFrame(parsed)) {
// Reconcile any terminal clarification cards, but preserve the first
// tool-call pause as the authoritative state for this turn. In
// particular, do not accept a later `done` frame / summary before the
// user has answered the displayed questions.
if (awaitingUserClarification) {
if (parsed.status === "clarification") {
const terminalCards = Array.isArray(parsed.clarifications)
? parsed.clarifications
.map(parseTerminalClarification)
.filter((item): item is { payload: ClarificationPayload; toolCallId: string } => item !== null)
: [];
if (terminalCards.length > 0) {
terminalCards.forEach((card, index) => {
collectClarification(card.payload, card.toolCallId || `terminal-${index}`);
});
}
}
continue;
}
result.status = parsed.status;
if (typeof parsed.content === "string") result.content = parsed.content;
if (parsed.summary && typeof parsed.summary === "object") {
result.summary = normalizeIntentReadySummary(parsed.summary);
}
if (parsed.status === "error") {
result.errorDetail = parseIntentErrorDetail(parsed.error_detail);
reportRoundtableError({
stage: "step1_intent", event: "intent_error_frame",
message: `第一步意图生成报错:${(result.content || "").slice(0, 800)}`,
detail: result.errorDetail,
});
}
if (parsed.status === "clarification") {
// Newer backend frames contain every card. Merge them even when a
// subset already arrived live: an SSE reconnect can otherwise leave
// the user with only the first card in a multi-question assistance
// round. Older backends still fall back to the legacy top-level card.
const terminalCards = Array.isArray(parsed.clarifications)
? parsed.clarifications
.map(parseTerminalClarification)
.filter((item): item is { payload: ClarificationPayload; toolCallId: string } => item !== null)
: [];
if (terminalCards.length > 0) {
terminalCards.forEach((card, index) => {
collectClarification(card.payload, card.toolCallId || `terminal-${index}`);
});
} else if (result.clarifications.length === 0) {
const fallback = parseTerminalClarification(parsed);
if (fallback) {
collectClarification(fallback.payload, fallback.toolCallId || "terminal-0");
}
}
}
continue;
}
if (lastEvent === "messages") {
// `ask_clarification` tool result frame → collect + surface in real-time.
// These carry the fully-formed clarification (one per question); the
// agent may emit several in a single turn.
const toolClar = extractClarificationToolMessage(parsed);
if (toolClar) {
collectClarification(toolClar.payload, toolClar.toolCallId);
continue;
}
// Do not show follow-up thinking / generated summary text while the
// user-assistance card is waiting for an answer. This also prevents a
// later AI delta from hiding the card through an [INTENT_READY] block.
if (awaitingUserClarification) continue;
// Phase inference from LangGraph messages-tuple frames. Tool calls
// (typically ask_clarification) flip phase to tool_calling; AI text
// chunks flip to streaming. Everything else leaves phase alone.
if (Array.isArray(parsed) && parsed.length > 0) {
const chunk = parsed[0];
if (chunk && typeof chunk === "object") {
const chunkObj = chunk as Record<string, unknown>;
const chunkType = String(chunkObj.type ?? "").toLowerCase();
const toolCalls = chunkObj.tool_calls;
const toolCallChunks = chunkObj.tool_call_chunks;
const content = chunkObj.content;
const hasToolCall =
(Array.isArray(toolCalls) && toolCalls.length > 0) ||
(Array.isArray(toolCallChunks) && toolCallChunks.length > 0);
if (hasToolCall) {
const firstCall = Array.isArray(toolCalls) && toolCalls.length > 0
? (toolCalls[0] as Record<string, unknown>)
: Array.isArray(toolCallChunks) && toolCallChunks.length > 0
? (toolCallChunks[0] as Record<string, unknown>)
: null;
const toolName = firstCall && typeof firstCall.name === "string" ? firstCall.name : "";
// 把 tool args 一并透传给 UI;onPhaseChange 在 useStep1Intent 里
// 会把 args 累积进 message.steps,让 MessageStepsCard 渲染路径 chip /
// 命令代码块等子节点(与主聊天 MessageGroup 视觉一致)。
const toolArgs =
firstCall && typeof firstCall.args === "object" && firstCall.args !== null
? (firstCall.args as Record<string, unknown>)
: undefined;
setPhase("tool_calling", toolName, toolArgs);
} else if (
(chunkType === "ai" || chunkType === "aimessagechunk" || chunkType === "aimessage") &&
((typeof content === "string" && content.length > 0) || (Array.isArray(content) && content.length > 0))
) {
setPhase("streaming");
}
}
}
const delta = extractAiTextDelta(parsed);
if (delta) req.onTextDelta?.(delta.text, delta.id);
}
}
}
if (awaitingUserClarification && result.clarifications.length > 0) {
result.status = "clarification";
result.summary = null;
result.clarification = result.clarifications[0] ?? null;
}
return result;
}