kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(aiagent): allow giving messages history (#7395)

Browse Source 
* handle messages array for ai agent

* better

* nit

* make tool_calls and tool_call_id nullable

* fix empty json behavior

* nits

* cleaning

* feat(backend): replace messages/messages_context_length with history oneOf field

Replace the separate 'messages' array and 'messages_context_length' fields with
a single 'history' field that uses a oneOf discriminator.

The 'history' field can be either:
- 'auto' mode: automatically manages conversation history with memory, takes a
  'context_length' number parameter
- 'manual' mode: bypasses memory and uses explicitly provided messages array

Backward compatibility is maintained: if 'messages_context_length' is provided
in the old schema format, it is automatically converted to 'auto' mode with the
specified context_length.

Co-authored-by: centdix <centdix@users.noreply.github.com>

* feat(frontend): replace messages/messages_context_length with history oneOf field

Replace the separate 'messages' array and 'messages_context_length' fields with
a single 'history' field in the AI agent schema.

The 'history' field uses a oneOf discriminator with two modes:
- 'auto': { mode: 'auto', context_length: number } - automatically manages
  conversation history with memory
- 'manual': { mode: 'manual', messages: array } - bypasses memory and uses
  explicitly provided messages

The schema includes comprehensive descriptions for each mode explaining the
behavior. The order array has been updated to include 'history' in place of
the old 'messages_context_length' and 'messages' fields.

Co-authored-by: centdix <centdix@users.noreply.github.com>

* fix(frontend): add support for 'mode' discriminator in oneOf rendering

Update ArgInput.svelte to properly handle oneOf schemas that use 'mode' as the
discriminator field, in addition to the existing 'kind' and 'label' support.

Changes:
- Updated tagKey derivation to check for 'mode' first, then 'kind', then 'label'
- Added 'mode' to the onOneOfChange function to track mode changes
- Added 'mode' to the list of keys excluded from enum validation
- Added 'mode' to hiddenArgs to prevent it from being shown in the form
- Added title fields to the history oneOf variants in flowInfers.ts

This allows the AI agent's history field to properly render with toggle buttons
for 'auto' and 'manual' modes.

Co-authored-by: centdix <centdix@users.noreply.github.com>

* fixes

* frontend fix

* nit

* cleaning

* cleaning

* better

* reword

* reword

---------

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: centdix <centdix@users.noreply.github.com>
This commit is contained in:
centdix
2025-12-20 14:43:44 +01:00
committed by GitHub
parent 582898b5cf
commit bb4b4cf450
10 changed files with 308 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
backend/windmill-worker/src/ai/types.rs
View File

@@ -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<OpenAIMessage>,
},
}
#[derive(Deserialize)]
struct AIAgentArgsRaw {
provider: ProviderWithResource,
system_prompt: Option<String>,
user_message: Option<String>,
temperature: Option<f32>,
max_completion_tokens: Option<u32>,
output_schema: Option<OpenAPISchema>,
output_type: Option<OutputType>,
user_images: Option<Vec<S3Object>>,
streaming: Option<bool>,
max_iterations: Option<usize>,
memory: Option<Memory>,
// Legacy field for backward compatibility
messages_context_length: Option<usize>,
}
#[derive(Debug, Deserialize)]
#[serde(from = "AIAgentArgsRaw")]
pub struct AIAgentArgs {
pub provider: ProviderWithResource,
pub system_prompt: Option<String>,
pub user_message: String,
pub user_message: Option<String>,
pub temperature: Option<f32>,
pub max_completion_tokens: Option<u32>,
pub output_schema: Option<OpenAPISchema>,
pub output_type: Option<OutputType>,
pub user_images: Option<Vec<S3Object>>,
pub streaming: Option<bool>,
pub messages_context_length: Option<usize>,
pub max_iterations: Option<usize>,
pub memory: Option<Memory>,
}
impl From<AIAgentArgsRaw> 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)]

176
backend/windmill-worker/src/ai_executor.rs
View File

@@ -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<OpenAIMessage> =
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<OpenAIMessage> =
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
);
}
}
}
}

5
frontend/src/lib/common.ts
View File

@@ -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[]

11
frontend/src/lib/components/ArgInput.svelte
View File

@@ -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']}
<Alert
type="warning"
title={obj['x-no-s3-storage-workspace-warning']}
size="xs"
titleClass="text-2xs"
/>
{/if}
{:else if disabled}
<textarea disabled></textarea>
{:else}

5
frontend/src/lib/components/JsonEditor.svelte
View File

@@ -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) {

4
frontend/src/lib/components/common/alert/Alert.svelte
View File

@@ -80,7 +80,7 @@
)}
style={bgStyle}
>
<div class="flex">
<div class="flex flex-row items-center">
<div class="flex h-8 w-8 items-center justify-center rounded-full">
<SvelteComponent
class={twMerge(classes[type].iconClass, iconClass)}
@@ -89,7 +89,7 @@
/>
</div>
<div class={twMerge('ml-1 w-full')}>
<div class={twMerge('w-full flex flex-row items-center justify-between h-8')}>
<div class={twMerge('w-full flex flex-row items-center justify-between')}>
<span
class={twMerge('text-xs font-semibold', classes[type].titleClass, titleClass)}
style={titleStyle}

8
frontend/src/lib/components/flows/content/FlowInput.svelte
View File

@@ -464,8 +464,8 @@
(accu, key) => {
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(

128
frontend/src/lib/components/flows/flowInfers.ts
View File

@@ -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<string, InputTransform>
): Record<string, InputTransform> {
// 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<string, InputTransform>
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] ?? {

1
frontend/src/lib/components/flows/flowStore.svelte.ts
View File

@@ -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<Flow | undefined>(undefined)

2
openflow.openapi.yaml
View File

@@ -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"