Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -43,13 +43,10 @@ export function useStreamManager() {
currentStream = new ConversationStream(
response,
onStream,
() => {
currentStream = null
onFinish?.()
},
(e) => {
currentStream = null
onFailure?.(e)
if (e) onFailure?.(e)
else onFinish?.()
},
)
currentStream.start()
Expand Down Expand Up @@ -92,13 +89,10 @@ export function useStreamManager() {
currentStream = new ConversationStream(
response,
onStream,
() => {
currentStream = null
onFinish?.()
},
() => {
(e) => {
currentStream = null
onFailure?.()
if (e) onFailure?.()
else onFinish?.()
},
)
currentStream.start()
Expand Down
45 changes: 21 additions & 24 deletions ui/src/conversation-panel/stream.ts
Original file line number Diff line number Diff line change
@@ -1,43 +1,40 @@
export class ConversationStream {
private response: any
private onChunk: (chunk: any) => void
private onFinish: () => void
private onError: (e: any) => void
private onNext: (chunk: any) => void
private onComplete: (e?: any) => void
private cancelled = false
private finished = false
private completed = false
private reader: ReadableStreamDefaultReader<any> | null = null

constructor(
response: any,
onChunk: (chunk: any) => void,
onFinish: () => void,
onError: (e: any) => void,
onNext: (chunk: any) => void,
onComplete: (e?: any) => void,
) {
this.response = response
this.onChunk = onChunk
this.onFinish = onFinish
this.onError = onError
this.onComplete = onComplete
this.onNext = onNext
}

private finish() {
if (this.finished || this.cancelled) return
this.finished = true
this.onFinish()
private complete(e?:any) {
if (this.completed || this.cancelled) return
this.completed = true
this.onComplete(e)
}

async start() {
try {
this.reader = this.response.body?.getReader()
if (!this.reader) {
this.finish()
this.complete()
return
}

const decoder = new TextDecoder()
let buffer = ''

while (true) {
if (this.cancelled || this.finished) break
if (this.cancelled || this.completed) break

const { done, value } = await this.reader.read()

Expand All @@ -50,21 +47,21 @@ export class ConversationStream {
buffer = lines.pop() || ''

for (const line of lines) {
if (this.cancelled || this.finished) break
if (this.cancelled || this.completed) break
const trimmed = line.trim()
if (!trimmed) continue

// SSE 格式:data: {...}
if (trimmed.startsWith('data:')) {
const data = trimmed.slice(5).trim()
if (data === '[DONE]') {
this.finish()
this.complete()
return
}

try {
const chunk = JSON.parse(data)
this.onChunk(chunk)
this.onNext(chunk)
} catch (e) {
// Skip invalid JSON
}
Expand All @@ -73,14 +70,14 @@ export class ConversationStream {
}

// 处理 buffer 中剩余的数据
if (!this.cancelled && !this.finished && buffer.trim()) {
if (!this.cancelled && !this.completed && buffer.trim()) {
const trimmed = buffer.trim()
if (trimmed.startsWith('data:')) {
const data = trimmed.slice(5).trim()
if (data !== '[DONE]') {
try {
const chunk = JSON.parse(data)
this.onChunk(chunk)
this.onNext(chunk)
} catch (e) {
// Skip invalid JSON
}
Expand All @@ -89,11 +86,11 @@ export class ConversationStream {
}

if (!this.cancelled) {
this.finish()
this.complete()
}
} catch (e) {
if (!this.cancelled && !this.finished) {
this.onError(e)
if (!this.cancelled && !this.completed) {
this.complete(e)
}
} finally {
this.reader = null
Expand Down
30 changes: 15 additions & 15 deletions ui/src/views/workflow/tool/debug/ResultDrawer.vue
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import type { ToolWorkflowRecord } from '@/api/types'
import { MsgError } from '@/utils/message'
import { ConversationStream } from '@/conversation-panel/stream'
import { aggregators } from '@/conversation-panel'
import ContentItem from '@/conversation-panel/content/index.vue'
import ContentList from '@/conversation-panel/content-list/index.vue'
import ExecutionDetailContent from '@/workflow-canvas/details/index.vue'
import { WorkflowMode } from '@/workflow-canvas/types'

Expand All @@ -25,6 +25,8 @@ interface ResumeParameters {
formData?: Record<string, unknown>
position?: unknown
chunkId?: string
chatRecordId?:string

}
const props = defineProps<{ toolId: string }>()
const running = defineModel<boolean>('running', { default: false })
Expand Down Expand Up @@ -71,9 +73,9 @@ async function execute(extra: Record<string, unknown> = {}) {
stream = new ConversationStream(
response,
receiveChunk,
() => {},
(error: unknown) => {
streamError = error
(error?: unknown) => {
if (error) streamError = error
running.value = false
},
)
await stream.start()
Expand All @@ -88,16 +90,14 @@ async function execute(extra: Record<string, unknown> = {}) {
}
} finally {
stream = undefined
running.value = false

}
}

// 表单节点沿用同一条执行记录续跑,复用现有回复组件的 sendMessage 协议。
provide('sendMessage', (options: ResumeParameters) => {
if (running.value || !visible.value) return
const block = blocks.value.find((entry) => entry.id === options.chunkId)
if (block) Object.assign(block, { is_submit: true, form_data: options.formData })
return execute({ ...options.formData, position: options.position })
return execute({ 'form_data':options.formData,'chat_record_id':options.chatRecordId, 'chunk_id':options.chunkId,position: options.position })
})

function open(parameters: Record<string, unknown>) {
Expand Down Expand Up @@ -136,21 +136,21 @@ defineExpose({ open, close })
<h4 class="mk-title-decoration mb-4 mt-4">回复内容</h4>

<!-- // TODO: 回复内容 -->
<el-card>
<ContentList :content-list="blocks" />
<div v-if="running">回答中...</div>
</el-card>

<template v-if="record">
<h4 class="mk-title-decoration my-4">输出参数</h4>
<el-alert
:title="record.state === 'SUCCESS' ? '运行成功' : '运行失败'"
:type="record.state === 'SUCCESS' ? 'success' : 'error'"
:closable="false"
show-icon
class="mb-4"
/>
<el-alert :title="record.state === 'SUCCESS' ? '运行成功' : '运行失败'"
:type="record.state === 'SUCCESS' ? 'success' : 'error'" :closable="false" show-icon class="mb-4" />
<pre class="mk-gray-card whitespace-pre-wrap break-all">{{ output }}</pre>
</template>
</el-tab-pane>
<el-tab-pane label="执行详情" name="details">
<!-- TODO 执行详情 -->
<ExecutionDetailContent :detail="executionDetails" :workflow-mode="WorkflowMode.Tool"></ExecutionDetailContent>
</el-tab-pane>
</el-tabs>
</MkDrawer>
Expand Down
Loading