mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
fix(AI Agent Node): Replay Anthropic thinking turns with their tool arguments (#41241)
This commit is contained in:
parent
1405d4a351
commit
ea98bd0600
|
|
@ -1,4 +1,5 @@
|
|||
import { AIMessage } from '@langchain/core/messages';
|
||||
import omit from 'lodash/omit';
|
||||
import { nodeNameToToolName } from 'n8n-workflow';
|
||||
import type { EngineResponse, EngineResult, IDataObject } from 'n8n-workflow';
|
||||
|
||||
|
|
@ -74,9 +75,18 @@ function extractProviderMetadata(metadata?: RequestResponseMetadata): ProviderMe
|
|||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* A tool call to replay as an Anthropic tool_use content block.
|
||||
*/
|
||||
interface AnthropicToolCall {
|
||||
toolInput: IDataObject;
|
||||
toolId: string;
|
||||
toolName: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds Anthropic-specific content blocks for thinking mode.
|
||||
* Creates an array with thinking block followed by tool_use block.
|
||||
* Creates an array with the thinking block followed by one tool_use block per tool call.
|
||||
*
|
||||
* IMPORTANT: The thinking block must come before tool_use in the message.
|
||||
* When content is an array, LangChain ignores tool_calls field for Anthropic,
|
||||
|
|
@ -85,18 +95,14 @@ function extractProviderMetadata(metadata?: RequestResponseMetadata): ProviderMe
|
|||
* @param thinkingContent - The thinking content from Anthropic
|
||||
* @param thinkingType - Type of thinking block (thinking or redacted_thinking)
|
||||
* @param thinkingSignature - Optional signature for thinking block
|
||||
* @param toolInput - The tool input data
|
||||
* @param toolId - The tool call ID
|
||||
* @param toolName - The tool name
|
||||
* @param toolCalls - The tool calls that the model made in this turn
|
||||
* @returns Array of content blocks with thinking and tool_use
|
||||
*/
|
||||
function buildAnthropicContentBlocks(
|
||||
thinkingContent: string,
|
||||
thinkingType: 'thinking' | 'redacted_thinking',
|
||||
thinkingSignature: string | undefined,
|
||||
toolInput: IDataObject,
|
||||
toolId: string,
|
||||
toolName: string,
|
||||
toolCalls: AnthropicToolCall[],
|
||||
): Array<ThinkingContentBlock | RedactedThinkingContentBlock | ToolUseContentBlock> {
|
||||
// Create thinking block with correct field names for Anthropic API
|
||||
const thinkingBlock: ThinkingContentBlock | RedactedThinkingContentBlock =
|
||||
|
|
@ -111,19 +117,16 @@ function buildAnthropicContentBlocks(
|
|||
data: thinkingContent,
|
||||
};
|
||||
|
||||
// Create tool_use block (required for Anthropic when using structured content)
|
||||
const toolInputData = toolInput.input;
|
||||
const toolUseBlock: ToolUseContentBlock = {
|
||||
// toolInput is the tool call arguments. `tool` is a toolkit routing field added by
|
||||
// createEngineRequests. The model did not send it, so omit it from the replayed input.
|
||||
const toolUseBlocks: ToolUseContentBlock[] = toolCalls.map(({ toolInput, toolId, toolName }) => ({
|
||||
type: 'tool_use',
|
||||
id: toolId,
|
||||
name: toolName,
|
||||
input:
|
||||
toolInputData && typeof toolInputData === 'object'
|
||||
? (toolInputData as Record<string, unknown>)
|
||||
: {},
|
||||
};
|
||||
input: omit(toolInput, 'tool'),
|
||||
}));
|
||||
|
||||
return [thinkingBlock, toolUseBlock];
|
||||
return [thinkingBlock, ...toolUseBlocks];
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -149,14 +152,9 @@ function buildMessageContent(
|
|||
// Anthropic thinking mode: build content blocks
|
||||
// Empty thinking text is valid because the signature carries the reasoning.
|
||||
if (thinkingContent !== undefined && thinkingType) {
|
||||
return buildAnthropicContentBlocks(
|
||||
thinkingContent,
|
||||
thinkingType,
|
||||
thinkingSignature,
|
||||
toolInput,
|
||||
toolId,
|
||||
toolName,
|
||||
);
|
||||
return buildAnthropicContentBlocks(thinkingContent, thinkingType, thinkingSignature, [
|
||||
{ toolInput, toolId, toolName },
|
||||
]);
|
||||
}
|
||||
|
||||
return null;
|
||||
|
|
@ -306,6 +304,69 @@ function buildSharedGeminiAIMessage(
|
|||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Anthropic thinking metadata shared by every tool call in a parallel batch.
|
||||
*/
|
||||
interface SharedAnthropicThinking {
|
||||
thinkingContent: string;
|
||||
thinkingType: 'thinking' | 'redacted_thinking';
|
||||
thinkingSignature: string | undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the thinking block when every tool call in the batch carries the same one.
|
||||
* Replay that batch as one assistant message. A second copy of the signature follows
|
||||
* content that was not present when Anthropic signed the block, and the next request fails.
|
||||
*/
|
||||
function getSharedAnthropicThinking(
|
||||
batchTools: ProcessedToolResponse[],
|
||||
): SharedAnthropicThinking | undefined {
|
||||
if (batchTools.length < 2) return undefined;
|
||||
|
||||
const first = batchTools[0].providerMetadata;
|
||||
if (first.thinkingContent === undefined || !first.thinkingType) return undefined;
|
||||
|
||||
const shared: SharedAnthropicThinking = {
|
||||
thinkingContent: first.thinkingContent,
|
||||
thinkingType: first.thinkingType,
|
||||
thinkingSignature: first.thinkingSignature,
|
||||
};
|
||||
|
||||
const matches = batchTools.every((tool) => {
|
||||
const metadata = tool.providerMetadata;
|
||||
return (
|
||||
metadata.thinkingContent === shared.thinkingContent &&
|
||||
metadata.thinkingType === shared.thinkingType &&
|
||||
metadata.thinkingSignature === shared.thinkingSignature
|
||||
);
|
||||
});
|
||||
|
||||
return matches ? shared : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds one AIMessage for parallel Anthropic tool calls that share a thinking block.
|
||||
* The content is the original turn: the thinking block, then one tool_use block per call.
|
||||
* LangChain ignores tool_calls when content is an array, so this message sets content only.
|
||||
*/
|
||||
function buildSharedAnthropicAIMessage(
|
||||
processedTools: ProcessedToolResponse[],
|
||||
sharedThinking: SharedAnthropicThinking,
|
||||
): AIMessage {
|
||||
const content = buildAnthropicContentBlocks(
|
||||
sharedThinking.thinkingContent,
|
||||
sharedThinking.thinkingType,
|
||||
sharedThinking.thinkingSignature,
|
||||
processedTools.map(({ toolInput, toolId, toolName }) => ({
|
||||
toolInput,
|
||||
toolId,
|
||||
toolName,
|
||||
})),
|
||||
);
|
||||
|
||||
return new AIMessage({ content });
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the observation string from tool result data.
|
||||
*/
|
||||
|
|
@ -331,9 +392,9 @@ function buildObservation(toolData: {
|
|||
* Rebuilds the agent steps from previous tool call responses.
|
||||
* This is used to continue agent execution after tool calls have been made.
|
||||
*
|
||||
* For parallel tool calls with Gemini thought signatures, all function calls are grouped
|
||||
* into a single AIMessage to match the original model response structure. This is required
|
||||
* because the thought_signature is cryptographically tied to the combined parallel call turn.
|
||||
* For parallel tool calls with Gemini thought signatures or an Anthropic thinking block, all
|
||||
* tool calls are grouped into a single AIMessage to match the original model response
|
||||
* structure. This is required because the signature is tied to the combined parallel call turn.
|
||||
*
|
||||
* This is a generalized version that can be used across different agent types
|
||||
* (Tools Agent, OpenAI Functions Agent, etc.).
|
||||
|
|
@ -389,10 +450,17 @@ export function buildSteps(
|
|||
const sharedThoughtSignature = batchTools.find((bt) => bt.providerMetadata.thoughtSignature)
|
||||
?.providerMetadata.thoughtSignature;
|
||||
|
||||
const sharedAIMessage =
|
||||
sharedThoughtSignature && batchTools.length > 1
|
||||
? buildSharedGeminiAIMessage(batchTools, sharedThoughtSignature)
|
||||
: undefined;
|
||||
// Anthropic binds the thinking signature to the whole turn. Parallel tool calls that
|
||||
// share one thinking block must be replayed as a single assistant message.
|
||||
let sharedAIMessage: AIMessage | undefined;
|
||||
if (sharedThoughtSignature && batchTools.length > 1) {
|
||||
sharedAIMessage = buildSharedGeminiAIMessage(batchTools, sharedThoughtSignature);
|
||||
} else {
|
||||
const sharedAnthropicThinking = getSharedAnthropicThinking(batchTools);
|
||||
if (sharedAnthropicThinking) {
|
||||
sharedAIMessage = buildSharedAnthropicAIMessage(batchTools, sharedAnthropicThinking);
|
||||
}
|
||||
}
|
||||
|
||||
// Second pass: build steps
|
||||
for (let i = 0; i < batchTools.length; i++) {
|
||||
|
|
@ -403,7 +471,7 @@ export function buildSteps(
|
|||
// Exclude metadata fields (id, log, type) from the tool input forwarded to the result
|
||||
const { id, log, type, ...toolInputForResult } = toolInput;
|
||||
|
||||
// Parallel Gemini tool calls: first step gets the shared AIMessage,
|
||||
// Parallel Gemini or Anthropic tool calls: first step gets the shared AIMessage,
|
||||
// subsequent steps get empty messageLog. LangChain's formatToToolMessages
|
||||
// will produce: [SharedAIMessage, ToolMsg_1, ToolMsg_2, ...]
|
||||
const messageLog = sharedAIMessage
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import fc from 'fast-check';
|
||||
import type { EngineResponse } from 'n8n-workflow';
|
||||
import type { EngineResponse, IDataObject } from 'n8n-workflow';
|
||||
import { NodeConnectionTypes } from 'n8n-workflow';
|
||||
|
||||
import { buildSteps } from '../buildSteps';
|
||||
|
|
@ -920,10 +920,7 @@ describe('buildSteps', () => {
|
|||
action: {
|
||||
actionType: 'ExecutionNodeAction',
|
||||
nodeName: 'Calculator',
|
||||
input: {
|
||||
id: 'call_omitted_1',
|
||||
input: { expression: '2+2' },
|
||||
},
|
||||
input: { expression: '2+2' },
|
||||
type: NodeConnectionTypes.AiTool,
|
||||
id: 'call_omitted_1',
|
||||
metadata: {
|
||||
|
|
@ -1943,4 +1940,277 @@ describe('buildSteps', () => {
|
|||
).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('Anthropic parallel tool calls with a shared thinking block', () => {
|
||||
const buildToolResponse = (
|
||||
nodeName: string,
|
||||
id: string,
|
||||
input: IDataObject,
|
||||
anthropic: RequestResponseMetadata['anthropic'] | undefined,
|
||||
result: IDataObject,
|
||||
): NonNullable<EngineResponse<RequestResponseMetadata>['actionResponses']>[number] => ({
|
||||
action: {
|
||||
actionType: 'ExecutionNodeAction',
|
||||
nodeName,
|
||||
input,
|
||||
type: NodeConnectionTypes.AiTool,
|
||||
id,
|
||||
metadata: {
|
||||
itemIndex: 0,
|
||||
...(anthropic && { anthropic }),
|
||||
},
|
||||
},
|
||||
data: {
|
||||
data: {
|
||||
ai_tool: [[{ json: result }]],
|
||||
},
|
||||
executionTime: 0,
|
||||
startTime: 0,
|
||||
executionIndex: 0,
|
||||
source: [],
|
||||
},
|
||||
});
|
||||
|
||||
it('should group parallel tool calls that share one thinking block', () => {
|
||||
// createEngineRequests copies one thinking block onto every tool call of the turn.
|
||||
const sharedThinking = {
|
||||
thinkingContent: '',
|
||||
thinkingType: 'thinking' as const,
|
||||
thinkingSignature: 'turn_signature_abc',
|
||||
};
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
buildToolResponse('Calculator', 'call_1', { expression: '2+2' }, sharedThinking, {
|
||||
result: '4',
|
||||
}),
|
||||
buildToolResponse('Weather', 'call_2', { location: 'NYC' }, sharedThinking, {
|
||||
temp: '72F',
|
||||
}),
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
expect(result).toHaveLength(2);
|
||||
|
||||
expect(result[0].action.messageLog).toHaveLength(1);
|
||||
const sharedMessage = result[0].action.messageLog![0];
|
||||
expect(sharedMessage.content).toEqual([
|
||||
{ type: 'thinking', thinking: '', signature: 'turn_signature_abc' },
|
||||
{ type: 'tool_use', id: 'call_1', name: 'Calculator', input: { expression: '2+2' } },
|
||||
{ type: 'tool_use', id: 'call_2', name: 'Weather', input: { location: 'NYC' } },
|
||||
]);
|
||||
// Array content is the Anthropic payload. tool_calls stays empty.
|
||||
expect(sharedMessage.tool_calls).toEqual([]);
|
||||
|
||||
expect(result[1].action.messageLog).toEqual([]);
|
||||
|
||||
expect(result[0].action.toolCallId).toBe('call_1');
|
||||
expect(result[1].action.toolCallId).toBe('call_2');
|
||||
expect(result[0].observation).toBe(JSON.stringify([{ result: '4' }]));
|
||||
expect(result[1].observation).toBe(JSON.stringify([{ temp: '72F' }]));
|
||||
});
|
||||
|
||||
it('should group parallel tool calls that share one redacted_thinking block', () => {
|
||||
const sharedThinking = {
|
||||
thinkingContent: 'redacted_payload',
|
||||
thinkingType: 'redacted_thinking' as const,
|
||||
};
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
buildToolResponse('Calculator', 'call_1', { expression: '2+2' }, sharedThinking, {
|
||||
result: '4',
|
||||
}),
|
||||
buildToolResponse('Weather', 'call_2', { location: 'NYC' }, sharedThinking, {
|
||||
temp: '72F',
|
||||
}),
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
expect(result).toHaveLength(2);
|
||||
expect(result[0].action.messageLog![0].content).toEqual([
|
||||
{ type: 'redacted_thinking', data: 'redacted_payload' },
|
||||
{ type: 'tool_use', id: 'call_1', name: 'Calculator', input: { expression: '2+2' } },
|
||||
{ type: 'tool_use', id: 'call_2', name: 'Weather', input: { location: 'NYC' } },
|
||||
]);
|
||||
expect(result[1].action.messageLog).toEqual([]);
|
||||
});
|
||||
|
||||
it('should not group parallel tool calls whose thinking signatures differ', () => {
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
buildToolResponse(
|
||||
'Calculator',
|
||||
'call_1',
|
||||
{ expression: '2+2' },
|
||||
{ thinkingContent: '', thinkingType: 'thinking', thinkingSignature: 'sig_1' },
|
||||
{ result: '4' },
|
||||
),
|
||||
buildToolResponse(
|
||||
'Weather',
|
||||
'call_2',
|
||||
{ location: 'NYC' },
|
||||
{ thinkingContent: '', thinkingType: 'thinking', thinkingSignature: 'sig_2' },
|
||||
{ temp: '72F' },
|
||||
),
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
expect(result).toHaveLength(2);
|
||||
expect(result[0].action.messageLog).toHaveLength(1);
|
||||
expect(result[0].action.messageLog![0].content).toEqual([
|
||||
{ type: 'thinking', thinking: '', signature: 'sig_1' },
|
||||
{ type: 'tool_use', id: 'call_1', name: 'Calculator', input: { expression: '2+2' } },
|
||||
]);
|
||||
expect(result[1].action.messageLog).toHaveLength(1);
|
||||
expect(result[1].action.messageLog![0].content).toEqual([
|
||||
{ type: 'thinking', thinking: '', signature: 'sig_2' },
|
||||
{ type: 'tool_use', id: 'call_2', name: 'Weather', input: { location: 'NYC' } },
|
||||
]);
|
||||
});
|
||||
|
||||
it('should not group when only some parallel tool calls carry a thinking block', () => {
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
buildToolResponse(
|
||||
'Calculator',
|
||||
'call_1',
|
||||
{ expression: '2+2' },
|
||||
{ thinkingContent: '', thinkingType: 'thinking', thinkingSignature: 'sig_1' },
|
||||
{ result: '4' },
|
||||
),
|
||||
buildToolResponse('Weather', 'call_2', { location: 'NYC' }, undefined, { temp: '72F' }),
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
expect(result).toHaveLength(2);
|
||||
expect(result[0].action.messageLog![0].content).toHaveLength(2);
|
||||
expect(result[1].action.messageLog).toHaveLength(1);
|
||||
expect(result[1].action.messageLog![0].content).toEqual([]);
|
||||
expect(result[1].action.messageLog![0].tool_calls).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('should keep a single tool call with a thinking block as an individual AIMessage', () => {
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
buildToolResponse(
|
||||
'Calculator',
|
||||
'call_1',
|
||||
{ expression: '2+2' },
|
||||
{ thinkingContent: '', thinkingType: 'thinking', thinkingSignature: 'sig_1' },
|
||||
{ result: '4' },
|
||||
),
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
expect(result).toHaveLength(1);
|
||||
expect(result[0].action.messageLog![0].content).toEqual([
|
||||
{ type: 'thinking', thinking: '', signature: 'sig_1' },
|
||||
{ type: 'tool_use', id: 'call_1', name: 'Calculator', input: { expression: '2+2' } },
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Anthropic thinking block tool arguments', () => {
|
||||
it('should replay the tool call arguments in the tool_use block', () => {
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
{
|
||||
action: {
|
||||
actionType: 'ExecutionNodeAction',
|
||||
nodeName: 'Research Agent',
|
||||
input: { topic: 'solar panels', depth: 2 },
|
||||
type: NodeConnectionTypes.AiTool,
|
||||
id: 'call_args_1',
|
||||
metadata: {
|
||||
itemIndex: 0,
|
||||
anthropic: {
|
||||
thinkingContent: 'I should research the topic.',
|
||||
thinkingType: 'thinking',
|
||||
thinkingSignature: 'sig_args',
|
||||
},
|
||||
},
|
||||
},
|
||||
data: {
|
||||
data: {
|
||||
ai_tool: [[{ json: { result: 'done' } }]],
|
||||
},
|
||||
executionTime: 0,
|
||||
startTime: 0,
|
||||
executionIndex: 0,
|
||||
source: [],
|
||||
},
|
||||
},
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
const message = result[0].action.messageLog![0];
|
||||
expect(message.content[1]).toEqual({
|
||||
type: 'tool_use',
|
||||
id: 'call_args_1',
|
||||
name: 'Research_Agent',
|
||||
input: { topic: 'solar panels', depth: 2 },
|
||||
});
|
||||
});
|
||||
|
||||
it('should omit the toolkit routing field from the replayed tool arguments', () => {
|
||||
const response: EngineResponse<RequestResponseMetadata> = {
|
||||
actionResponses: [
|
||||
{
|
||||
action: {
|
||||
actionType: 'ExecutionNodeAction',
|
||||
nodeName: 'MCP Client',
|
||||
input: { query: 'open tickets', tool: 'list_tickets' },
|
||||
type: NodeConnectionTypes.AiTool,
|
||||
id: 'call_mcp_1',
|
||||
metadata: {
|
||||
itemIndex: 0,
|
||||
anthropic: {
|
||||
thinkingContent: '',
|
||||
thinkingType: 'thinking',
|
||||
thinkingSignature: 'sig_mcp',
|
||||
},
|
||||
},
|
||||
},
|
||||
data: {
|
||||
data: {
|
||||
ai_tool: [[{ json: { tickets: [] } }]],
|
||||
},
|
||||
executionTime: 0,
|
||||
startTime: 0,
|
||||
executionIndex: 0,
|
||||
source: [],
|
||||
},
|
||||
},
|
||||
],
|
||||
metadata: {},
|
||||
};
|
||||
|
||||
const result = buildSteps(response, itemIndex);
|
||||
|
||||
expect(result[0].action.tool).toBe('list_tickets');
|
||||
expect(result[0].action.messageLog![0].content[1]).toEqual({
|
||||
type: 'tool_use',
|
||||
id: 'call_mcp_1',
|
||||
name: 'list_tickets',
|
||||
input: { query: 'open tickets' },
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user