fix(server): align web AI agent with Electron implementation

- Replace mapCoreEventToStream with shared AgentEventHandler for rich
  status events (contextTokens, totalUsage, updatedAt)
- Add formatAIError for user-friendly error messages (429/503/quota)
- Compute maxToolResultTokens dynamically from model contextWindow
- Add steerMessage (answerWithoutTools) and onConvertToLlm callback
- Forward ownerInfo/mentionedMembers from web frontend to server
This commit is contained in:
digua committed 2026-05-18 22:23:44 +08:00
1 parent 7190606ca7
commit 58a558b7aa
4 files changed
+131 -85

No files matched your search

+111 -76
View File
@@ -2,7 +2,7 @@
* 服务端 Agent
*
* 使用 @openchatlab/node-runtime 的 runAgentCore 编排对话流程,
* 将流式事件通过回调输出给 SSE 端点。
* 通过 AgentEventHandler 输出与 Electron 端一致的流式事件。
*/
import {
@@ -11,7 +11,9 @@ import {
checkAndCompress,
buildSystemPrompt,
createAiTranslate,
type AgentCoreEvent,
AgentEventHandler,
type AgentStreamChunk,
type PiMessage,
type SimpleHistoryMessage,
type AIConversationManager,
type CompressionConfig,
@@ -26,24 +28,7 @@ import {
import { getDefaultAssistantConfig, buildPiModel } from './llm-config'
import { getServerAiLogger } from './logger'
export interface AgentStreamEvent {
type: 'content' | 'think' | 'tool_start' | 'tool_result' | 'status' | 'done' | 'error'
content?: string
thinkTag?: string
thinkDurationMs?: number
toolName?: string
toolParams?: Record<string, unknown>
toolResult?: unknown
error?: { name: string | null; message: string | null }
isFinished?: boolean
usage?: { promptTokens: number; completionTokens: number; totalTokens: number }
status?: {
phase: string
round: number
toolsUsed: number
currentTool?: string
}
}
export type { AgentStreamChunk }
export interface RunAgentOptions {
userMessage: string
@@ -56,57 +41,83 @@ export interface RunAgentOptions {
tools?: AgentTool[]
aiDataDir: string
convManager: AIConversationManager
onEvent: (event: AgentStreamEvent) => void
onEvent: (event: AgentStreamChunk) => void
abortSignal?: AbortSignal
ownerInfo?: OwnerInfo
mentionedMembers?: MentionedMember[]
dataSnapshot?: DataSnapshot
}
function mapCoreEventToStream(
event: AgentCoreEvent,
onEvent: (event: AgentStreamEvent) => void,
toolsUsedCount: { value: number },
currentRound: { value: number }
): void {
switch (event.type) {
case 'content':
onEvent({ type: 'content', content: event.content })
break
case 'thinking_start':
onEvent({
type: 'status',
status: { phase: 'thinking', round: currentRound.value, toolsUsed: toolsUsedCount.value },
})
break
case 'thinking_delta':
onEvent({ type: 'think', content: event.content, thinkTag: 'thinking' })
break
case 'thinking_end':
onEvent({ type: 'think', content: '', thinkTag: 'thinking', thinkDurationMs: event.durationMs })
break
case 'tool_start':
toolsUsedCount.value += 1
onEvent({ type: 'tool_start', toolName: event.toolName, toolParams: event.toolParams })
onEvent({
type: 'status',
status: {
phase: 'tool_running',
round: currentRound.value,
toolsUsed: toolsUsedCount.value,
currentTool: event.toolName,
},
})
break
case 'tool_end':
onEvent({ type: 'tool_result', toolName: event.toolName, toolResult: event.toolResult })
break
case 'turn_end':
currentRound.value = event.round
break
case 'usage_update':
break
/**
* Format AI errors with user-friendly messages (mirrors Electron's formatAIError)
*/
function formatAIError(error: unknown): string {
const candidates: unknown[] = []
if (error) candidates.push(error)
const errorObj = error as { lastError?: unknown; errors?: unknown[] }
if (errorObj?.lastError) candidates.push(errorObj.lastError)
if (Array.isArray(errorObj?.errors)) candidates.push(...errorObj.errors)
let rawMessage = ''
let statusCode: number | undefined
let retrySeconds: number | undefined
for (const candidate of candidates) {
if (!candidate || typeof candidate !== 'object') {
if (!rawMessage && typeof candidate === 'string') rawMessage = candidate
continue
}
const record = candidate as Record<string, unknown>
if (typeof record.statusCode === 'number') statusCode = record.statusCode
if (!rawMessage && typeof record.message === 'string') rawMessage = record.message
if (!rawMessage && record.data && typeof record.data === 'object') {
const data = record.data as { error?: { message?: string } }
if (data.error?.message) rawMessage = data.error.message
}
if (record.responseBody && typeof record.responseBody === 'string') {
try {
const parsed = JSON.parse(record.responseBody) as { error?: { message?: string } }
if (!rawMessage && parsed.error?.message) rawMessage = parsed.error.message
} catch {
if (!rawMessage) rawMessage = record.responseBody
}
}
if (rawMessage) {
const retryMatch = rawMessage.match(/retry in ([0-9.]+)s/i)
if (retryMatch) retrySeconds = Math.ceil(Number(retryMatch[1]))
}
}
const fallbackMessage = rawMessage || String(error)
const lowerMessage = fallbackMessage.toLowerCase()
if (statusCode === 429 || lowerMessage.includes('quota') || lowerMessage.includes('resource_exhausted')) {
return retrySeconds
? `API quota exhausted, please retry after ${retrySeconds}s or upgrade your quota.`
: `API quota exhausted, please retry later or upgrade your quota.`
}
if (
statusCode === 403 &&
(lowerMessage.includes('quota') || lowerMessage.includes('not enough') || lowerMessage.includes('insufficient'))
) {
return `API rejected the request due to insufficient quota or balance.`
}
if (statusCode === 503 || lowerMessage.includes('overloaded') || lowerMessage.includes('unavailable')) {
return `Model is overloaded, please retry later.`
}
if (fallbackMessage.length > 300) {
return `${fallbackMessage.slice(0, 300)}...`
}
return fallbackMessage
}
export async function runServerAgent(options: RunAgentOptions): Promise<void> {
@@ -128,6 +139,8 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
dataSnapshot,
} = options
const aiLogger = getServerAiLogger()
const llmConfig = getDefaultAssistantConfig(aiDataDir)
if (!llmConfig) {
onEvent({ type: 'error', error: { name: 'ConfigError', message: 'LLM service not configured' } })
@@ -154,11 +167,17 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
dataSnapshot,
})
const handler = new AgentEventHandler({
onChunk: onEvent,
context: {},
systemPrompt,
})
if (compressionConfig?.enabled) {
const llmAdapter: CompressionLlmAdapter = {
contextWindow: piModel.contextWindow ?? 128000,
compress: async (prompt: string, maxTokens: number) => {
onEvent({ type: 'status', status: { phase: 'compressing', round: 0, toolsUsed: 0 } })
handler.emitStatus('compressing', [])
try {
const result = await completeSimple(
piModel,
@@ -178,7 +197,6 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
}
},
}
const aiLogger = getServerAiLogger()
const compressionResult = await checkAndCompress(
conversationId,
compressionConfig,
@@ -188,12 +206,21 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
aiLogger ?? undefined
)
if (compressionResult.compressed) {
onEvent({ type: 'status', status: { phase: 'compression_done', round: 0, toolsUsed: 0 } })
onEvent({
type: 'compression_done',
compressionResult: {
summaryContent: compressionResult.summaryContent ?? '',
tokensBefore: compressionResult.tokensBefore ?? 0,
tokensAfter: compressionResult.tokensAfter ?? 0,
timestamp: Date.now(),
},
})
}
}
if (abortSignal?.aborted) {
onEvent({ type: 'done', isFinished: true, usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 } })
handler.emitStatus('aborted', [], { force: true })
onEvent({ type: 'done', isFinished: true, usage: handler.cloneUsage() })
return
}
@@ -204,10 +231,10 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
// empty history on failure
}
const toolsUsedCount = { value: 0 }
const currentRound = { value: 0 }
handler.emitStatus('preparing', [], { pendingUserMessage: userMessage, force: true })
onEvent({ type: 'status', status: { phase: 'preparing', round: 0, toolsUsed: 0 } })
const steerMessage = t('ai.agent.answerWithoutTools')
let cachedMessages: PiMessage[] = []
try {
const result = await runAgentCore({
@@ -219,7 +246,11 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
userMessage,
maxToolRounds: 5,
abortSignal,
onEvent: (coreEvent) => mapCoreEventToStream(coreEvent, onEvent, toolsUsedCount, currentRound),
steerMessage,
onConvertToLlm: (filteredMessages) => {
cachedMessages = filteredMessages as PiMessage[]
},
onEvent: (coreEvent) => handler.handleCoreEvent(coreEvent, cachedMessages),
onDebugContext: (messages) => {
try {
convManager.setPendingDebugContext(conversationId, JSON.stringify(messages, null, 2))
@@ -230,13 +261,17 @@ export async function runServerAgent(options: RunAgentOptions): Promise<void> {
})
if (result.error) {
onEvent({ type: 'error', error: { name: 'AgentError', message: result.error } })
const friendlyMessage = formatAIError(result.error)
onEvent({ type: 'error', error: { name: 'AgentError', message: friendlyMessage } })
}
handler.emitStatus('completed', cachedMessages, { force: true })
onEvent({ type: 'done', isFinished: true, usage: result.usage })
} catch (error) {
const msg = error instanceof Error ? error.message : String(error)
onEvent({ type: 'error', error: { name: 'AgentError', message: msg } })
onEvent({ type: 'done', isFinished: true, usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 } })
const friendlyMessage = formatAIError(error)
aiLogger?.error('ServerAgent', 'Agent execution error', { error: String(error) })
handler.emitStatus('error', cachedMessages, { force: true })
onEvent({ type: 'error', error: { name: 'AgentError', message: friendlyMessage } })
onEvent({ type: 'done', isFinished: true, usage: handler.cloneUsage() })
}
}
+9 -2
View File
@@ -48,10 +48,17 @@ function convertJsonSchemaToParameters(schema: ToolDefinition['inputSchema']) {
}
}
export interface AdaptToolsOptions {
maxToolResultTokens?: number
}
export function adaptToolsForAgent(
tools: ToolDefinition[],
getContext: () => ServerToolContext
getContext: () => ServerToolContext,
options?: AdaptToolsOptions
): AgentTool<any, any>[] {
const tokenBudget = options?.maxToolResultTokens ?? DEFAULT_MAX_TOOL_RESULT_TOKENS
return tools.map((tool) => ({
name: tool.name,
label: tool.name,
@@ -73,7 +80,7 @@ export function adaptToolsForAgent(
const pipelineResult = applyPreprocessingPipeline({
rawMessages: result.rawMessages as PreprocessableMessage[],
locale: ctx.locale,
maxToolResultTokens: DEFAULT_MAX_TOOL_RESULT_TOKENS,
maxToolResultTokens: tokenBudget,
truncationStrategy: TOOL_TRUNCATION_STRATEGY[tool.name] ?? 'keep_last',
extraDetails: (result.data ?? {}) as Record<string, unknown>,
logger: getServerAiLogger() ?? undefined,
+9 -7
View File
@@ -25,9 +25,10 @@ import {
import type { DataSnapshot, OwnerInfo, MentionedMember } from '@openchatlab/node-runtime'
import { AGENT_TOOL_REGISTRY } from '@openchatlab/tools'
import { adaptToolsForAgent } from '../../ai/tool-adapter'
import { getDefaultAssistantConfig, buildPiModel } from '../../ai/llm-config'
import { loadAssistantConfig } from '../../ai/assistant-loader'
import { getAssistantManager, getSkillManagerCore } from '../../ai/manager-factory'
import { runServerAgent, type AgentStreamEvent } from '../../ai/agent'
import { runServerAgent, type AgentStreamChunk } from '../../ai/agent'
import {
addLlmConfig,
updateLlmConfig,
@@ -534,13 +535,14 @@ export function registerAiRoutes(
}
}
const llmConfig = getDefaultAssistantConfig(aiDataDir)
const maxToolResultPercent = compressionConfig?.maxToolResultPercent ?? 50
const contextWindow = llmConfig ? (buildPiModel(llmConfig).contextWindow ?? 128000) : 128000
const maxToolResultTokens = Math.floor(contextWindow * (maxToolResultPercent / 100))
const db = (dbManager as any).open?.(sessionId)
const agentTools = db
? adaptToolsForAgent(AGENT_TOOL_REGISTRY, () => ({
db,
sessionId,
locale,
}))
? adaptToolsForAgent(AGENT_TOOL_REGISTRY, () => ({ db, sessionId, locale }), { maxToolResultTokens })
: []
const skillManager = new SkillManager(aiDataDir)
@@ -558,7 +560,7 @@ export function registerAiRoutes(
agentTools.push(activateSkillTool as any)
}
const onEvent = (event: AgentStreamEvent) => {
const onEvent = (event: AgentStreamChunk) => {
sendSSE(event.type, event)
if (event.type === 'done') {
activeAgentAborts.delete(requestId)
+2
View File
@@ -280,6 +280,8 @@ async function installAiApiShims(): Promise<void> {
locale: locale || 'zh-CN',
assistantId,
compressionConfig: _compressionConfig,
ownerInfo: context.ownerInfo,
mentionedMembers: context.mentionedMembers,
},
signal: abortController.signal,
onEvent: ({ event, data }) => {