364 lines
14 KiB
TypeScript
364 lines
14 KiB
TypeScript
import { setTimeout as delay } from 'node:timers/promises';
|
|
import type { PublicUsage } from '../../shared/coding-conversation-contracts';
|
|
import type { TeacherTopic } from '../../shared/coding-teacher';
|
|
import { proxyAwareFetch } from '../utils/proxy-fetch';
|
|
import {
|
|
assertTeacherAccount,
|
|
teacherCloudRequest,
|
|
TeacherError,
|
|
type TeacherAccount,
|
|
} from './config-client';
|
|
import type { TeacherModelMessage } from './context';
|
|
import { createTeacherReadTools, type TeacherReadAccess } from './read-tools';
|
|
|
|
interface TeacherSession {
|
|
access_token: string;
|
|
expires_at: number;
|
|
api_base_url: string;
|
|
scope: string;
|
|
}
|
|
type Json = Record<string, unknown>;
|
|
function object(value: unknown): Json {
|
|
if (!value || typeof value !== 'object' || Array.isArray(value))
|
|
throw new TeacherError(502, 'teacher_protocol_invalid', '老师服务返回的数据无效。');
|
|
return value as Json;
|
|
}
|
|
function identifier(value: unknown): string {
|
|
if (typeof value !== 'string' || !value)
|
|
throw new TeacherError(502, 'teacher_protocol_invalid', '老师服务未返回有效的运行标识。');
|
|
return value;
|
|
}
|
|
export interface TeacherCloudTransport {
|
|
json(path: string, body?: unknown, signal?: AbortSignal): Promise<Json>;
|
|
events(
|
|
path: string,
|
|
signal: AbortSignal,
|
|
accept: (event: string, data: Json, id: string) => void
|
|
): Promise<void>;
|
|
}
|
|
|
|
export function teacherCloudTransport(account: TeacherAccount): TeacherCloudTransport {
|
|
let session: TeacherSession | undefined;
|
|
const fetchCloud = async (path: string, body?: unknown, signal?: AbortSignal) => {
|
|
// 退出账号后只允许用已经持有的短凭据停止原问题，不能再读状态或发送内容。
|
|
const stopping = /^\/questions\/[^/]+\/cancel$/.test(path) && body !== undefined;
|
|
for (let attempt = 0; attempt < 3; attempt++) {
|
|
if (!stopping || !session) assertTeacherAccount(account);
|
|
signal?.throwIfAborted();
|
|
if (!session || (!stopping && session.expires_at * 1000 < Date.now() + 10000)) {
|
|
session = await teacherCloudRequest<TeacherSession>(
|
|
account,
|
|
'/api/coding-teacher/session',
|
|
'POST'
|
|
);
|
|
if (
|
|
session.scope !== 'makelore-teachers' ||
|
|
typeof session.access_token !== 'string' ||
|
|
!session.access_token ||
|
|
!Number.isFinite(session.expires_at) ||
|
|
session.expires_at * 1000 <= Date.now() ||
|
|
typeof session.api_base_url !== 'string' ||
|
|
!/^https?:\/\//.test(session.api_base_url)
|
|
) {
|
|
throw new TeacherError(502, 'teacher_session_invalid', '老师接入凭据无效，请重新连接。');
|
|
}
|
|
}
|
|
if (!stopping) assertTeacherAccount(account);
|
|
try {
|
|
const response = await proxyAwareFetch(
|
|
session.api_base_url.replace(/\/+$/, '') + '/api/makelore/teachers' + path,
|
|
{
|
|
method: body === undefined ? 'GET' : 'POST',
|
|
headers: {
|
|
Authorization: 'Bearer ' + session.access_token,
|
|
'Content-Type': 'application/json',
|
|
},
|
|
body: body === undefined ? undefined : JSON.stringify(body),
|
|
signal: AbortSignal.any([...(signal ? [signal] : []), AbortSignal.timeout(45000)]),
|
|
}
|
|
);
|
|
if (!stopping) assertTeacherAccount(account);
|
|
if (response.status === 401 && attempt < 2) {
|
|
await response.body?.cancel();
|
|
session = undefined;
|
|
continue;
|
|
}
|
|
if (response.status >= 500 && attempt < 2) {
|
|
await response.body?.cancel();
|
|
await delay(500, undefined, { signal });
|
|
continue;
|
|
}
|
|
if (!response.ok) {
|
|
const data = object(await response.json());
|
|
const detail = data.detail;
|
|
const message =
|
|
typeof detail === 'string'
|
|
? detail
|
|
: detail && typeof detail === 'object'
|
|
? object(detail).message
|
|
: undefined;
|
|
throw new TeacherError(
|
|
response.status,
|
|
'teacher_cloud_failed',
|
|
typeof message === 'string' ? message : '老师服务暂不可用，请稍后重试。'
|
|
);
|
|
}
|
|
return response;
|
|
} catch (error) {
|
|
signal?.throwIfAborted();
|
|
if (error instanceof TeacherError || attempt === 2) throw error;
|
|
await delay(500, undefined, { signal });
|
|
}
|
|
}
|
|
throw new TeacherError(502, 'teacher_connection_failed', '连接老师失败。');
|
|
};
|
|
return {
|
|
async json(path, body, signal) {
|
|
const response = await fetchCloud(path, body, signal);
|
|
const data = object(await response.json());
|
|
if (!path.endsWith('/cancel')) assertTeacherAccount(account);
|
|
return data;
|
|
},
|
|
async events(path, signal, accept) {
|
|
const response = await fetchCloud(path, undefined, signal);
|
|
if (!response.body)
|
|
throw new TeacherError(502, 'teacher_stream_missing', '老师回复连接不可用。');
|
|
const reader = response.body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = '';
|
|
try {
|
|
while (true) {
|
|
const next = await reader.read();
|
|
if (next.done) break;
|
|
assertTeacherAccount(account);
|
|
signal.throwIfAborted();
|
|
buffer += decoder.decode(next.value, { stream: true }).replaceAll('\r', '');
|
|
let end: number;
|
|
while ((end = buffer.indexOf('\n\n')) >= 0) {
|
|
const block = buffer.slice(0, end);
|
|
buffer = buffer.slice(end + 2);
|
|
let event = 'message',
|
|
id = '';
|
|
const data: string[] = [];
|
|
for (const line of block.split('\n')) {
|
|
if (line.startsWith('event:')) event = line.slice(6).trim();
|
|
else if (line.startsWith('id:')) id = line.slice(3).trim();
|
|
else if (line.startsWith('data:')) data.push(line.slice(5).trimStart());
|
|
}
|
|
if (data.length) accept(event, object(JSON.parse(data.join('\n'))), id);
|
|
}
|
|
}
|
|
} finally {
|
|
await reader.cancel().catch(() => undefined);
|
|
reader.releaseLock();
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
export function prepareCloudTeacher(
|
|
account: TeacherAccount,
|
|
topic: TeacherTopic,
|
|
requestId: string,
|
|
access: TeacherReadAccess,
|
|
onProgress: (text: string) => void,
|
|
saveRequestId: (id: string) => Promise<void>,
|
|
transport: TeacherCloudTransport = teacherCloudTransport(account)
|
|
) {
|
|
const tools = createTeacherReadTools(access);
|
|
return {
|
|
inputLimit: topic.definition.limits.max_input_tokens,
|
|
async run(
|
|
messages: TeacherModelMessage[],
|
|
signal: AbortSignal,
|
|
onText: (delta: string) => void
|
|
): Promise<PublicUsage | undefined> {
|
|
const localContext = {
|
|
id: requestId,
|
|
scope: { project_id: topic.projectId, source_session_id: topic.sourceConversationId },
|
|
tools: tools.definitions.map((item) => item.function.name),
|
|
};
|
|
const deadline = Date.now() + 60 * 60 * 1000;
|
|
const bounded = AbortSignal.any([signal, AbortSignal.timeout(60 * 60 * 1000)]);
|
|
let questionId: string | undefined;
|
|
let completed = false;
|
|
try {
|
|
// 重启后本地快照已经丢失，先停止旧请求，再以新的问题建立读取作用域。
|
|
const previous = topic.requests.filter((item) => item.id !== requestId).at(-1);
|
|
if (previous && previous.status !== 'completed') {
|
|
try {
|
|
await transport.json(
|
|
'/questions/' + encodeURIComponent(previous.id) + '/cancel',
|
|
{},
|
|
bounded
|
|
);
|
|
} catch (error) {
|
|
if (!(error instanceof TeacherError && error.status === 404)) throw error;
|
|
}
|
|
}
|
|
let queued = await transport.json(
|
|
'/questions',
|
|
{
|
|
teacher_version: topic.version,
|
|
thread_id: topic.id,
|
|
request_id: requestId,
|
|
query: messages
|
|
.filter((item) => item.role === 'user')
|
|
.map((item) => item.content)
|
|
.join('\n\n'),
|
|
local_context: localContext,
|
|
},
|
|
bounded
|
|
);
|
|
questionId = identifier(queued.request_id);
|
|
await saveRequestId(questionId);
|
|
while (!queued.run_id) {
|
|
if (!['queued', 'pending'].includes(String(queued.status)))
|
|
throw new TeacherError(
|
|
409,
|
|
'teacher_request_stopped',
|
|
'老师提问未能启动，请重新提问。'
|
|
);
|
|
onProgress('正在等待老师…');
|
|
await delay(800, undefined, { signal: bounded });
|
|
queued = await transport.json(
|
|
'/requests/' + encodeURIComponent(questionId),
|
|
undefined,
|
|
bounded
|
|
);
|
|
}
|
|
let runId = identifier(queued.run_id),
|
|
cursor = '0-0',
|
|
rounds = 0;
|
|
let runText = '',
|
|
messageId = '';
|
|
while (Date.now() < deadline) {
|
|
bounded.throwIfAborted();
|
|
access.assertCurrent();
|
|
const view = await transport.json(
|
|
'/runs/' + encodeURIComponent(runId),
|
|
undefined,
|
|
bounded
|
|
);
|
|
if (view.continued_run_id) {
|
|
runId = identifier(view.continued_run_id);
|
|
cursor = '0-0';
|
|
runText = '';
|
|
messageId = '';
|
|
continue;
|
|
}
|
|
if (view.status === 'interrupted') {
|
|
const pending = object(view.interrupt);
|
|
if (pending.source !== 'client_read_tools')
|
|
throw new TeacherError(
|
|
409,
|
|
'teacher_interaction_unsupported',
|
|
'老师请求了当前面板不支持的交互，请联系运营调整该智能体。'
|
|
);
|
|
if (
|
|
pending.context_id !== requestId ||
|
|
!Array.isArray(pending.calls) ||
|
|
!pending.calls.length ||
|
|
pending.calls.length > 32 ||
|
|
++rounds > 6
|
|
)
|
|
throw new TeacherError(
|
|
409,
|
|
'teacher_context_expired',
|
|
'老师读取请求已失效，请重新提问。'
|
|
);
|
|
onProgress('正在读取项目与会话…');
|
|
const results = [];
|
|
for (const raw of pending.calls) {
|
|
const call = object(raw);
|
|
if (typeof call.tool_call_id !== 'string' || typeof call.name !== 'string')
|
|
throw new TeacherError(502, 'teacher_protocol_invalid', '老师读取请求无效。');
|
|
results.push({
|
|
tool_call_id: call.tool_call_id,
|
|
...(await tools.executeResult(call.name, JSON.stringify(call.arguments), bounded)),
|
|
});
|
|
}
|
|
// POST 重试使用完全相同的结果，文件变化也不会导致重复续接或不同输入。
|
|
const resumed = await transport.json(
|
|
'/runs/' + encodeURIComponent(runId) + '/tool-results',
|
|
{ context_id: requestId, results },
|
|
bounded
|
|
);
|
|
runId = identifier(resumed.run_id);
|
|
cursor = '0-0';
|
|
runText = '';
|
|
messageId = '';
|
|
onProgress('老师正在继续思考…');
|
|
continue;
|
|
}
|
|
if (view.status === 'completed') {
|
|
const output = typeof view.output === 'string' ? view.output : '';
|
|
if (output.startsWith(runText)) onText(output.slice(runText.length));
|
|
else if (output) onText('\n\n' + output);
|
|
completed = true;
|
|
onProgress('');
|
|
// Yuxi 的账本记录每个模型调用；不把线程累计 token 当作本问题费用。
|
|
return undefined;
|
|
}
|
|
if (['failed', 'cancelled'].includes(String(view.status))) {
|
|
const detail = view.error ? object(view.error).message : undefined;
|
|
throw new TeacherError(
|
|
409,
|
|
'teacher_run_failed',
|
|
typeof detail === 'string' ? detail : '老师回复已停止。'
|
|
);
|
|
}
|
|
onProgress('老师正在思考…');
|
|
try {
|
|
await transport.events(
|
|
'/runs/' +
|
|
encodeURIComponent(runId) +
|
|
'/events?after_seq=' +
|
|
encodeURIComponent(cursor),
|
|
bounded,
|
|
(_event, envelope, id) => {
|
|
if (id) cursor = id;
|
|
const payload = envelope.payload ? object(envelope.payload) : {};
|
|
for (const item of Array.isArray(payload.items)
|
|
? payload.items
|
|
: payload.chunk
|
|
? [payload.chunk]
|
|
: []) {
|
|
const chunk = object(item);
|
|
const event = chunk.stream_event ? object(chunk.stream_event) : {};
|
|
if (event.type === 'message_delta' && typeof event.content === 'string') {
|
|
if (typeof event.message_id === 'string' && event.message_id !== messageId) {
|
|
if (messageId && runText) onText('\n\n');
|
|
messageId = event.message_id;
|
|
runText = '';
|
|
}
|
|
runText += event.content;
|
|
onText(event.content);
|
|
}
|
|
}
|
|
}
|
|
);
|
|
} catch (error) {
|
|
bounded.throwIfAborted();
|
|
access.assertCurrent();
|
|
if (error instanceof TeacherError && error.status < 500) throw error;
|
|
onProgress('连接中断，正在恢复老师回复…');
|
|
}
|
|
await delay(400, undefined, { signal: bounded });
|
|
}
|
|
throw new TeacherError(408, 'teacher_question_expired', '本次老师提问已超时，请重新提问。');
|
|
} finally {
|
|
if (!completed) {
|
|
await transport
|
|
.json(
|
|
'/questions/' + encodeURIComponent(requestId) + '/cancel',
|
|
{},
|
|
AbortSignal.timeout(10000)
|
|
)
|
|
.catch(() => undefined);
|
|
}
|
|
}
|
|
},
|
|
};
|
|
}
|