fix: N1-N3 — crypto 顶层导入 + 移除冗余 AbortController + _def 重命名
- N1: require('crypto') 内联 → import * as crypto 顶层导入
- N2: 移除 Engine 内未连接的 AbortController(timeout 委托给 AiGatewayService)
- N3: writeUsageLog 的 _def 参数 → definition
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
cdcf056ba5
commit
aa1ce45d80
@ -1,4 +1,5 @@
|
|||||||
import { Injectable, Logger } from '@nestjs/common';
|
import { Injectable, Logger } from '@nestjs/common';
|
||||||
|
import * as crypto from 'crypto';
|
||||||
import { AiGatewayService } from '../ai/gateway/ai-gateway.service';
|
import { AiGatewayService } from '../ai/gateway/ai-gateway.service';
|
||||||
import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository';
|
import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository';
|
||||||
import { JobDefinitionRegistry } from './job-definition-registry';
|
import { JobDefinitionRegistry } from './job-definition-registry';
|
||||||
@ -142,13 +143,7 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
|||||||
snapshot = snap.content;
|
snapshot = snap.content;
|
||||||
}
|
}
|
||||||
|
|
||||||
// 7. Provider timeout via AbortController
|
// 7. 取消检查(Provider 调用前)
|
||||||
const controller = new AbortController();
|
|
||||||
const timeoutMs = def.execution.timeoutMs || 30000;
|
|
||||||
const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
|
|
||||||
|
|
||||||
try {
|
|
||||||
// 8. 取消检查(Provider 调用前)
|
|
||||||
const currentJob = await this.prisma.aiJob.findUnique({
|
const currentJob = await this.prisma.aiJob.findUnique({
|
||||||
where: { id: aiJobId },
|
where: { id: aiJobId },
|
||||||
select: { cancelRequestedAt: true },
|
select: { cancelRequestedAt: true },
|
||||||
@ -161,7 +156,10 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
|||||||
await context.updateProgress(30);
|
await context.updateProgress(30);
|
||||||
|
|
||||||
// ── EXECUTE ──
|
// ── EXECUTE ──
|
||||||
// 9. 调用 AiGatewayService(不直接导入 Provider SDK)
|
// 8. 调用 AiGatewayService(不直接导入 Provider SDK)
|
||||||
|
// timeout 通过 timeoutMs 参数委托给 AiGatewayService 内部的 AbortController
|
||||||
|
const timeoutMs = def.execution.timeoutMs || 30000;
|
||||||
|
try {
|
||||||
const response = await this.aiGateway.generate(
|
const response = await this.aiGateway.generate(
|
||||||
{
|
{
|
||||||
userId: job.userId,
|
userId: job.userId,
|
||||||
@ -169,15 +167,13 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
|||||||
tier: def.model.modelTier as any,
|
tier: def.model.modelTier as any,
|
||||||
promptKey: def.prompt.promptKey,
|
promptKey: def.prompt.promptKey,
|
||||||
promptVersion: def.prompt.promptVersion,
|
promptVersion: def.prompt.promptVersion,
|
||||||
messages: [], // Snapshot content 应通过 promptTemplate 渲染,此处由 AiGateway 处理
|
messages: [],
|
||||||
maxTokens: def.model.maxTokens,
|
maxTokens: def.model.maxTokens,
|
||||||
outputSchema: undefined, // 由 AiGateway 的 parseJson 处理
|
outputSchema: undefined,
|
||||||
},
|
},
|
||||||
timeoutMs,
|
timeoutMs,
|
||||||
);
|
);
|
||||||
|
|
||||||
clearTimeout(timeoutId);
|
|
||||||
|
|
||||||
await context.updateProgress(70);
|
await context.updateProgress(70);
|
||||||
|
|
||||||
// 10. 取消检查(Projector 前)
|
// 10. 取消检查(Projector 前)
|
||||||
@ -203,19 +199,17 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
|||||||
await context.updateProgress(90);
|
await context.updateProgress(90);
|
||||||
|
|
||||||
// ── COMPLETE ──
|
// ── COMPLETE ──
|
||||||
// 11. Usage logging
|
// 9. Usage logging
|
||||||
await this.writeUsageLog(job.userId, aiJobId, def, response, lockedJob.attemptCount || 0).catch((err) => {
|
await this.writeUsageLog(job.userId, aiJobId, def, response, lockedJob.attemptCount || 0).catch((err) => {
|
||||||
this.logger.error(`Usage log failed for job=${aiJobId}: ${err.message}`);
|
this.logger.error(`Usage log failed for job=${aiJobId}: ${err.message}`);
|
||||||
});
|
});
|
||||||
|
|
||||||
// 12. 标记成功
|
// 10. 标记成功
|
||||||
await this.lifecycleRepo.markSucceeded(aiJobId);
|
await this.lifecycleRepo.markSucceeded(aiJobId);
|
||||||
|
|
||||||
await context.updateProgress(100);
|
await context.updateProgress(100);
|
||||||
this.logger.log(`Job ${aiJobId} (${job.jobType}) completed successfully`);
|
this.logger.log(`Job ${aiJobId} (${job.jobType}) completed successfully`);
|
||||||
} catch (execErr: any) {
|
} catch (execErr: any) {
|
||||||
clearTimeout(timeoutId);
|
|
||||||
|
|
||||||
// 取消检查
|
// 取消检查
|
||||||
if (execErr?.message?.includes('cancelled')) {
|
if (execErr?.message?.includes('cancelled')) {
|
||||||
await this.lifecycleRepo.markCancelled(aiJobId);
|
await this.lifecycleRepo.markCancelled(aiJobId);
|
||||||
@ -282,14 +276,13 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private computeHash(content: string): string {
|
private computeHash(content: string): string {
|
||||||
const crypto = require('crypto');
|
|
||||||
return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16);
|
return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async writeUsageLog(
|
private async writeUsageLog(
|
||||||
userId: string,
|
userId: string,
|
||||||
jobId: string,
|
jobId: string,
|
||||||
_def: any,
|
definition: any,
|
||||||
response: any,
|
response: any,
|
||||||
attemptNo: number,
|
attemptNo: number,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
@ -300,8 +293,8 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
|||||||
provider: response?.usage?.provider || 'deepseek',
|
provider: response?.usage?.provider || 'deepseek',
|
||||||
model: response?.usage?.model || 'deepseek-chat',
|
model: response?.usage?.model || 'deepseek-chat',
|
||||||
tier: 'primary',
|
tier: 'primary',
|
||||||
promptKey: _def?.prompt?.promptKey || 'unknown',
|
promptKey: definition?.prompt?.promptKey || 'unknown',
|
||||||
promptVersion: _def?.prompt?.promptVersion || 'unknown',
|
promptVersion: definition?.prompt?.promptVersion || 'unknown',
|
||||||
inputTokens: response?.usage?.inputTokens || 0,
|
inputTokens: response?.usage?.inputTokens || 0,
|
||||||
outputTokens: response?.usage?.outputTokens || 0,
|
outputTokens: response?.usage?.outputTokens || 0,
|
||||||
estimatedCost: response?.usage?.estimatedCost || 0,
|
estimatedCost: response?.usage?.estimatedCost || 0,
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user