AI SDK v7 是运行于 TypeScript 与 Next.js 环境的标准化大模型开发工具包，通过统一模型接口、双层消息转换协议与流式数据通道屏蔽底层模型 API 差异，支持结构化输出、多步工具编排与前后端单消息增量同步。
AI SDK 将对话消息划分为面向界面与持久化的 UIMessage 和面向模型推理的 ModelMessage 两套结构，两者的职责边界与字段形态如下：
| 消息类型 | 职责定位 | 核心字段结构 | 典型消费端 |
|---|---|---|---|
UIMessage |
界面渲染与数据库持久化 | 包含 id、role 与多态 parts 数组（文本、工具调用、自定义数据块） |
React useChat、消息历史表 |
ModelMessage |
模型推理协议输入 | 去除消息 id，包含 role（user / assistant / tool）与 content 数组 |
generateText、streamText |
在服务端接收前端请求或从数据库加载历史 UIMessage 后，需调用异步函数 convertToModelMessages 将其转换为 ModelMessage 序列，再交由模型推理函数消费。
AI SDK Core 提供两个核心生成原语，并通过 output 选项统一承载结构化数据抽取：
generateText：非流式生成函数，阻塞等待模型完成完整推理并返回文本、工具调用结果或结构化对象。适用于后台异步任务（如会话标题提取、滚动摘要生成）与离线评测裁判。streamText：流式生成函数，支持多步工具调用循环（stopWhen）、步间动态参数干预（prepareStep）以及面向前端的实时增量数据推送。Output.object：在 generateText 或 streamText 中挂载 Zod Schema，约束模型按指定结构生成并校验强类型对象，替代早期版本中独立的 generateObject 与 streamObject 接口。当在 streamText 中挂载 tools 时，单次请求可跨越多个推理步骤（Step）。每个步骤由一次模型推理及其触发的工具执行构成。运行期通过两类钩子控制循环边界：
stopWhen：定义循环终止条件数组（如达到最大步数 isStepCount(8) 或捕获到管理端写工具发出的审批请求），任一条件满足即停止后续模型调用。prepareStep：在每个推理步骤发起前执行，接收已完成的 steps、stepNumber 与累积的 responseMessages，可按步动态修改当前步生效的 instructions、可用工具列表 activeTools 与工具选择策略 toolChoice。在实际工程中，不同任务对推理能力、响应延迟与 Token 成本的要求存在差异。将模型按职责划分为主对话模型、辅助轻量模型、向量模型与语义重排模型，并通过统一的 Provider 工厂管理端点配置。
定义运行时模型常量与阈值配置：
// 主对话默认模型
export const DEFAULT_CHAT_MODEL = 'qwen3.5-plus';
// 辅助与评测任务轻量模型（用于会话标题生成、压缩滚动摘要生成、评测裁判）
export const DEFAULT_AUXILIARY_MODEL = 'qwen3.5-flash';
// 原生向量模型与维度
export const DEFAULT_EMBEDDING_MODEL = 'text-embedding-v4';
export const EMBEDDING_DIMENSION = 1024;
// 默认语义重排模型
export const DEFAULT_RERANK_MODEL = 'qwen3-rerank';
// 国内 DashScope 兼容模式与原生 API 默认端点
export const DASHSCOPE_DEFAULT_BASE_URL =
'https://dashscope.aliyuncs.com/compatible-mode/v1';
export const DASHSCOPE_DEFAULT_EMBEDDING_BASE_URL =
'https://dashscope.aliyuncs.com/api/v1';
// 主对话单步最大输出 Token 与单轮最大步数
export const CHAT_MAX_OUTPUT_TOKENS = 16384;
export const MAX_AGENT_STEPS = 8;使用 @ai-sdk/alibaba 封装模型实例化工厂：
import { createAlibaba } from '@ai-sdk/alibaba';
import {
DEFAULT_CHAT_MODEL,
DEFAULT_AUXILIARY_MODEL,
DEFAULT_EMBEDDING_MODEL,
DASHSCOPE_DEFAULT_BASE_URL,
DASHSCOPE_DEFAULT_EMBEDDING_BASE_URL,
} from './constants';
export interface DashScopeProviderOptions {
apiKey?: string;
baseURL?: string;
embeddingBaseURL?: string;
}
// 创建配置了兼容端点与原生向量端点的 Provider 实例
export function createDashScopeProvider(options?: DashScopeProviderOptions) {
const apiKey = options?.apiKey ?? process.env.DASHSCOPE_API_KEY;
const baseURL =
options?.baseURL ??
process.env.DASHSCOPE_BASE_URL ??
DASHSCOPE_DEFAULT_BASE_URL;
const embeddingBaseURL =
options?.embeddingBaseURL ??
process.env.DASHSCOPE_EMBEDDING_BASE_URL ??
DASHSCOPE_DEFAULT_EMBEDDING_BASE_URL;
return createAlibaba({
apiKey,
baseURL,
embeddingBaseURL,
});
}
// 获取主对话模型实例（默认 qwen3.5-plus）
export function getChatModel(provider = createDashScopeProvider()) {
const modelName = process.env.BLOG_AGENT_MODEL || DEFAULT_CHAT_MODEL;
return provider(modelName);
}
// 获取辅助任务模型实例（默认 qwen3.5-flash）
export function getAuxiliaryModel(provider = createDashScopeProvider()) {
const modelName =
process.env.BLOG_AGENT_AUXILIARY_MODEL || DEFAULT_AUXILIARY_MODEL;
return provider(modelName);
}
// 获取原生向量模型实例（默认 text-embedding-v4）
export function getEmbeddingModel(provider = createDashScopeProvider()) {
const modelName =
process.env.BLOG_AGENT_EMBEDDING_MODEL || DEFAULT_EMBEDDING_MODEL;
return provider.embeddingModel(modelName);
}分层路由的设计考量如下：
qwen3.5-plus）：承担多步工具调度、复杂技术问答与长文创作，要求较强的指令遵循与长上下文推理能力。qwen3.5-flash）：承担首轮会话标题提取、历史消息滚动摘要与自动化评测裁判。此类任务输入输出边界固定，使用轻量模型可降低推理延迟与调用成本。当新会话完成首轮问答后，后台任务调用 generateText 提取 15 至 20 字的简短标题，并通过 providerOptions 显式关闭思考模式以降低响应延迟。
实现首轮会话标题生成：
import { asc, eq } from 'drizzle-orm';
import { generateText } from 'ai';
import { db } from '@/services/db/db';
import { agentMessageTable, agentSessionTable } from '@/services/db/schema';
import { extractTextMessage } from '@/validators/blog-agent';
import { getAuxiliaryModel } from './provider';
import { buildTitlePrompt } from './prompt';
export async function generateAndSaveFirstTurnTitle(
sessionId: string,
dbInstance: any = db,
): Promise<void> {
const messages = await dbInstance
.select()
.from(agentMessageTable)
.where(eq(agentMessageTable.session_id, sessionId))
.orderBy(asc(agentMessageTable.seq))
.limit(2);
if (messages.length < 2) return;
const userText = extractTextMessage(
messages[0].parts as Array<Record<string, unknown>>,
).slice(0, 200);
const assistantText = extractTextMessage(
messages[1].parts as Array<Record<string, unknown>>,
).slice(0, 200);
const model = getAuxiliaryModel();
const { text } = await generateText({
model,
providerOptions: {
alibaba: {
enableThinking: false,
},
},
prompt: buildTitlePrompt(userText, assistantText),
});
const cleanedTitle = text
.trim()
.replace(/^["'《]+|["'》]+$/g, '')
.replace(/[。！？\n\r]+$/g, '')
.trim();
if (cleanedTitle) {
await dbInstance
.update(agentSessionTable)
.set({ title: cleanedTitle.slice(0, 100), update_time: new Date() })
.where(eq(agentSessionTable.id, sessionId));
}
}在需要模型返回严格符合类型定义的数据结构（如评测打分、元数据抽取）时，向 generateText 传入 Output.object({ schema })。
使用 Output.object 生成结构化评估结果：
import { generateText, Output, type LanguageModel } from 'ai';
import { z } from 'zod';
const summaryFidelitySchema = z.object({
feedback: z
.string()
.describe('逐条核查 keyFacts 是否保留、是否存在原文未出现的编造事实的简要分析'),
score: z
.enum(['A', 'B', 'C', 'D'])
.describe(
'A: 关键事实全部保留且无幻觉 (1.0); B: 保留绝大部分事实，仅遗漏次要细节 (0.7); C: 遗漏重要事实 (0.3); D: 严重遗漏或包含编造事实 (0.0)',
),
});
export async function evaluateSummaryWithStructuredOutput(
judgeModel: LanguageModel,
prompt: string,
) {
const result = await generateText({
model: judgeModel,
prompt,
output: Output.object({
schema: summaryFidelitySchema,
}),
providerOptions: {
alibaba: {
enableThinking: false,
},
},
});
// result.output 自动具备 { feedback: string; score: 'A' | 'B' | 'C' | 'D' } 类型
return result.output;
}关键特性说明：
.describe() 会被转换为 JSON Schema 的字段说明注入模型上下文，用于约束每个字段的语义粒度与枚举分档规则。feedback 字段定义在 score 之前，促使模型在自回归生成时先输出分析依据，再落定最终档位。AI SDK 的 tool 函数将工具的输入校验、业务执行与模型可见输出解耦。在构建对话应用时，前端 UI 渲染所需的结构化数据（如引用卡片列表、总条数）与模型推理所需的紧凑文本往往不同，可通过 toModelOutput 实现双视图分离。
定义具备 execute 与 toModelOutput 的只读工具：
import { tool } from 'ai';
import { z } from 'zod';
import {
BLOG_SITE_URL,
GET_BLOG_OVERVIEW_DEFAULT_LIMIT,
GET_BLOG_OVERVIEW_MAX_LIMIT,
} from './constants';
import {
getPublicPostCount,
getPublicCollectionCount,
getRecentPublicPosts,
getDraftPostCount,
} from '@/services/posts';
import { formatShanghaiDate } from './time';
export const getBlogOverviewInputSchema = z.object({
recentLimit: z
.number()
.int()
.min(1)
.max(GET_BLOG_OVERVIEW_MAX_LIMIT)
.default(GET_BLOG_OVERVIEW_DEFAULT_LIMIT)
.optional()
.describe('返回最近发布的文章数量（1-20，默认 5）'),
});
export function createGetBlogOverviewTool(isAdmin: boolean) {
return tool({
description:
'获取博客整体概况，包括文章总数、草稿统计、专栏总数及最近发布的文章列表。',
inputSchema: getBlogOverviewInputSchema,
execute: async ({ recentLimit = GET_BLOG_OVERVIEW_DEFAULT_LIMIT }) => {
const [totalPosts, totalCollections, recentRawPosts, draftPosts] =
await Promise.all([
getPublicPostCount(),
getPublicCollectionCount(),
getRecentPublicPosts(recentLimit),
isAdmin ? getDraftPostCount() : Promise.resolve(undefined),
]);
const recentPosts = recentRawPosts.map((post) => ({
id: post.id,
title: post.title,
publishDate: formatShanghaiDate(post.publish_time),
url: `${BLOG_SITE_URL}/post/${post.id}`,
}));
// 返回结构化对象，完整保存在 UIMessage 的 tool part output 中供前端组件消费
return {
totalPosts,
totalCollections,
recentPosts,
...(isAdmin && draftPosts !== undefined ? { draftPosts } : {}),
};
},
toModelOutput: ({ output }) => {
// 将结构化输出投影为紧凑纯文本交由模型消费，减少 JSON 语法符号占用的 Token
const listText =
output.recentPosts.length > 0
? output.recentPosts
.map((p) => `- [${p.title}](${p.url}) (${p.publishDate})`)
.join('\n')
: '暂无已发布的文章';
const draftLine =
output.draftPosts !== undefined ? `草稿数: ${output.draftPosts}\n` : '';
const value = `文章总数: ${output.totalPosts}\n${draftLine}专栏总数: ${output.totalCollections}\n\n最近的 ${output.recentPosts.length} 篇文章:\n${listText}`;
return { type: 'text', value };
},
});
}核心属性机制如下：
inputSchema：接收 Zod Schema，在发送给模型时序列化为 JSON Schema，并在模型发起调用时对入参执行运行时校验。execute：运行在服务端进程中的异步函数，其返回值会作为 tool-output-available 数据块通过流推送到前端，并持久化到数据库。toModelOutput：定义工具执行结果在多步推理循环中如何转换为 ModelMessage 的 tool-result 内容。例如在 searchPosts 工具中，execute 同时返回 formattedContext 与供前端展示引用胶囊的 citations 数组，而 toModelOutput 仅提取 formattedContext 纯文本送入模型，避免重复注入冗余 JSON 结构。在 streamText 中结合 stopWhen 与 prepareStep，可以在保持模型自主规划能力的同时，强制实施步数上限、单工具配额与 Token 预算控制。
封装单轮对话执行函数 runAgentTurn：
import {
isStepCount,
streamText,
type LanguageModel,
type StopCondition,
type UIMessage,
} from 'ai';
import type { UserRole } from '@/services/db/schema';
import { CHAT_MAX_OUTPUT_TOKENS, MAX_AGENT_STEPS } from './constants';
import { assembleModelTurnMessages } from './compaction';
import { evaluateStepGuardrails } from './guardrails';
import { ADMIN_WRITE_TOOL_NAMES } from './hitl';
import { buildSystemPrompt } from './prompt';
import { createBlogAgentTools } from './tools';
// 检查当前轮次最后一个完成步骤中是否有管理写工具成功发出了审批请求
export function hasAdminApprovalRequestEmitted(): StopCondition<any, any> {
return ({ steps }) => {
const lastStep = steps[steps.length - 1];
if (!lastStep?.toolResults) return false;
return lastStep.toolResults.some((tr: any) => {
if (!ADMIN_WRITE_TOOL_NAMES.includes(tr.toolName)) return false;
const output = tr.output;
return (
output && typeof output === 'object' && (output as any).success === true
);
});
};
}
export async function runAgentTurn(params: {
role: UserRole;
userId: number;
uiMessages: UIMessage[];
summary?: string;
model: LanguageModel;
abortSignal?: AbortSignal;
writer?: any;
}) {
const { role, userId, uiMessages, summary, model, abortSignal, writer } =
params;
const tools = createBlogAgentTools({ role, userId, writer });
const availableToolNames = Object.keys(tools);
const boundedMessages = await assembleModelTurnMessages({
uiMessages,
tools,
summary,
});
const initialInstructions = buildSystemPrompt({
role,
conversationSummary: summary,
});
const guardrailState = {
turnToolOutputTokens: 0,
forcedStopAtMaxSteps: false,
budgetExhausted: false,
};
const streamResult = streamText({
model,
instructions: initialInstructions,
messages: boundedMessages,
tools,
stopWhen: [
isStepCount(MAX_AGENT_STEPS),
hasAdminApprovalRequestEmitted(),
],
maxOutputTokens: CHAT_MAX_OUTPUT_TOKENS,
abortSignal,
providerOptions: {
alibaba: {
enableThinking: false,
},
},
prepareStep: async ({ steps, stepNumber, responseMessages }) => {
const guardResult = evaluateStepGuardrails({
stepNumber,
steps,
initialInstructions,
responseMessages,
availableToolNames,
});
guardrailState.turnToolOutputTokens = guardResult.turnToolOutputTokens;
guardrailState.forcedStopAtMaxSteps = guardResult.forcedStopAtMaxSteps;
guardrailState.budgetExhausted = guardResult.budgetExhausted;
return {
toolChoice: guardResult.toolChoice,
activeTools: guardResult.activeTools as any,
instructions: guardResult.instructions,
};
},
});
return Object.assign(streamResult, { guardrailState });
}prepareStep 调用的 evaluateStepGuardrails 根据当前步号、已调用工具频次以及累积工具输出 Token 数，将推理状态划分为三个等级：
实现三级步内守卫纯函数：
import {
MAX_AGENT_STEPS,
MAX_GET_POST_CALLS_PER_TURN,
MAX_SEARCH_POSTS_CALLS_PER_TURN,
MAX_TURN_TOOL_OUTPUT_TOKENS,
} from './constants';
export function evaluateStepGuardrails<T extends string>(params: {
stepNumber: number;
steps?: Array<Record<string, unknown>>;
initialInstructions: string;
responseMessages: Array<Record<string, unknown>>;
availableToolNames: T[];
}) {
const {
stepNumber,
steps = [],
initialInstructions,
responseMessages = [],
availableToolNames,
} = params;
const turnToolOutputTokens =
estimateToolOutputsInResponseMessages(responseMessages);
const forcedStopAtMaxSteps = stepNumber >= MAX_AGENT_STEPS - 1;
const budgetExhausted = turnToolOutputTokens >= MAX_TURN_TOOL_OUTPUT_TOKENS;
// 第一级：达到最大步数（第 8 步）或累积工具输出达到 18,000 Token 预算上限，强制收尾
if (forcedStopAtMaxSteps || budgetExhausted) {
return {
toolChoice: 'none' as const,
activeTools: [] as T[],
instructions: `${initialInstructions}\n\n<system-guardrail>\n[系统提示：本轮工具调用步数或工具输出上下文已达上限，请勿再调用工具，直接基于目前已获取的工具结果向用户输出完整、准确的总结回答；若现有信息仍不足以完整回答，请如实说明已查到的部分。]\n</system-guardrail>`,
forcedStopAtMaxSteps,
budgetExhausted,
turnToolOutputTokens,
};
}
// 第二级：检查单工具调用配额（searchPosts 最多 2 次，getPost 最多 3 次）
const toolCounts = countToolCalls(steps, responseMessages);
const isSearchPostsLimitReached =
(toolCounts.searchPosts || 0) >= MAX_SEARCH_POSTS_CALLS_PER_TURN;
const isGetPostLimitReached =
(toolCounts.getPost || 0) >= MAX_GET_POST_CALLS_PER_TURN;
if (isSearchPostsLimitReached || isGetPostLimitReached) {
const activeTools = availableToolNames.filter((toolName) => {
if (toolName === 'searchPosts' && isSearchPostsLimitReached) return false;
if (toolName === 'getPost' && isGetPostLimitReached) return false;
return true;
});
const limitTip =
isSearchPostsLimitReached && isGetPostLimitReached
? '[系统提示：本轮检索文章（searchPosts）与读取文章全文（getPost）均已达本轮调用上限，请基于已获取的内容作答，或仅使用其他工具。]'
: isGetPostLimitReached
? '[系统提示：本轮读取文章全文（getPost）已达 3 次上限，请基于已获取的内容作答，或仅使用其他工具。]'
: '[系统提示：本轮检索文章（searchPosts）已达 2 次上限，请基于已获取的内容作答，或仅使用其他工具。]';
return {
toolChoice: 'auto' as const,
activeTools,
instructions: `${initialInstructions}\n\n<system-guardrail>\n${limitTip}\n</system-guardrail>`,
forcedStopAtMaxSteps: false,
budgetExhausted: false,
turnToolOutputTokens,
};
}
// 第三级：正常状态，由模型自主判定是否调用工具
return {
toolChoice: 'auto' as const,
activeTools: undefined,
instructions: initialInstructions,
forcedStopAtMaxSteps: false,
budgetExhausted: false,
turnToolOutputTokens,
};
}步间干预机制解决的核心问题如下：
stopWhen: isStepCount(8)，当模型在第 8 步仍选择调用工具时，循环会直接终止，导致前端收到空文本回复。通过在 stepNumber === 7（0 索引的第 8 步）将 toolChoice 设为 'none' 并清空 activeTools，可强制模型在最后一步基于已有材料生成文本总结。searchPosts 达到 2 次或 getPost 达到 3 次配额时，将其从 activeTools 中剔除并追加 <system-guardrail> 尾部提示，防止模型在同一查询上陷入重复翻页或无限检索循环。系统提示词（instructions）采用 kebab-case XML 标签划分语义边界，并按静态程度从前到后排列以利用模型服务商的 Prompt Caching 机制：
<task-context>、<rules>、<tool-guidelines>、<citation-rules>。配合工具 Schema 定义共同构成超过 1,024 Token 的静态前缀，跨用户、跨会话保持一致。<time-context>、管理员专属的 <user-memories> 以及会话内随水位推进更新的 <conversation-summary>。<the-ask> 位于提示词末尾，在步内守卫触发时紧随其后追加 <system-guardrail>，利用模型对输入序列末端的注意力权重强化当前步约束。当将用户历史输入、长期记忆或会话摘要嵌入 XML 容器时，若文本中恰好包含 </conversation-summary> 或 <system-guardrail> 等同名标签，会导致 XML 容器提前闭合与指令穿越。
实现仅针对保留标签的正则转义函数：
const PROMPT_RESERVED_XML_TAGS = [
'task-context',
'rules',
'tool-guidelines',
'citation-rules',
'time-context',
'user-memories',
'memory',
'conversation-summary',
'the-ask',
'system-guardrail',
'examples',
'example',
'input',
'expected',
'conversation-history',
'user',
'assistant',
'output-format',
'background-data',
'previous-summary',
'new-messages',
'required-key-facts',
'candidate-summary',
] as const;
const RESERVED_XML_TAG_REGEX = new RegExp(
`<(\\/?(?:${PROMPT_RESERVED_XML_TAGS.join('|')})\\b[^>]*)>`,
'gi',
);
// 仅转义与系统提示词结构同名的 XML 开闭标签，保留代码片段中的泛型（如 Promise<void>）与比较运算符
export function sanitizePromptXmlTags(text: string): string {
if (!text) return '';
return text.replace(RESERVED_XML_TAG_REGEX, '<$1>');
}在服务端路由中，使用 createUIMessageStream 可以将自定义数据块（如审批执行结果 data-approval-result）与 streamText 产出的模型流合并到同一条 HTTP SSE 响应流中，并在 onEnd 回调中统一完成链接净化与数据库落库。
当模型单步输出触及 maxOutputTokens 上限（finishReason === 'length'）时，可通过 TransformStream 拦截流尾的 finish 块，向客户端注入显式的截断提示文本块：
import {
createUIMessageStream,
createUIMessageStreamResponse,
toUIMessageStream,
} from 'ai';
export const LENGTH_TRUNCATION_NOTICE_TEXT =
'\n\n> [系统提示：本次生成已达单步输出长度上限，内容或工具调用未能完整闭合。请发送“继续”或指示助手使用分段追加（appendContent）方式补齐剩余内容。]';
// 包装 UI 消息流，在因长度超限结束前追加截断提示文本块
export function withLengthTruncationNotice<T extends Record<string, any>>(
sourceStream: ReadableStream<T>,
): ReadableStream<T> {
return sourceStream.pipeThrough(
new TransformStream<T, T>({
transform(chunk, controller) {
if (
chunk &&
chunk.type === 'finish' &&
chunk.finishReason === 'length'
) {
const tipId = `truncation-${crypto.randomUUID()}`;
controller.enqueue({
type: 'text-start',
id: tipId,
} as unknown as T);
controller.enqueue({
type: 'text-delta',
id: tipId,
delta: LENGTH_TRUNCATION_NOTICE_TEXT,
} as unknown as T);
controller.enqueue({
type: 'text-end',
id: tipId,
} as unknown as T);
}
controller.enqueue(chunk);
},
}),
);
}在 createUIMessageStream 中组合自定义数据块与子流：
const uiStream = createUIMessageStream({
execute: async ({ writer }) => {
// 1. 写入前置自定义数据块
for (const res of approvalResults) {
writer.write({
type: 'data-approval-result',
data: res,
} as any);
}
// 2. 启动主模型多步推理流
const streamResult = await runAgentTurn({
role: user.role,
userId: user.id,
uiMessages,
summary: compactionSummary,
memories,
writer,
model: getChatModel(),
abortSignal: combinedSignal,
});
const childStream = toUIMessageStream({
stream: streamResult.stream,
sendReasoning: false,
generateMessageId: () => crypto.randomUUID(),
});
// 3. 合并包装了长度截断提示的模型流
writer.merge(withLengthTruncationNotice(childStream));
},
onEnd: async ({ responseMessage, isAborted, finishReason, isCancelled }) => {
// 流结束时净化未落地文章链接、持久化 assistant 消息并触发后台异步任务
await saveAssistantMessageOnEnd({
sessionId,
assistantSeq,
responseMessage,
});
},
});
return createUIMessageStreamResponse({
stream: decoupleUIMessageStream(uiStream, abortController, combinedSignal),
});@ai-sdk/react 的 useChat 默认在每次请求时将客户端内存中的全量 messages 数组发送至服务端。在服务端已持久化完整历史消息的架构下，可通过 DefaultChatTransport 的 prepareSendMessagesRequest 钩子重写请求体，仅上传当前轮次的用户消息、sessionId 与触发动作类型（submit-message 或 regenerate-message）。
配置客户端 DefaultChatTransport 与 useChat：
import { useMemo } from 'react';
import { useChat } from '@ai-sdk/react';
import { DefaultChatTransport } from 'ai';
export function useBlogAgentChat(sessionIdRef: React.RefObject<string>) {
const transport = useMemo(
() =>
new DefaultChatTransport({
api: '/api/tools/blog-agent/chat',
prepareSendMessagesRequest: ({
id,
messages: msgs,
trigger,
messageId,
}) => {
const userMessage =
(messageId
? msgs.find((m) => m.id === messageId && m.role === 'user')
: undefined) ||
[...msgs].reverse().find((m) => m.role === 'user') ||
msgs[msgs.length - 1];
const rawParts = userMessage?.parts;
const parts =
Array.isArray(rawParts) && rawParts.length > 0
? rawParts
: [{ type: 'text', text: (userMessage as any)?.content || '' }];
return {
body: {
sessionId: sessionIdRef.current,
trigger,
message: {
id: userMessage?.id || id || crypto.randomUUID(),
role: 'user',
parts,
},
...(messageId ? { messageId } : {}),
},
};
},
}),
[sessionIdRef],
);
return useChat({ transport });
}在 Web Streams 标准中，TransformStream 内部维护有限的缓冲队列。当客户端用户在流式生成中途刷新页面、关闭标签页，或反向代理提前断开连接且未触发底层 cancel 信号时，如果无人继续从 uiStream 读取剩余数据块，上游 TransformStream 会因背压（Backpressure）挂起，导致 createUIMessageStream 的 onEnd 回调与用户并发流锁释放逻辑无法执行。
实现后台主动拉取与双向取消联动的解耦包装器：
function decoupleUIMessageStream<T>(
uiStream: ReadableStream<T>,
abortController: AbortController,
abortSignal: AbortSignal,
): ReadableStream<T> {
const uiReader = uiStream.getReader();
let downstreamController: ReadableStreamDefaultController<T> | undefined;
let downstreamClosed = false;
const closeDownstream = () => {
if (downstreamController) {
try {
downstreamController.close();
} catch {
// 忽略已关闭流的重复关闭异常
}
}
};
const cancelUpstream = (reason?: unknown) => {
downstreamClosed = true;
if (!abortController.signal.aborted) {
abortController.abort(reason);
}
void uiReader.cancel(reason).catch(() => {});
closeDownstream();
};
const onSignalAbort = () => cancelUpstream(abortSignal.reason);
const responseStream = new ReadableStream<T>({
start(controller) {
downstreamController = controller;
if (downstreamClosed) closeDownstream();
},
cancel(reason) {
cancelUpstream(reason);
},
});
if (abortSignal.aborted) {
onSignalAbort();
} else {
abortSignal.addEventListener('abort', onSignalAbort, { once: true });
}
// 后台循环主动排空 uiReader，确保上游 onEnd 与 finally 锁释放始终触发
void (async () => {
try {
while (true) {
const { done, value } = await uiReader.read();
if (done) {
if (!downstreamClosed) {
downstreamClosed = true;
closeDownstream();
}
break;
}
if (!downstreamClosed && downstreamController) {
try {
downstreamController.enqueue(value);
} catch {
downstreamClosed = true;
}
}
}
} catch (error) {
if (!downstreamClosed && downstreamController) {
downstreamClosed = true;
try {
downstreamController.error(error);
} catch {
// 忽略下游错误写入异常
}
}
} finally {
abortSignal.removeEventListener('abort', onSignalAbort);
try {
uiReader.releaseLock();
} catch {
// 忽略释放锁异常
}
}
})();
return responseStream;
}大语言模型在生成带引用链接的回答时，可能会根据预训练记忆或猜测拼接出形如 https://heibaimeng.com/post/9999 的未验证链接。在流结束落库前，服务端从本轮工具输出、历史尾部消息、会话压缩摘要及审批执行结果中提取全部真实出现过的文章 ID 白名单，对生成的文本执行正则扫描，将不在白名单内的 Markdown 链接降级为纯文本标题。
实现未落地文章链接的校验与降级函数：
function markdownPostLinkPattern(): RegExp {
return /\[([^\]]+)\]\((?:https?:\/\/heibaimeng\.com)?(?:\/post\/|\/preview\/post\/|\/admin\/posts\/)(\d+)\)/g;
}
export function sanitizeUngroundedPostLinks(
text: string,
allowedPostIds: Set<number>,
): { sanitizedText: string; strippedLinkCount: number } {
if (!text) return { sanitizedText: text, strippedLinkCount: 0 };
let strippedLinkCount = 0;
const sanitizedText = text.replace(
markdownPostLinkPattern(),
(match, title, idStr) => {
const postId = Number(idStr);
if (allowedPostIds.has(postId)) {
return match;
}
strippedLinkCount++;
return title;
},
);
return { sanitizedText, strippedLinkCount };
}