diff --git a/backend/windmill-worker/src/ai/types.rs b/backend/windmill-worker/src/ai/types.rs index 724ea6bbf6..97641ba175 100644 --- a/backend/windmill-worker/src/ai/types.rs +++ b/backend/windmill-worker/src/ai/types.rs @@ -106,19 +106,73 @@ impl Default for OutputType { } } -#[derive(Deserialize, Debug)] +#[derive(Deserialize, Debug, Clone)] +#[serde(tag = "kind", rename_all = "lowercase")] +pub enum Memory { + Auto { + #[serde(default)] + context_length: usize, + }, + Manual { + messages: Vec, + }, +} + +#[derive(Deserialize)] +struct AIAgentArgsRaw { + provider: ProviderWithResource, + system_prompt: Option, + user_message: Option, + temperature: Option, + max_completion_tokens: Option, + output_schema: Option, + output_type: Option, + user_images: Option>, + streaming: Option, + max_iterations: Option, + memory: Option, + // Legacy field for backward compatibility + messages_context_length: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(from = "AIAgentArgsRaw")] pub struct AIAgentArgs { pub provider: ProviderWithResource, pub system_prompt: Option, - pub user_message: String, + pub user_message: Option, pub temperature: Option, pub max_completion_tokens: Option, pub output_schema: Option, pub output_type: Option, pub user_images: Option>, pub streaming: Option, - pub messages_context_length: Option, pub max_iterations: Option, + pub memory: Option, +} + +impl From for AIAgentArgs { + fn from(raw: AIAgentArgsRaw) -> Self { + // Backward compatibility: if messages_context_length is set, use auto mode + let memory = raw.memory.or_else(|| { + raw.messages_context_length + .map(|context_length| Memory::Auto { context_length }) + }); + + AIAgentArgs { + provider: raw.provider, + system_prompt: raw.system_prompt, + user_message: raw.user_message, + temperature: raw.temperature, + max_completion_tokens: raw.max_completion_tokens, + output_schema: raw.output_schema, + output_type: raw.output_type, + user_images: raw.user_images, + streaming: raw.streaming, + max_iterations: raw.max_iterations, + memory, + } + } } #[derive(Deserialize, Debug)] diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 00a58e2c41..810fec7063 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -397,38 +397,70 @@ pub async fn run_agent( // Fetch flow context for input transforms context, chat and memory let mut flow_context = get_flow_context(db, job).await; - // Load previous messages from memory for text output mode (only if context length is set) + // Determine if we're using manual messages (which bypasses memory) + let use_manual_messages = matches!(args.memory, Some(Memory::Manual { .. })); + + // Check if user_message is provided and non-empty + let has_user_message = args + .user_message + .as_ref() + .map(|m| !m.is_empty()) + .unwrap_or(false); + + // Validate: at least one of memory with manual messages or user_message must be provided + if !use_manual_messages && !has_user_message { + return Err(Error::internal_err( + "Either 'memory' with manual messages or 'user_message' must be provided".to_string(), + )); + } + + // Load messages based on history mode if matches!(output_type, OutputType::Text) { - if let Some(context_length) = args.messages_context_length.filter(|&n| n > 0) { - if let Some(step_id) = job.flow_step_id.as_deref() { - if let Some(memory_id) = flow_context - .flow_status - .as_ref() - .and_then(|fs| fs.memory_id) - { - // Read messages from memory - match read_from_memory(db, &job.workspace_id, memory_id, step_id).await { - Ok(Some(loaded_messages)) => { - // Take the last n messages - let start_idx = loaded_messages.len().saturating_sub(context_length); - let mut messages_to_load = loaded_messages[start_idx..].to_vec(); - let first_non_tool_message_index = - messages_to_load.iter().position(|m| m.role != "tool"); + match &args.memory { + Some(Memory::Manual { messages: manual_messages }) => { + // Use explicitly provided messages (bypass memory) + if !manual_messages.is_empty() { + messages.extend(manual_messages.clone()); + } + } + Some(Memory::Auto { context_length }) if *context_length > 0 => { + // Auto mode: load from memory + if let Some(step_id) = job.flow_step_id.as_deref() { + if let Some(memory_id) = flow_context + .flow_status + .as_ref() + .and_then(|fs| fs.memory_id) + { + // Read messages from memory + match read_from_memory(db, &job.workspace_id, memory_id, step_id).await { + Ok(Some(loaded_messages)) => { + // Take the last n messages + let start_idx = + loaded_messages.len().saturating_sub(*context_length); + let mut messages_to_load = loaded_messages[start_idx..].to_vec(); + let first_non_tool_message_index = + messages_to_load.iter().position(|m| m.role != "tool"); - // Remove the first messages if their role is "tool" to avoid OpenAI API error - if let Some(index) = first_non_tool_message_index { - messages_to_load = messages_to_load[index..].to_vec(); + // Remove the first messages if their role is "tool" to avoid OpenAI API error + if let Some(index) = first_non_tool_message_index { + messages_to_load = messages_to_load[index..].to_vec(); + } + + messages.extend(messages_to_load); + } + Ok(None) => {} + Err(e) => { + tracing::error!( + "Failed to read memory for step {}: {}", + step_id, + e + ); } - - messages.extend(messages_to_load); - } - Ok(None) => {} - Err(e) => { - tracing::error!("Failed to read memory for step {}: {}", step_id, e); } } } } + _ => {} } } @@ -463,22 +495,33 @@ pub async fn run_agent( } }; - // Create user message with optional images - let mut parts = vec![ContentPart::Text { text: args.user_message.clone() }]; - if let Some(images) = &args.user_images { - for image in images.iter() { - if !image.s3.is_empty() { - parts.push(ContentPart::S3Object { s3_object: image.clone() }); - } + // Add user message if provided and non-empty + if let Some(ref user_message) = args.user_message { + if !user_message.is_empty() { + messages.push(OpenAIMessage { + role: "user".to_string(), + content: Some(OpenAIContent::Text(user_message.clone())), + ..Default::default() + }); } } - let user_content = OpenAIContent::Parts(parts); - messages.push(OpenAIMessage { - role: "user".to_string(), - content: Some(user_content), - ..Default::default() - }); + // Add user images if provided + if let Some(ref user_images) = args.user_images { + if !user_images.is_empty() { + let mut parts = vec![]; + for image in user_images.iter() { + if !image.s3.is_empty() { + parts.push(ContentPart::S3Object { s3_object: image.clone() }); + } + } + messages.push(OpenAIMessage { + role: "user".to_string(), + content: Some(OpenAIContent::Parts(parts)), + ..Default::default() + }); + } + } let mut actions = vec![]; let mut content = None; @@ -596,7 +639,7 @@ pub async fn run_agent( output_schema: args.output_schema.as_ref(), output_type, system_prompt: args.system_prompt.as_deref(), - user_message: &args.user_message, + user_message: args.user_message.as_deref().unwrap_or(""), images: args.user_images.as_deref(), }; @@ -882,36 +925,41 @@ pub async fn run_agent( } } - // Persist complete conversation to memory at the end (only if context length is set) + // Persist complete conversation to memory at the end (only if in auto mode with context length) + // Skip memory persistence if using manual messages (bypass memory entirely) // final_messages contains the complete history (old messages + new ones) - if matches!(output_type, OutputType::Text) { - if let Some(context_length) = args.messages_context_length.filter(|&n| n > 0) { - if let Some(step_id) = job.flow_step_id.as_deref() { - // Extract OpenAIMessages from final_messages - let all_messages: Vec = - final_messages.iter().map(|m| m.message.clone()).collect(); + if matches!(output_type, OutputType::Text) && !use_manual_messages { + if let Some(Memory::Auto { context_length }) = &args.memory { + if *context_length > 0 { + if let Some(step_id) = job.flow_step_id.as_deref() { + // Extract OpenAIMessages from final_messages + let all_messages: Vec = + final_messages.iter().map(|m| m.message.clone()).collect(); - if !all_messages.is_empty() { - // Keep only the last n messages - let start_idx = all_messages.len().saturating_sub(context_length); - let messages_to_persist = all_messages[start_idx..].to_vec(); + if !all_messages.is_empty() { + // Keep only the last n messages + let start_idx = all_messages.len().saturating_sub(*context_length); + let messages_to_persist = all_messages[start_idx..].to_vec(); - if let Some(memory_id) = flow_context.flow_status.and_then(|fs| fs.memory_id) { - if let Err(e) = write_to_memory( - db, - &job.workspace_id, - memory_id, - step_id, - &messages_to_persist, - ) - .await + if let Some(memory_id) = + flow_context.flow_status.and_then(|fs| fs.memory_id) { - tracing::error!( - "Failed to persist {} messages to memory for step {}: {}", - messages_to_persist.len(), + if let Err(e) = write_to_memory( + db, + &job.workspace_id, + memory_id, step_id, - e - ); + &messages_to_persist, + ) + .await + { + tracing::error!( + "Failed to persist {} messages to memory for step {}: {}", + messages_to_persist.len(), + step_id, + e + ); + } } } } diff --git a/frontend/src/lib/common.ts b/frontend/src/lib/common.ts index 5f30b6056e..8955be2e4a 100644 --- a/frontend/src/lib/common.ts +++ b/frontend/src/lib/common.ts @@ -31,6 +31,7 @@ export interface SchemaProperty { enum?: string[] resourceType?: string properties?: { [name: string]: SchemaProperty } + required?: string[] } min?: number max?: number @@ -110,8 +111,8 @@ export function modalToSchema(schema: ModalSchemaProperty): SchemaProperty { export type Schema = { $schema: string | undefined type: string - "x-windmill-dyn-select-code"?: string - "x-windmill-dyn-select-lang"?: ScriptLang + 'x-windmill-dyn-select-code'?: string + 'x-windmill-dyn-select-lang'?: ScriptLang properties: { [name: string]: SchemaProperty } order?: string[] required: string[] diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index 8bc2d68cd3..078585cda3 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -266,7 +266,7 @@ } else if (inputCat == 'boolean') { nvalue = false } else if (inputCat == 'list') { - nvalue = [] + nvalue = nullable ? null : [] } } else if (inputCat === 'object') { evalValueToRaw() @@ -1165,6 +1165,15 @@ /> {/if} {/key} + {#if !s3StorageConfigured && obj['x-no-s3-storage-workspace-warning']} + + {/if} + {:else if disabled} {:else} diff --git a/frontend/src/lib/components/JsonEditor.svelte b/frontend/src/lib/components/JsonEditor.svelte index ea6d063f1f..07db17b9c9 100644 --- a/frontend/src/lib/components/JsonEditor.svelte +++ b/frontend/src/lib/components/JsonEditor.svelte @@ -43,10 +43,9 @@ try { if (code == '') { value = undefined - error = '' - return + } else { + value = JSON.parse(code ?? '') } - value = JSON.parse(code ?? '') dispatchIfMounted('changeValue', value) error = '' } catch (e) { diff --git a/frontend/src/lib/components/common/alert/Alert.svelte b/frontend/src/lib/components/common/alert/Alert.svelte index 3bea9efb7d..3a1361b2ff 100644 --- a/frontend/src/lib/components/common/alert/Alert.svelte +++ b/frontend/src/lib/components/common/alert/Alert.svelte @@ -80,7 +80,7 @@ )} style={bgStyle} > -
+
-
+
{ if (key === 'user_message') { accu[key] = { type: 'javascript', expr: 'flow_input.user_message' } - } else if (key === 'messages_context_length') { - accu[key] = { type: 'static', value: 10 } + } else if (key === 'memory') { + accu[key] = { type: 'static', value: { kind: 'auto', context_length: 10 } } } else { accu[key] = { type: 'static', @@ -495,9 +495,9 @@ } // Set messages_context_length to 10 - value.input_transforms['messages_context_length'] = { + value.input_transforms['memory'] = { type: 'static', - value: 10 + value: { kind: 'auto', context_length: 10 } } sendUserToast( diff --git a/frontend/src/lib/components/flows/flowInfers.ts b/frontend/src/lib/components/flows/flowInfers.ts index 9d24adf5d9..a750aa687b 100644 --- a/frontend/src/lib/components/flows/flowInfers.ts +++ b/frontend/src/lib/components/flows/flowInfers.ts @@ -4,7 +4,7 @@ import type { Schema } from '$lib/common' import { emptySchema } from '$lib/utils' import type { FlowModule, InputTransform } from '$lib/gen' -export const AI_AGENT_SCHEMA = { +export const AI_AGENT_SCHEMA: Schema = { $schema: 'https://json-schema.org/draft/2020-12/schema', properties: { provider: { @@ -21,7 +21,7 @@ export const AI_AGENT_SCHEMA = { user_message: { type: 'string', description: - 'The message to give as input to the AI agent. You can turn on chat input mode on the input interface to link this field to the message sent by the user.' + 'The message to give as input to the AI agent. Optional when messages array is provided. You can turn on chat input mode on the input interface to link this field to the message sent by the user.' }, system_prompt: { type: 'string', @@ -33,12 +33,86 @@ export const AI_AGENT_SCHEMA = { default: true, showExpr: "fields.output_type === 'text'" }, - messages_context_length: { - type: 'number', + memory: { + type: 'object', description: - 'Maximum number of conversation messages to store and retrieve from memory. If not set or 0, memory is disabled.', - 'x-no-s3-storage-workspace-warning': - 'When no S3 storage is configured in your workspace settings, memory will be stored in database, which implies a limit of 100KB per memory entry. If you need to store more messages, you should use S3 storage in your workspace settings.', + 'Configure how conversation memory is managed. Choose "auto" to let Windmill automatically store and load messages (up to N last messages), or "manual" to provide an explicit array of conversation messages. The system_prompt and user_message are added to the messages if provided.', + oneOf: [ + { + type: 'object', + title: 'auto', + properties: { + kind: { + type: 'string', + enum: ['auto'], + default: 'auto', + description: 'Automatically manage conversation history' + }, + context_length: { + type: 'number', + description: + 'Number of most recent messages to store and load. Set to 0 to disable memory.', + default: 0 + } + }, + required: ['kind'], + 'x-no-s3-storage-workspace-warning': + 'When no S3 storage is configured in your workspace settings, memory will be stored in database, which implies a limit of 100KB per memory entry. If you need to store more messages, you should use S3 storage in your workspace settings.' + }, + { + type: 'object', + title: 'manual', + properties: { + kind: { + type: 'string', + enum: ['manual'], + description: + 'Manually provide conversation messages, bypassing automatic memory management' + }, + messages: { + type: 'array', + description: 'Array of conversation messages to use as history', + items: { + type: 'object', + properties: { + role: { + type: 'string', + enum: ['user', 'assistant', 'system'] + }, + content: { + type: 'string' + }, + tool_calls: { + type: 'array', + nullable: true, + items: { + type: 'object', + properties: { + id: { type: 'string' }, + type: { type: 'string' }, + function: { + type: 'object', + properties: { + name: { type: 'string' }, + arguments: { type: 'string' } + } + } + } + } + }, + tool_call_id: { + type: 'string', + nullable: true, + description: 'The ID of the tool call this message is responding to' + } + }, + required: ['role'] + } + } + }, + required: ['kind', 'messages'] + } + ], showExpr: "fields.output_type === 'text'" }, output_schema: { @@ -52,7 +126,7 @@ export const AI_AGENT_SCHEMA = { description: 'Array of images to give as input to the AI agent. Requires a configured workspace S3 storage.', items: { - type: 'object' as const, + type: 'object', resourceType: 's3object' } }, @@ -73,14 +147,15 @@ export const AI_AGENT_SCHEMA = { default: 10 } }, - required: ['provider', 'user_message', 'output_type'], + required: ['provider', 'output_type'], type: 'object', order: [ 'provider', 'output_type', 'user_message', 'system_prompt', - 'messages_context_length', + 'streaming', + 'memory', 'output_schema', 'user_images', 'max_completion_tokens', @@ -89,6 +164,37 @@ export const AI_AGENT_SCHEMA = { ] } +function migrateAiAgentInputTransforms( + inputTransforms: Record +): Record { + // Check if this has the legacy format + if ('messages_context_length' in inputTransforms && !('memory' in inputTransforms)) { + const legacyValue = inputTransforms.messages_context_length + if (legacyValue) { + if (legacyValue?.type === 'static') { + inputTransforms.memory = { + type: 'static', + value: { + kind: 'auto', + context_length: legacyValue.value ?? 0 + } + } + } else if (legacyValue.type === 'javascript') { + // For dynamic expressions, wrap in the new format + inputTransforms.memory = { + type: 'javascript', + expr: `{ kind: 'auto', context_length: ${legacyValue.expr} }` + } + } + + // Remove the legacy field + delete inputTransforms.messages_context_length + } + } + + return inputTransforms +} + export async function loadSchemaFromModule(module: FlowModule): Promise<{ input_transforms: Record schema: Schema @@ -140,7 +246,7 @@ export async function loadSchemaFromModule(module: FlowModule): Promise<{ schema: schema ?? emptySchema() } } else if (mod.type === 'aiagent') { - let input_transforms = mod.input_transforms ?? {} + let input_transforms = migrateAiAgentInputTransforms(mod.input_transforms ?? {}) return { input_transforms: Object.keys(AI_AGENT_SCHEMA.properties ?? {}).reduce((accu, key) => { accu[key] = input_transforms[key] ?? { diff --git a/frontend/src/lib/components/flows/flowStore.svelte.ts b/frontend/src/lib/components/flows/flowStore.svelte.ts index 957e51b4fb..f73747e915 100644 --- a/frontend/src/lib/components/flows/flowStore.svelte.ts +++ b/frontend/src/lib/components/flows/flowStore.svelte.ts @@ -3,7 +3,6 @@ import { writable } from 'svelte/store' import { initFlowState, type FlowState } from './flowState' import { sendUserToast } from '$lib/toast' import type { StateStore } from '$lib/utils' - export type FlowMode = 'push' | 'pull' export const importFlowStore = writable(undefined) diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index aeb8232c78..ee509ee7a9 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -741,7 +741,7 @@ components: $ref: "#/components/schemas/InputTransform" streaming: $ref: "#/components/schemas/InputTransform" - messages_context_length: + memory: $ref: "#/components/schemas/InputTransform" output_schema: $ref: "#/components/schemas/InputTransform"