feat(ai): stream tool execution progress

This commit is contained in:
digua committed 2026-08-18 17:27:48 +08:00
1 parent cf1aee9e5b
commit 2e624f6beb
21 files changed
+227 -15

No files matched your search

+21
View File
@@ -123,4 +123,25 @@ describe('adaptToolsForAgent', () => {
assert.ok(text.includes('hello world'))
assert.ok(text.includes('hi there'))
})
it('bridges shared tool progress to Pi tool updates', async () => {
const tool: ToolDefinition = {
name: 'slow_tool',
description: 'Slow tool',
inputSchema: { type: 'object', properties: {} },
async handler(_params, context) {
context.reportProgress?.({ phase: 'searching' })
return { content: 'Done' }
},
}
const [agentTool] = adaptToolsForAgent([tool], () => ({
db: {} as DatabaseAdapter,
sessionId: 'session-1',
}))
const updates: unknown[] = []
await agentTool.execute('call-progress', {}, undefined, (update) => updates.push(update))
assert.deepEqual(updates, [{ content: [], details: { progress: { phase: 'searching' } } }])
})
})
+3 -1
View File
@@ -74,13 +74,15 @@ export function adaptToolsForAgent(
description: tool.description,
parameters: toAgentToolParameters(tool.inputSchema) as any,
executionMode: tool.executionMode,
async execute(_toolCallId: string, params: unknown) {
async execute(_toolCallId: string, params: unknown, signal, onUpdate) {
const ctx = getContext()
const execCtx: ToolExecutionContext = {
db: ctx.db,
dataProvider: new CoreDataProvider(ctx.db),
sessionId: ctx.sessionId,
locale: ctx.locale,
abortSignal: signal,
reportProgress: (progress) => onUpdate?.({ content: [], details: { progress } }),
semanticIndexService: ctx.semanticIndexService,
preprocessConfig: ctx.preprocessConfig,
ownerPlatformId: ctx.ownerPlatformId,
+2 -2
View File
@@ -90,8 +90,8 @@ function wrapWithPreprocessing(tool: AgentTool<any>, context: ToolContext): Agen
const originalExecute = tool.execute
return {
...tool,
execute: async (toolCallId: string, params: any, _signal?: AbortSignal, _onUpdate?: unknown) => {
const result = await originalExecute(toolCallId, params)
execute: async (toolCallId: string, params: any, signal, onUpdate) => {
const result = await originalExecute(toolCallId, params, signal, onUpdate)
const details = result.details as Record<string, unknown> | undefined
if (!details?.rawMessages || !Array.isArray(details.rawMessages)) {
@@ -10,6 +10,7 @@ import {
toAgentToolParameters,
type ToolDefinition,
type ToolExecutionContext,
type ToolProgress,
type RawMessage,
} from '@openchatlab/tools'
import type { AgentTool, PreprocessableMessage, PreprocessConfig } from '@openchatlab/node-runtime'
@@ -18,13 +19,19 @@ import type { ToolContext, ToolRegistryEntry } from './types'
import { WorkerDataProvider } from './worker-data-provider'
import { t as i18nT } from '../../i18n'
function buildExecutionContext(ctx: ToolContext): ToolExecutionContext {
function buildExecutionContext(
ctx: ToolContext,
signal?: AbortSignal,
reportProgress?: (progress: ToolProgress) => void
): ToolExecutionContext {
const abortSignal = signal ?? ctx.abortSignal
return {
dataProvider: new WorkerDataProvider(ctx.sessionId, ctx.abortSignal),
dataProvider: new WorkerDataProvider(ctx.sessionId, abortSignal),
sessionId: ctx.sessionId,
locale: ctx.locale,
timeFilter: ctx.timeFilter,
abortSignal: ctx.abortSignal,
abortSignal,
reportProgress,
searchContextBefore: ctx.searchContextBefore,
searchContextAfter: ctx.searchContextAfter,
maxMessagesLimit: ctx.maxMessagesLimit,
@@ -59,8 +66,12 @@ export function adaptSharedTool(tool: ToolDefinition): ToolRegistryEntry {
description: tool.description,
parameters: toAgentToolParameters(tool.inputSchema) as any,
executionMode: tool.executionMode,
async execute(_toolCallId: string, params: unknown) {
return executeToolForAgent(tool, params, buildExecutionContext(context))
async execute(_toolCallId: string, params: unknown, signal, onUpdate) {
return executeToolForAgent(
tool,
params,
buildExecutionContext(context, signal, (progress) => onUpdate?.({ content: [], details: { progress } }))
)
},
}
},
@@ -88,6 +88,8 @@ function emitMessage(stream: AssistantMessageEventStream, message: AssistantMess
if (message.stopReason === 'error' || message.stopReason === 'aborted') {
stream.push({ type: 'error', reason: message.stopReason, error: message })
} else if (message.stopReason === 'pending') {
throw new Error('A scripted assistant message must be complete before it is emitted')
} else {
stream.push({ type: 'done', reason: message.stopReason, message })
}
@@ -219,6 +221,37 @@ describe('runAgentCore runtime contract', () => {
)
})
it('forwards tool execution progress as a stable core event', async () => {
const events: AgentCoreEvent[] = []
const tool = createTool('lookup', async (_toolCallId, _params, _signal, onUpdate) => {
onUpdate?.({ content: [], details: { progress: { phase: 'searching' } } })
return { content: [{ type: 'text', text: 'tool result' }], details: null }
})
await runAgentCore(
createOptions({
tools: [tool],
streamFn: createScriptedStream([
assistantMessage([{ type: 'toolCall', id: 'tool-progress', name: 'lookup', arguments: {} }], {
stopReason: 'toolUse',
}),
assistantMessage([{ type: 'text', text: 'Done' }]),
]),
onEvent: (event) => events.push(event),
})
)
assert.deepEqual(
events.find((event) => event.type === 'tool_update'),
{
type: 'tool_update',
toolCallId: 'tool-progress',
toolName: 'lookup',
progress: { phase: 'searching' },
}
)
})
it('turns thrown tool errors into error tool results and keeps the agent loop recoverable', async () => {
const events: AgentCoreEvent[] = []
const tool = createTool('broken_tool', async () => {
@@ -45,6 +45,34 @@ describe('AgentEventHandler', () => {
assert.equal(startChunk?.toolCallId, 'call_1')
})
it('forwards tool progress without changing the result payload', () => {
const chunks: AgentStreamChunk[] = []
const handler = new AgentEventHandler({
onChunk: (chunk) => chunks.push(chunk),
context: {},
systemPrompt: 'test',
})
handler.handleCoreEvent(
{
type: 'tool_update',
toolCallId: 'call_1',
toolName: 'retrieve_chat_evidence',
progress: { phase: 'semantic_search' },
},
[]
)
assert.deepEqual(chunks, [
{
type: 'tool_update',
toolCallId: 'call_1',
toolName: 'retrieve_chat_evidence',
toolProgress: { phase: 'semantic_search' },
},
])
})
it('updates tool rounds on turn_end', () => {
const handler = new AgentEventHandler({
onChunk: () => {},
@@ -9,6 +9,7 @@ import { Agent as PiAgentCore } from '@earendil-works/pi-agent-core'
import type { AgentEvent as PiAgentEvent, AgentMessage as PiAgentMessage } from '@earendil-works/pi-agent-core'
import { type Message as PiMessage, type Usage as PiUsage, clampThinkingLevel } from '@earendil-works/pi-ai'
import { StreamingThinkTagParser, needsStreamingThinkParsing } from '@openchatlab/core'
import type { ToolProgress } from '@openchatlab/shared-types'
import type { AgentCoreOptions, AgentCoreResult, AgentTokenUsage } from './types'
import { initTokenizer } from '../tokenizer'
@@ -197,6 +198,16 @@ export async function runAgentCore(options: AgentCoreOptions): Promise<AgentCore
toolName: event.toolName,
toolParams: (event.args || {}) as Record<string, unknown>,
})
} else if (event.type === 'tool_execution_update') {
const details = event.partialResult?.details as { progress?: ToolProgress } | null
if (details?.progress) {
onEvent({
type: 'tool_update',
toolCallId: event.toolCallId,
toolName: event.toolName,
progress: details.progress,
})
}
} else if (event.type === 'tool_execution_end') {
onEvent({
type: 'tool_end',
@@ -6,6 +6,7 @@
*/
import type { Message as PiMessage } from '@earendil-works/pi-ai'
import type { ToolProgress } from '@openchatlab/shared-types'
import type { AgentCoreEvent } from './types'
import type { PlanContentBlock } from './planning-types'
import type { RouteDecision } from './routing-types'
@@ -35,6 +36,7 @@ export interface AgentStreamChunk {
| 'content'
| 'think'
| 'tool_start'
| 'tool_update'
| 'tool_result'
| 'status'
| 'compression_done'
@@ -50,6 +52,7 @@ export interface AgentStreamChunk {
toolCallId?: string
toolName?: string
toolParams?: Record<string, unknown>
toolProgress?: ToolProgress
toolResult?: unknown
toolIsError?: boolean
error?: unknown
@@ -187,6 +190,14 @@ export class AgentEventHandler {
this.emitStatus('tool_running', messages, { currentTool: event.toolName, force: true })
break
}
case 'tool_update':
this.onChunk({
type: 'tool_update',
toolCallId: event.toolCallId,
toolName: event.toolName,
toolProgress: event.progress,
})
break
case 'tool_end':
this.onChunk({
type: 'tool_result',
@@ -8,6 +8,7 @@
import type { AgentTool } from '@earendil-works/pi-agent-core'
import type { Model, Api, Message } from '@earendil-works/pi-ai'
import type { ThinkingLevel } from '@openchatlab/core'
import type { ToolProgress } from '@openchatlab/shared-types'
import type { ContentBlock } from '../chats'
export interface AgentTokenUsage {
@@ -31,6 +32,7 @@ export type AgentCoreEvent =
| { type: 'thinking_delta'; content: string }
| { type: 'thinking_end'; durationMs?: number }
| { type: 'tool_start'; toolCallId: string; toolName: string; toolParams: Record<string, unknown> }
| { type: 'tool_update'; toolCallId: string; toolName: string; progress: ToolProgress }
| { type: 'tool_end'; toolCallId: string; toolName: string; toolResult: unknown; isError: boolean }
| { type: 'turn_end'; round: number; hadToolCalls: boolean }
| { type: 'usage_update'; usage: AgentTokenUsage }
+9
View File
@@ -39,6 +39,15 @@ export type NavigationLayoutLoadResult =
| { status: 'saved'; layout: NavigationLayout }
| { status: 'missing' | 'invalid'; layout: null }
// ==================== AI tool progress ====================
/** Stable, transport-safe progress payload emitted by long-running AI tools. */
export interface ToolProgress {
phase: string
current?: number
total?: number
}
// ==================== 时间筛选 ====================
export interface TimeFilter {
@@ -86,6 +86,7 @@ describe('retrieveChatEvidenceTool schema', () => {
describe('retrieveChatEvidenceTool mode resolution', () => {
it('auto resolves to hybrid and calls both semantic + keyword when available with keywords', async () => {
let semanticCalled = false
const progressPhases: string[] = []
const capture: { args?: unknown[] } = {}
const service = makeService({
searchForTool: async () => {
@@ -96,12 +97,22 @@ describe('retrieveChatEvidenceTool mode resolution', () => {
const dp = makeDataProvider([rawMsg(200, 1714521600, '我们去乐山玩了')], capture)
const res = await retrieveChatEvidenceTool.handler(
{ query: '去过乐山几次', criteria: '实际出行', keywords: ['乐山'] },
makeContext({ semanticIndexService: service, dataProvider: dp })
makeContext({
semanticIndexService: service,
dataProvider: dp,
reportProgress: (progress) => progressPhases.push(progress.phase),
})
)
assert.equal(semanticCalled, true)
assert.ok(capture.args, 'keyword search should be called')
const payload = getPayload(res.data)
assert.equal(payload.mode, 'hybrid')
assert.deepEqual(progressPhases, [
'checking_capabilities',
'semantic_search',
'keyword_search',
'assembling_evidence',
])
})
it('semantic mode does not call keyword search', async () => {
@@ -347,6 +347,7 @@ async function handler(params: Record<string, unknown>, context: ToolExecutionCo
? params.mode
: 'auto'
context.reportProgress?.({ phase: 'checking_capabilities' })
const service = context.semanticIndexService
const semanticAvailable = !!service && (await service.canSearch(context.sessionId))
const rangeMs = parseEvidenceTimeRange(params, context)
@@ -367,6 +368,7 @@ async function handler(params: Record<string, unknown>, context: ToolExecutionCo
if (!service || !semanticAvailable) {
warnings.add('semantic_unavailable')
} else {
context.reportProgress?.({ phase: 'semantic_search' })
const result = await service.searchForTool(context.sessionId, query, {
maxResults: clampMaxResults(params.max_results),
preprocessConfig: context.preprocessConfig,
@@ -408,6 +410,7 @@ async function handler(params: Record<string, unknown>, context: ToolExecutionCo
} else if (!context.dataProvider) {
warnings.add('keyword_unavailable')
} else {
context.reportProgress?.({ phase: 'keyword_search' })
const secFilter = toKeywordTimeFilter(rangeMs)
const searchResult = await context.dataProvider.searchMessages(keywords, {
timeFilter: secFilter,
@@ -418,6 +421,7 @@ async function handler(params: Record<string, unknown>, context: ToolExecutionCo
}
}
context.reportProgress?.({ phase: 'assembling_evidence' })
const merged = mergeCandidates(semanticCandidates, keywordCandidates)
const groups = buildGroups(merged, !!criteria, locale)
+1
View File
@@ -53,6 +53,7 @@ export { parseExtendedTimeParams } from './utils/time-params'
export type {
ToolDefinition,
ToolExecutionContext,
ToolProgress,
ToolResult,
JsonSchema,
RawMessage,
+5
View File
@@ -5,6 +5,9 @@
*/
import type { ChartPayload, DatabaseAdapter, EvidenceTimeRangeMs } from '@openchatlab/core'
import type { ToolProgress } from '@openchatlab/shared-types'
export type { ToolProgress } from '@openchatlab/shared-types'
export type {
EvidenceRetrievalMode,
@@ -301,6 +304,8 @@ export interface ToolExecutionContext {
locale?: string
timeFilter?: ToolTimeRange
abortSignal?: AbortSignal
/** Report meaningful phases for long-running tool execution. */
reportProgress?: (progress: ToolProgress) => void
/** 抽象查询接口 */
dataProvider?: ToolDataProvider
/** 语义检索窄接口(仅当前会话可检索时由 adapter 注入) */
@@ -2,6 +2,7 @@
import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
import type { AgentRuntimeStatus } from '@electron/shared/types'
import type { ToolProgress } from '@openchatlab/shared-types'
const { t, te } = useI18n()
@@ -11,6 +12,12 @@ function localizedToolName(name: string, fallback?: string): string {
return te(key) ? t(key) : fallback || name
}
function localizedToolProgress(progress?: ToolProgress): string {
if (!progress) return ''
const key = `ai.chat.thinking.toolProgress.${progress.phase}`
return te(key) ? t(key) : ''
}
// Props
const props = defineProps<{
// 当前工具执行状态
@@ -18,6 +25,7 @@ const props = defineProps<{
name: string
displayName: string
status: 'running' | 'done' | 'error'
progress?: ToolProgress
} | null
// 当前轮次已使用的工具列表
toolsUsed: string[]
@@ -83,6 +91,13 @@ const isCompressing = computed(() => props.agentStatus?.phase === 'compressing')
<span class="h-1.5 w-1.5 animate-bounce rounded-full bg-pink-500 [animation-delay:300ms]" />
</span>
<span
v-if="currentToolStatus.status === 'running' && localizedToolProgress(currentToolStatus.progress)"
class="text-xs text-gray-500 dark:text-gray-400"
>
{{ localizedToolProgress(currentToolStatus.progress) }}
</span>
<!-- 完成后处理中状态 -->
<span
v-else-if="currentToolStatus.status === 'done'"
+7 -1
View File
@@ -115,7 +115,13 @@
"processingResult": "Processing result",
"called": "Called:",
"analyzing": "Analyzing question...",
"compressing": "Compressing context..."
"compressing": "Compressing context...",
"toolProgress": {
"checking_capabilities": "Checking search capabilities",
"semantic_search": "Running semantic search",
"keyword_search": "Running keyword search",
"assembling_evidence": "Assembling evidence"
}
},
"statusBar": {
"model": {
+7 -1
View File
@@ -115,7 +115,13 @@
"processingResult": "結果を整理中",
"called": "実行したツール:",
"analyzing": "問題を分析中...",
"compressing": "コンテキストを圧縮中..."
"compressing": "コンテキストを圧縮中...",
"toolProgress": {
"checking_capabilities": "検索機能を確認中",
"semantic_search": "セマンティック検索を実行中",
"keyword_search": "キーワード検索を実行中",
"assembling_evidence": "証拠を整理中"
}
},
"statusBar": {
"model": {
+7 -1
View File
@@ -115,7 +115,13 @@
"processingResult": "处理结果中",
"called": "已调用:",
"analyzing": "正在分析问题...",
"compressing": "正在压缩上下文..."
"compressing": "正在压缩上下文...",
"toolProgress": {
"checking_capabilities": "正在检查检索能力",
"semantic_search": "正在进行语义检索",
"keyword_search": "正在进行关键词检索",
"assembling_evidence": "正在整理证据"
}
},
"statusBar": {
"model": {
+7 -1
View File
@@ -115,7 +115,13 @@
"processingResult": "正在整理結果",
"called": "已執行:",
"analyzing": "正在分析問題...",
"compressing": "正在壓縮上下文..."
"compressing": "正在壓縮上下文...",
"toolProgress": {
"checking_capabilities": "正在檢查檢索能力",
"semantic_search": "正在進行語意檢索",
"keyword_search": "正在進行關鍵字檢索",
"assembling_evidence": "正在整理證據"
}
},
"statusBar": {
"model": {
+3 -1
View File
@@ -1,5 +1,5 @@
import type { PlanContentBlock } from '@/services/ai/planBlocks'
import type { ChartAutoMode } from '@openchatlab/shared-types'
import type { ChartAutoMode, ToolProgress } from '@openchatlab/shared-types'
export interface LlmStreamChunk {
content: string
@@ -15,6 +15,7 @@ export interface AgentStreamChunk {
| 'content'
| 'think'
| 'tool_start'
| 'tool_update'
| 'tool_result'
| 'status'
| 'compression_done'
@@ -30,6 +31,7 @@ export interface AgentStreamChunk {
toolCallId?: string
toolName?: string
toolParams?: Record<string, unknown>
toolProgress?: ToolProgress
toolResult?: unknown
toolIsError?: boolean
error?: unknown
+23 -1
View File
@@ -20,7 +20,7 @@ import { useAgentStreamService } from '@/services/ai-stream/service'
import { buildSerializablePreprocessConfig, shouldEnsureDesensitizeRulesBeforeSerialize } from './aiPreprocessConfig'
import type { ChartPayload, ChatEvidencePayload } from '@openchatlab/core'
import { extractToolResultText, truncateToolResultText } from '@openchatlab/core'
import { getDefaultGeneralAssistantId } from '@openchatlab/shared-types'
import { getDefaultGeneralAssistantId, type ToolProgress } from '@openchatlab/shared-types'
import {
createRenderOnlyToolPendingBlock,
extractChartPayloads,
@@ -129,6 +129,7 @@ export interface ToolStatus {
name: string
displayName: string
status: 'running' | 'done' | 'error'
progress?: ToolProgress
result?: unknown
}
@@ -1064,6 +1065,17 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => {
}
break
case 'tool_update':
if (
chunk.toolName &&
chunk.toolProgress &&
state.currentToolStatus?.name === chunk.toolName &&
state.currentToolStatus.status === 'running'
) {
state.currentToolStatus = { ...state.currentToolStatus, progress: chunk.toolProgress }
}
break
case 'tool_result':
if (chunk.toolName) {
const charts = extractChartPayloads(chunk.toolResult)
@@ -1643,6 +1655,16 @@ export const useAIChatStore = defineStore('aiChatRuntime', () => {
}
}
break
case 'tool_update':
if (
chunk.toolName &&
chunk.toolProgress &&
state.currentToolStatus?.name === chunk.toolName &&
state.currentToolStatus.status === 'running'
) {
state.currentToolStatus = { ...state.currentToolStatus, progress: chunk.toolProgress }
}
break
case 'tool_result':
if (chunk.toolName) {
const charts = extractChartPayloads(chunk.toolResult)