Skip to content

Commit 293a0e1

Browse files
HenryHengZJHenryHengZJ
committed
add reasoning capability to Agent and LLM
1 parent 7c4a884 commit 293a0e1

23 files changed

Lines changed: 725 additions & 53 deletions

‎packages/components/nodes/agentflow/Agent/Agent.ts‎

Lines changed: 238 additions & 25 deletions
Large diffs are not rendered by default.

‎packages/components/nodes/agentflow/LLM/LLM.ts‎

Lines changed: 90 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -457,8 +457,6 @@ class LLM_Agentflow implements INode {
457457
*/
458458
await addImageArtifactsToMessages(messages, options)
459459

460-
console.log('messages', messages[0])
461-
462460
// Configure structured output if specified
463461
const isStructuredOutput = _llmStructuredOutput && Array.isArray(_llmStructuredOutput) && _llmStructuredOutput.length > 0
464462
if (isStructuredOutput) {
@@ -484,7 +482,15 @@ class LLM_Agentflow implements INode {
484482
* Invoke LLM
485483
*/
486484
if (isStreamable) {
487-
response = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, abortController)
485+
response = await this.handleStreamingResponse(
486+
sseStreamer,
487+
llmNodeInstance,
488+
messages,
489+
chatId,
490+
abortController,
491+
isStructuredOutput,
492+
isLastNode
493+
)
488494
} else {
489495
response = await llmNodeInstance.invoke(messages, { signal: abortController?.signal })
490496

@@ -502,7 +508,6 @@ class LLM_Agentflow implements INode {
502508
sseStreamer.streamTokenEvent(chatId, finalResponse)
503509
}
504510
}
505-
console.log('response.contentBlocks', response.contentBlocks)
506511

507512
// Calculate execution time
508513
const endTime = Date.now()
@@ -555,6 +560,31 @@ class LLM_Agentflow implements INode {
555560
}
556561
}
557562

563+
// Extract reason content from response (reasoning_content/reasoning_duration or contentBlocks)
564+
let reasonContent = (response.additional_kwargs?.reasoning_content as string) || ''
565+
let thinkingDuration: number | undefined =
566+
typeof response.additional_kwargs?.reasoning_duration === 'number'
567+
? response.additional_kwargs.reasoning_duration
568+
: undefined
569+
if (!reasonContent && response.contentBlocks?.length && isLastNode && sseStreamer && !isStructuredOutput) {
570+
for (const block of response.contentBlocks) {
571+
if (block.type === 'reasoning' && (block as { reasoning?: string }).reasoning) {
572+
reasonContent += (block as { reasoning: string }).reasoning
573+
}
574+
if ((block as any).type === 'thinking' && (block as any).thinking) {
575+
reasonContent += (block as any).thinking
576+
}
577+
}
578+
if (reasonContent) {
579+
sseStreamer.streamThinkingEvent(chatId, reasonContent)
580+
const reasoningTokens = response.usage_metadata?.output_token_details?.reasoning || 0
581+
thinkingDuration = reasoningTokens > 0 ? Math.round(reasoningTokens / 50) : 2
582+
sseStreamer.streamThinkingEvent(chatId, '', thinkingDuration)
583+
}
584+
}
585+
const reasonContentObj =
586+
reasonContent !== undefined && reasonContent !== '' ? { thinking: reasonContent, thinkingDuration } : undefined
587+
558588
// Prepare final response and output object
559589
let finalResponse = ''
560590
if (response.content && Array.isArray(response.content)) {
@@ -575,7 +605,8 @@ class LLM_Agentflow implements INode {
575605
timeDelta,
576606
isStructuredOutput,
577607
artifacts,
578-
fileAnnotations
608+
fileAnnotations,
609+
reasonContentObj
579610
)
580611

581612
// End analytics tracking
@@ -839,16 +870,43 @@ class LLM_Agentflow implements INode {
839870
llmNodeInstance: BaseChatModel,
840871
messages: BaseMessageLike[],
841872
chatId: string,
842-
abortController: AbortController
873+
abortController: AbortController,
874+
isStructuredOutput: boolean = false,
875+
isLastNode: boolean = false
843876
): Promise<AIMessageChunk> {
844877
let response = new AIMessageChunk('')
878+
let reasonContent = ''
879+
let thinkingDuration: number | undefined
880+
let thinkingStartTime: number | null = null
881+
let wasThinking = false
882+
let sentLastThinkingEvent = false
845883

846884
try {
847885
for await (const chunk of await llmNodeInstance.stream(messages, { signal: abortController?.signal })) {
848-
if (sseStreamer) {
886+
if (sseStreamer && !isStructuredOutput) {
849887
let content = ''
850888

851-
console.log('chunk.contentBlocks', chunk.contentBlocks)
889+
if (chunk.contentBlocks?.length) {
890+
for (const block of chunk.contentBlocks) {
891+
if (isLastNode) {
892+
// As soon as we see the first non-reasoning block, send last thinking event with duration (only when isLastNode)
893+
if (block.type !== 'reasoning' && wasThinking && !sentLastThinkingEvent && thinkingStartTime != null) {
894+
thinkingDuration = Math.round((Date.now() - thinkingStartTime) / 1000)
895+
sseStreamer.streamThinkingEvent(chatId, '', thinkingDuration)
896+
sentLastThinkingEvent = true
897+
}
898+
if (block.type === 'reasoning' && (block as { reasoning?: string }).reasoning) {
899+
if (!thinkingStartTime) {
900+
thinkingStartTime = Date.now()
901+
}
902+
wasThinking = true
903+
const reasoningContent = (block as { reasoning: string }).reasoning
904+
sseStreamer.streamThinkingEvent(chatId, reasoningContent)
905+
reasonContent += reasoningContent
906+
}
907+
}
908+
}
909+
}
852910

853911
if (typeof chunk === 'string') {
854912
content = chunk
@@ -868,10 +926,26 @@ class LLM_Agentflow implements INode {
868926
console.error('Error during streaming:', error)
869927
throw error
870928
}
929+
930+
// Only convert to string if all content items are text (no inlineData or other special types)
871931
if (Array.isArray(response.content) && response.content.length > 0) {
872-
const responseContents = response.content as ContentBlock.Text[]
873-
response.content = responseContents.map((item) => item.text).join('')
932+
const hasNonTextContent = response.content.some(
933+
(item: any) => item.type === 'inlineData' || item.type === 'executableCode' || item.type === 'codeExecutionResult'
934+
)
935+
if (!hasNonTextContent) {
936+
const responseContents = response.content as ContentBlock.Text[]
937+
response.content = responseContents.map((item) => item.text).join('')
938+
}
874939
}
940+
941+
if (reasonContent.length > 0) {
942+
response.additional_kwargs = {
943+
...response.additional_kwargs,
944+
reasoning_content: reasonContent,
945+
reasoning_duration: thinkingDuration
946+
}
947+
}
948+
875949
return response
876950
}
877951

@@ -886,7 +960,8 @@ class LLM_Agentflow implements INode {
886960
timeDelta: number,
887961
isStructuredOutput: boolean,
888962
artifacts: any[] = [],
889-
fileAnnotations: any[] = []
963+
fileAnnotations: any[] = [],
964+
reasonContent?: { thinking: string; thinkingDuration?: number }
890965
): any {
891966
const output: any = {
892967
content: finalResponse,
@@ -926,6 +1001,10 @@ class LLM_Agentflow implements INode {
9261001
output.fileAnnotations = fileAnnotations
9271002
}
9281003

1004+
if (reasonContent) {
1005+
output.reasonContent = reasonContent
1006+
}
1007+
9291008
return output
9301009
}
9311010

‎packages/components/nodes/chatmodels/Deepseek/Deepseek.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,16 @@ class Deepseek_ChatModels implements INode {
110110
description: 'List of stop words to use when generating. Use comma to separate multiple stop words.',
111111
additionalParams: true
112112
},
113+
{
114+
label: 'Thinking',
115+
name: 'thinking',
116+
type: 'boolean',
117+
default: false,
118+
optional: true,
119+
additionalParams: true,
120+
description:
121+
'Enable deep thinking mode for complex reasoning tasks. When enabled, the model will use extended thinking before responding.'
122+
},
113123
{
114124
label: 'Base Options',
115125
name: 'baseOptions',
@@ -138,6 +148,7 @@ class Deepseek_ChatModels implements INode {
138148
const timeout = nodeData.inputs?.timeout as string
139149
const stopSequence = nodeData.inputs?.stopSequence as string
140150
const streaming = nodeData.inputs?.streaming as boolean
151+
const thinking = nodeData.inputs?.thinking as boolean
141152
const baseOptions = nodeData.inputs?.baseOptions
142153

143154
if (nodeData.inputs?.credentialId) {
@@ -166,6 +177,12 @@ class Deepseek_ChatModels implements INode {
166177
const stopSequenceArray = stopSequence.split(',').map((item) => item.trim())
167178
obj.stop = stopSequenceArray
168179
}
180+
if (thinking) {
181+
obj.modelKwargs = {
182+
...obj.modelKwargs,
183+
thinking: { type: 'enabled' }
184+
}
185+
}
169186

170187
if (baseOptions) {
171188
try {

‎packages/components/src/Interface.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -426,6 +426,7 @@ export * from './Interface.Evaluation'
426426
export interface IServerSideEventStreamer {
427427
streamStartEvent(chatId: string, data: any): void
428428
streamTokenEvent(chatId: string, data: string): void
429+
streamThinkingEvent(chatId: string, data: string, duration?: number): void
429430
streamCustomEvent(chatId: string, eventType: string, data: any): void
430431
streamSourceDocumentsEvent(chatId: string, data: any): void
431432
streamUsedToolsEvent(chatId: string, data: any): void

‎packages/server/src/Interface.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,7 @@ export interface IChatMessage {
8383
usedTools?: string
8484
fileAnnotations?: string
8585
agentReasoning?: string
86+
reasonContent?: string
8687
fileUploads?: string
8788
artifacts?: string
8889
chatType: string

‎packages/server/src/controllers/chat-messages/index.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -336,6 +336,9 @@ const parseAPIResponse = (apiResponse: ChatMessage | ChatMessage[]): ChatMessage
336336
if (parsedResponse.agentReasoning) {
337337
parsedResponse.agentReasoning = JSON.parse(parsedResponse.agentReasoning)
338338
}
339+
if (parsedResponse.reasonContent) {
340+
parsedResponse.reasonContent = JSON.parse(parsedResponse.reasonContent)
341+
}
339342
if (parsedResponse.fileUploads) {
340343
parsedResponse.fileUploads = JSON.parse(parsedResponse.fileUploads)
341344
}

‎packages/server/src/database/entities/ChatMessage.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,9 @@ export class ChatMessage implements IChatMessage {
3737
@Column({ nullable: true, type: 'text' })
3838
agentReasoning?: string
3939

40+
@Column({ nullable: true, type: 'text' })
41+
reasonContent?: string
42+
4043
@Column({ nullable: true, type: 'text' })
4144
fileUploads?: string
4245

‎packages/server/src/database/migrations/mariadb/1764759496768-AddReasonContentToChatMessage.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
import { MigrationInterface, QueryRunner } from 'typeorm'
2+
3+
export class AddReasonContentToChatMessage1764759496768 implements MigrationInterface {
4+
public async up(queryRunner: QueryRunner): Promise<void> {
5+
const columnExists = await queryRunner.hasColumn('chat_message', 'reasonContent')
6+
if (!columnExists) queryRunner.query(`ALTER TABLE \`chat_message\` ADD COLUMN \`reasonContent\` LONGTEXT;`)
7+
}
8+
9+
public async down(queryRunner: QueryRunner): Promise<void> {
10+
await queryRunner.query(`ALTER TABLE \`chat_message\` DROP COLUMN \`reasonContent\`;`)
11+
}
12+
}

‎packages/server/src/database/migrations/mariadb/index.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ import { AddTextToSpeechToChatFlow1759419231100 } from './1759419231100-AddTextT
4242
import { AddChatFlowNameIndex1759424809984 } from './1759424809984-AddChatFlowNameIndex'
4343
import { FixDocumentStoreFileChunkLongText1765000000000 } from './1765000000000-FixDocumentStoreFileChunkLongText'
4444
import { AddApiKeyPermission1765360298674 } from './1765360298674-AddApiKeyPermission'
45-
45+
import { AddReasonContentToChatMessage1764759496768 } from './1764759496768-AddReasonContentToChatMessage'
4646
import { AddAuthTables1720230151482 } from '../../../enterprise/database/migrations/mariadb/1720230151482-AddAuthTables'
4747
import { AddWorkspace1725437498242 } from '../../../enterprise/database/migrations/mariadb/1725437498242-AddWorkspace'
4848
import { AddWorkspaceShared1726654922034 } from '../../../enterprise/database/migrations/mariadb/1726654922034-AddWorkspaceShared'
@@ -110,5 +110,6 @@ export const mariadbMigrations = [
110110
AddTextToSpeechToChatFlow1759419231100,
111111
AddChatFlowNameIndex1759424809984,
112112
FixDocumentStoreFileChunkLongText1765000000000,
113-
AddApiKeyPermission1765360298674
113+
AddApiKeyPermission1765360298674,
114+
AddReasonContentToChatMessage1764759496768
114115
]

‎packages/server/src/database/migrations/mysql/1764759496768-AddReasonContentToChatMessage.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
import { MigrationInterface, QueryRunner } from 'typeorm'
2+
3+
export class AddReasonContentToChatMessage1764759496768 implements MigrationInterface {
4+
public async up(queryRunner: QueryRunner): Promise<void> {
5+
const columnExists = await queryRunner.hasColumn('chat_message', 'reasonContent')
6+
if (!columnExists) queryRunner.query(`ALTER TABLE \`chat_message\` ADD COLUMN \`reasonContent\` LONGTEXT;`)
7+
}
8+
9+
public async down(queryRunner: QueryRunner): Promise<void> {
10+
await queryRunner.query(`ALTER TABLE \`chat_message\` DROP COLUMN \`reasonContent\`;`)
11+
}
12+
}

0 commit comments

Comments
 (0)