diff --git a/apps/rowboat/app/lib/agent_instructions.ts b/apps/rowboat/app/lib/agent_instructions.ts index ad6567ec..6b0d442e 100644 --- a/apps/rowboat/app/lib/agent_instructions.ts +++ b/apps/rowboat/app/lib/agent_instructions.ts @@ -129,4 +129,22 @@ export const TASK_TYPE_INSTRUCTIONS = (): string => ` - Reading the messages in the chat history will give you context about the conversation. - Seeing the tool calls that transfer / handoff control will help you understand the flow of the conversation and which agent produced each message. - These are high level instructions only. The user will provide more specific instructions which will be below. +`; + +export const PIPELINE_TYPE_INSTRUCTIONS = (): string => ` +- You are a pipeline agent that is part of a sequential execution chain within a larger workflow. +- You are executing as one step in a multi-step pipeline process. +- Your input comes from the previous step in the pipeline (or the initial input if you're the first step). +- Your output will be passed to the next step in the pipeline (or returned as the final result if you're the last step). +- CRITICAL: You CANNOT transfer to other agents or pipelines. You can only use tools to complete your specific task. +- Focus ONLY on your designated role in the pipeline. Process the input, perform your specific task, and provide clear output. +- Use the JSON format to convey your responses. The JSON should have 3 keys: + - "thought": Analyze the input from the previous pipeline step and plan what you need to do + - "response": Your processed output that will be passed to the next pipeline step. Make this clear and actionable. + - "pipeline_context": Brief notes about what you accomplished for the pipeline flow +- Do NOT attempt to handle tasks outside your specific pipeline role. +- Do NOT mention other agents or the pipeline structure to users. +- Your response should be self-contained and ready to be consumed by the next pipeline step. +- Reading the message history will show you the pipeline execution flow up to your step. +- These are high level instructions only. The user will provide more specific instructions which will be below. `; \ No newline at end of file diff --git a/apps/rowboat/app/lib/agents.ts b/apps/rowboat/app/lib/agents.ts index 35d5ca8c..3703a8f5 100644 --- a/apps/rowboat/app/lib/agents.ts +++ b/apps/rowboat/app/lib/agents.ts @@ -16,11 +16,12 @@ import { getMcpClient } from "./mcp"; import { dataSourceDocsCollection, dataSourcesCollection, projectsCollection } from "./mongodb"; import { qdrantClient } from '../lib/qdrant'; import { EmbeddingRecord } from "./types/datasource_types"; -import { ConnectedEntity, sanitizeTextWithMentions, Workflow, WorkflowAgent, WorkflowPrompt, WorkflowTool } from "./types/workflow_types"; -import { CHILD_TRANSFER_RELATED_INSTRUCTIONS, CONVERSATION_TYPE_INSTRUCTIONS, RAG_INSTRUCTIONS, TASK_TYPE_INSTRUCTIONS } from "./agent_instructions"; +import { ConnectedEntity, sanitizeTextWithMentions, Workflow, WorkflowAgent, WorkflowPipeline, WorkflowPrompt, WorkflowTool } from "./types/workflow_types"; +import { CHILD_TRANSFER_RELATED_INSTRUCTIONS, CONVERSATION_TYPE_INSTRUCTIONS, PIPELINE_TYPE_INSTRUCTIONS, RAG_INSTRUCTIONS, TASK_TYPE_INSTRUCTIONS } from "./agent_instructions"; import { PrefixLogger } from "./utils"; import { Message, AssistantMessage, AssistantMessageWithToolCalls, ToolMessage } from "./types/types"; +// Make everything available as a promise const PROVIDER_API_KEY = process.env.PROVIDER_API_KEY || process.env.OPENAI_API_KEY || ''; const PROVIDER_BASE_URL = process.env.PROVIDER_BASE_URL || undefined; const MODEL = process.env.PROVIDER_DEFAULT_MODEL || 'gpt-4o'; @@ -518,7 +519,11 @@ ${config.description} ## About You -${config.outputVisibility === 'user_facing' ? CONVERSATION_TYPE_INSTRUCTIONS() : TASK_TYPE_INSTRUCTIONS()} +${config.outputVisibility === 'user_facing' + ? CONVERSATION_TYPE_INSTRUCTIONS() + : config.type === 'pipeline' + ? PIPELINE_TYPE_INSTRUCTIONS() + : TASK_TYPE_INSTRUCTIONS()} ## Instructions @@ -531,20 +536,16 @@ ${'-'.repeat(100)} ${CHILD_TRANSFER_RELATED_INSTRUCTIONS} `; - let { sanitized, entities } = sanitizeTextWithMentions(instructions, workflow); + let { sanitized, entities } = sanitizeTextWithMentions(instructions, workflow, config); + + // Remove agent transfer instructions for pipeline agents + if (config.type === 'pipeline') { + sanitized = sanitized.replace(CHILD_TRANSFER_RELATED_INSTRUCTIONS, ''); + } + agentLogger.log(`instructions: ${JSON.stringify(sanitized)}`); agentLogger.log(`mentions: ${JSON.stringify(entities)}`); - // // add prompts to instructions - // for (const e of entities) { - // if (e.type === 'prompt') { - // const prompt = promptConfig[e.name]; - // if (prompt) { - // compiledInstructions = compiledInstructions + '\n\n# ' + prompt.name + '\n' + prompt.prompt; - // } - // } - // } - const agentTools = entities.filter(e => e.type === 'tool').map(e => tools[e.name]).filter(Boolean) as Tool[]; // Add RAG tool if needed @@ -641,22 +642,23 @@ function getStartOfTurnAgentName( logger.log(`last agent ${lastAgentName} not found in agent config, returning start agent: ${workflow.startAgent}`); return workflow.startAgent; } + + // For other agents, check control type switch (lastAgentConfig.controlType) { case 'retain': logger.log(`last agent ${lastAgentName} control type is retain, returning last agent: ${lastAgentName}`); return lastAgentName; case 'relinquish_to_parent': const parentAgentName = startAgentStack.pop() || workflow.startAgent; - if (startAgentStack.length > 0) { - logger.log(`popped agent from stack: ${lastAgentName} || reason: relinquish to parent triggered`); - } else { - logger.log(`using start agent: ${lastAgentName} || reason: empty stack`); - } logger.log(`last agent ${lastAgentName} control type is relinquish_to_parent, returning most recent parent: ${parentAgentName}`); return parentAgentName; case 'relinquish_to_start': logger.log(`last agent ${lastAgentName} control type is relinquish_to_start, returning start agent: ${workflow.startAgent}`); return workflow.startAgent; + default: + // Fallback for any unexpected control type + logger.log(`last agent ${lastAgentName} has unexpected control type: ${lastAgentConfig.controlType}, returning start agent: ${workflow.startAgent}`); + return workflow.startAgent; } } @@ -767,6 +769,7 @@ function mapConfig(workflow: z.infer): { agentConfig: Record>; toolConfig: Record>; promptConfig: Record>; + pipelineConfig: Record>; } { const agentConfig: Record> = workflow.agents.reduce((acc, agent) => ({ ...acc, @@ -780,7 +783,13 @@ function mapConfig(workflow: z.infer): { ...acc, [prompt.name]: prompt }), {}); - return { agentConfig, toolConfig, promptConfig }; + + const pipelineConfig: Record> = (workflow.pipelines || []).reduce((acc, pipeline) => ({ + ...acc, + [pipeline.name]: pipeline + }), {}); + + return { agentConfig, toolConfig, promptConfig, pipelineConfig }; } async function* emitGreetingTurn(logger: PrefixLogger, workflow: z.infer): AsyncIterable | z.infer> { @@ -807,27 +816,29 @@ function createTools( toolConfig: Record>, ): Record { const tools: Record = {}; + const toolLogger = logger.child('createTools'); + + toolLogger.log(`=== CREATING ${Object.keys(toolConfig).length} TOOLS ===`); + for (const [toolName, config] of Object.entries(toolConfig)) { - if (workflow.mockTools?.[toolName]) { - tools[toolName] = createMockTool(logger, { - ...config, - mockInstructions: workflow.mockTools?.[toolName], // override mock instructions - }); - logger.log(`created mock tool: ${toolName}`); - } else if (config.mockTool) { + toolLogger.log(`creating tool: ${toolName} (type: ${config.mockTool ? 'mock' : config.isMcp ? 'mcp' : config.isComposio ? 'composio' : 'webhook'})`); + + if (config.mockTool) { tools[toolName] = createMockTool(logger, config); - logger.log(`created mock tool: ${toolName}`); + toolLogger.log(`✓ created mock tool: ${toolName}`); } else if (config.isMcp) { tools[toolName] = createMcpTool(logger, config, projectId); - logger.log(`created mcp tool: ${toolName}`); + toolLogger.log(`✓ created mcp tool: ${toolName} (server: ${config.mcpServerName || 'unknown'})`); } else if (config.isComposio) { tools[toolName] = createComposioTool(logger, config, projectId); - logger.log(`created composio tool: ${toolName}`); + toolLogger.log(`✓ created composio tool: ${toolName}`); } else { tools[toolName] = createWebhookTool(logger, config, projectId); - logger.log(`created webhook tool: ${toolName}`); + toolLogger.log(`✓ created webhook tool: ${toolName} (fallback)`); } } + + toolLogger.log(`=== TOOL CREATION COMPLETE ===`); return tools; } @@ -838,14 +849,34 @@ function createAgents( agentConfig: Record>, tools: Record, promptConfig: Record>, + pipelineConfig: Record>, ): { agents: Record, mentions: Record[]>, originalInstructions: Record, originalHandoffs: Record } { + const agentsLogger = logger.child('createAgents'); const agents: Record = {}; const mentions: Record[]> = {}; const originalInstructions: Record = {}; const originalHandoffs: Record = {}; + agentsLogger.log(`=== CREATING ${Object.keys(agentConfig).length} AGENTS ===`); + + // Create pipeline entities that will be available for @ referencing + const pipelineEntities: z.infer[] = Object.keys(pipelineConfig).map(pipelineName => ({ + type: 'pipeline' as const, + name: pipelineName, + })); + if (pipelineEntities.length > 0) { + agentsLogger.log(`available pipeline entities for @ referencing: ${pipelineEntities.map(p => p.name).join(', ')}`); + } + // create agents for (const [agentName, config] of Object.entries(agentConfig)) { + agentsLogger.log(`creating agent: ${agentName} (type: ${config.outputVisibility}, control: ${config.controlType})`); + + // Pipeline agents get special handling: + // - Different instruction template (PIPELINE_TYPE_INSTRUCTIONS) + // - Filtered mentions (tools only, no agents) + // - No agent transfer instructions + const { agent, entities } = createAgent( logger, projectId, @@ -855,18 +886,96 @@ function createAgents( promptConfig, ); agents[agentName] = agent; - mentions[agentName] = entities; + + // Add pipeline entities to the agent's available mentions (unless it's a pipeline agent itself) + // Pipeline agents cannot reference other agents or pipelines, only tools + let agentEntities = entities; + if (config.type !== 'pipeline') { + agentEntities = [...entities, ...pipelineEntities]; + agentsLogger.log(`${agentName} can reference: ${entities.length} entities + ${pipelineEntities.length} pipelines`); + } else { + agentsLogger.log(`${agentName} (pipeline agent) can reference: ${entities.length} entities only`); + } + + mentions[agentName] = agentEntities; originalInstructions[agentName] = agent.instructions as string; // handoffs will be set after all agents are created } + agentsLogger.log(`=== SETTING UP HANDOFFS ===`); + // set handoffs for (const [agentName, agent] of Object.entries(agents)) { const connectedAgentNames = (mentions[agentName] || []).filter(e => e.type === 'agent').map(e => e.name); + const connectedPipelineNames = (mentions[agentName] || []).filter(e => e.type === 'pipeline').map(e => e.name); + + // Pipeline agents have no agent handoffs (filtered out in validatePipelineAgentMentions) + // They only have tool connections, no agent transfers allowed + + // Filter out pipeline agents from being handoff targets + // Only allow handoffs to non-pipeline agents + const validAgentNames = connectedAgentNames.filter(name => { + const targetConfig = agentConfig[name]; + return targetConfig && targetConfig.type !== 'pipeline'; + }); + + // Convert pipeline mentions to handoffs to the first agent in each pipeline + const pipelineFirstAgents: string[] = []; + for (const pipelineName of connectedPipelineNames) { + const pipeline = pipelineConfig[pipelineName]; + if (pipeline && pipeline.agents.length > 0) { + const firstAgent = pipeline.agents[0]; + if (agentConfig[firstAgent] && !pipelineFirstAgents.includes(firstAgent)) { + pipelineFirstAgents.push(firstAgent); + agentsLogger.log(`${agentName} pipeline mention ${pipelineName} -> handoff to first agent: ${firstAgent}`); + } + } + } + + // Combine regular agent handoffs with pipeline first agents + const allHandoffTargets = [...validAgentNames, ...pipelineFirstAgents]; + // Only store Agent objects in handoffs (filter out Handoff if present) - agent.handoffs = connectedAgentNames.map(e => agents[e]).filter(Boolean) as Agent[]; - originalHandoffs[agentName] = agent.handoffs.filter(h => h instanceof Agent); - logger.log(`set handoffs for ${agentName}: ${JSON.stringify(connectedAgentNames)}`); + const agentHandoffs = allHandoffTargets.map(e => agents[e]).filter(Boolean) as Agent[]; + agent.handoffs = agentHandoffs; + originalHandoffs[agentName] = agentHandoffs.filter(h => h instanceof Agent); + agentsLogger.log(`set handoffs for ${agentName}: ${JSON.stringify(allHandoffTargets)}`); + } + + // Set up pipeline agent handoff chains + agentsLogger.log(`=== SETTING UP PIPELINE CHAINS ===`); + for (const [pipelineName, pipeline] of Object.entries(pipelineConfig)) { + agentsLogger.log(`setting up pipeline chain: ${pipelineName} -> [${pipeline.agents.join(' -> ')}]`); + + for (let i = 0; i < pipeline.agents.length; i++) { + const currentAgentName = pipeline.agents[i]; + const currentAgent = agents[currentAgentName]; + + if (!currentAgent) { + agentsLogger.log(`warning: pipeline agent ${currentAgentName} not found in agent config`); + continue; + } + + // Pipeline agents have NO handoffs - they just execute once + currentAgent.handoffs = []; + + // Add pipeline metadata to the agent for easy lookup + (currentAgent as any).pipelineName = pipelineName; + (currentAgent as any).pipelineIndex = i; + (currentAgent as any).isLastInPipeline = i === pipeline.agents.length - 1; + + // Update originalHandoffs to reflect the final pipeline state + originalHandoffs[currentAgentName] = []; + + agentsLogger.log(`pipeline agent ${currentAgentName} has no handoffs (will be controlled by pipeline controller)`); + agentsLogger.log(`pipeline agent ${currentAgentName} metadata: pipeline=${pipelineName}, index=${i}, isLast=${i === pipeline.agents.length - 1}`); + + // Configure pipeline agents to relinquish control after completing their task + const agentConfigObj = agentConfig[currentAgentName]; + if (agentConfigObj && agentConfigObj.type === 'pipeline') { + agentsLogger.log(`configuring pipeline agent ${currentAgentName} to relinquish control after task completion`); + } + } } return { agents, mentions, originalInstructions, originalHandoffs }; @@ -910,10 +1019,19 @@ function maybeInjectGiveUpControlInstructions( const agentConfigObj = agentConfig[childAgentName]; const isInternal = agentConfigObj?.outputVisibility === 'internal'; + const isPipeline = agentConfigObj?.type === 'pipeline'; const isRetain = agentConfigObj?.controlType === 'retain'; const injectLogger = logger.child(`inject`); injectLogger.log(`isInternal: ${isInternal}`); + injectLogger.log(`isPipeline: ${isPipeline}`); injectLogger.log(`isRetain: ${isRetain}`); + + // For pipeline agents, they should continue pipeline execution, so no need to inject give up control + if (isPipeline) { + injectLogger.log(`Pipeline agent ${childAgentName} continues pipeline execution, no give up control needed`); + return; + } + if (!isInternal && isRetain) { // inject give up control instructions agents[childAgentName].instructions = getGiveUpControlInstructions(agents[childAgentName], parentAgentName, injectLogger); @@ -926,6 +1044,64 @@ function maybeInjectGiveUpControlInstructions( } } +// Pipeline controller function to handle pipeline agent execution and transfers +function handlePipelineAgentExecution( + currentAgent: Agent, + currentAgentName: string, + pipelineConfig: Record>, + stack: string[], + logger: PrefixLogger, + turnMsgs: z.infer[], + transferCounter: AgentTransferCounter, + createTransferEvents: (fromAgent: string, toAgent: string) => [z.infer, z.infer] +): { nextAgentName: string | null; shouldContinue: boolean; transferEvents?: [z.infer, z.infer] } { + const pipelineName = (currentAgent as any).pipelineName; + const pipelineIndex = (currentAgent as any).pipelineIndex; + const isLastInPipeline = (currentAgent as any).isLastInPipeline; + + if (!pipelineName || pipelineIndex === undefined) { + logger.log(`warning: pipeline agent ${currentAgentName} missing pipeline metadata`); + return { nextAgentName: null, shouldContinue: false }; + } + + const pipeline = pipelineConfig[pipelineName]; + if (!pipeline) { + logger.log(`warning: pipeline ${pipelineName} not found in config`); + return { nextAgentName: null, shouldContinue: false }; + } + + let nextAgentName: string | null = null; + + if (!isLastInPipeline) { + // Not the last agent - continue to next agent in pipeline + nextAgentName = pipeline.agents[pipelineIndex + 1]; + logger.log(`-- pipeline controller: ${currentAgentName} -> ${nextAgentName} (continuing pipeline ${pipelineName})`); + } else { + // Last agent - return to calling agent + nextAgentName = stack.pop()!; + logger.log(`-- pipeline controller: ${currentAgentName} -> ${nextAgentName} (pipeline ${pipelineName} complete, returning to caller)`); + } + + if (nextAgentName) { + // Create transfer events for pipeline continuation + const transferEvents = createTransferEvents(currentAgentName, nextAgentName); + const [transferStart, transferComplete] = transferEvents; + + // Add messages to turn + turnMsgs.push(transferStart); + turnMsgs.push(transferComplete); + + // Update transfer counter + transferCounter.increment(currentAgentName, nextAgentName); + + logger.log(`switched to agent: ${nextAgentName} || reason: pipeline controller transfer`); + + return { nextAgentName, shouldContinue: true, transferEvents }; + } + + return { nextAgentName: null, shouldContinue: false }; +} + // Main function to stream an agentic response // using OpenAI Agents SDK export async function* streamResponse( @@ -949,8 +1125,16 @@ export async function* streamResponse( } // create map of agent, tool and prompt configs - const { agentConfig, toolConfig, promptConfig } = mapConfig(workflow); + const { agentConfig, toolConfig, promptConfig, pipelineConfig } = mapConfig(workflow); + // Debug: Log configuration summary + logger.log(`=== WORKFLOW CONFIGURATION ===`); + logger.log(`agents: ${Object.keys(agentConfig).length} (${Object.keys(agentConfig).join(', ')})`); + logger.log(`tools: ${Object.keys(toolConfig).length} (${Object.keys(toolConfig).join(', ')})`); + logger.log(`prompts: ${Object.keys(promptConfig).length} (${Object.keys(promptConfig).join(', ')})`); + logger.log(`pipelines: ${Object.keys(pipelineConfig).length} (${Object.keys(pipelineConfig).join(', ')})`); + logger.log(`start agent: ${workflow.startAgent}`); + logger.log(`=== END CONFIGURATION ===`); const stack: string[] = []; logger.log(`initialized stack: ${JSON.stringify(stack)}`); @@ -959,35 +1143,45 @@ export async function* streamResponse( const tools = createTools(logger, projectId, workflow, toolConfig); // create agents - const { agents, originalInstructions, originalHandoffs } = createAgents(logger, projectId, workflow, agentConfig, tools, promptConfig); + const { agents, originalInstructions, originalHandoffs } = createAgents(logger, projectId, workflow, agentConfig, tools, promptConfig, pipelineConfig); // track agent to agent calls const transferCounter = new AgentTransferCounter(); - // track usage + // get the agent that should be starting this turn + const startOfTurnAgentName = getStartOfTurnAgentName(logger, messages, agentConfig, workflow); + logger.log(`🎯 START AGENT DECISION: ${startOfTurnAgentName}`); + + let agentName = startOfTurnAgentName; + + // start the turn loop const usageTracker = new UsageTracker(); - - // get next agent name - let agentName = getStartOfTurnAgentName(logger, messages, agentConfig, workflow); - - // set up initial state for loop - logger.log('@@ starting agent turn @@'); - let iter = 0; const turnMsgs: z.infer[] = [...messages]; + logger.log('🎬 STARTING AGENT TURN'); + + // stack-based agent execution loop + let iter = 0; + const MAXTURNITERATIONS = 10; + // loop indefinitely turnLoop: while (true) { - logger.log(`starting turn loop iteration: ${iter}`); + logger.log(`🔄 TURN ITERATION: ${iter + 1}/${MAXTURNITERATIONS}`); + const loopLogger = logger.child(`iter-${iter + 1}`); + + loopLogger.log(`🤖 CURRENT AGENT: ${agentName}`); + loopLogger.log(`📚 AGENT STACK: [${stack.join(' -> ')}]`); + // increment loop counter iter++; // set up logging - const loopLogger = logger.child(`iter-${iter}`); + // const loopLogger = logger.child(`iter-${iter}`); // log agent info - loopLogger.log(`agent name: ${agentName}`); - loopLogger.log(`stack: ${JSON.stringify(stack)}`); + // loopLogger.log(`agent name: ${agentName}`); + // loopLogger.log(`stack: ${JSON.stringify(stack)}`); if (!agents[agentName]) { throw new Error(`agent not found in agent config!`); } @@ -1050,9 +1244,11 @@ export async function* streamResponse( case 'run_item_stream_event': // handle handoff event if (event.name === 'handoff_occurred' && event.item.type === 'handoff_output_item') { + eventLogger.log(`🔄 HANDOFF EVENT: ${agentName} -> ${event.item.targetAgent.name}`); + // skip if its the same agent if (agentName === event.item.targetAgent.name) { - eventLogger.log(`skipping handoff to same agent: ${agentName}`); + eventLogger.log(`⚠️ SKIPPING: handoff to same agent: ${agentName}`); break; } @@ -1062,9 +1258,10 @@ export async function* streamResponse( const maxCalls = targetAgentConfig?.maxCallsPerParentAgent || 3; const currentCalls = transferCounter.get(agentName, event.item.targetAgent.name); if (currentCalls >= maxCalls) { - eventLogger.log(`skipping handoff to ${event.item.targetAgent.name} || reason: max calls ${maxCalls} exceeded from ${agentName} to internal agent ${event.item.targetAgent.name}`); + eventLogger.log(`⚠️ SKIPPING: handoff to ${event.item.targetAgent.name} - max calls ${maxCalls} exceeded from ${agentName}`); continue; } + eventLogger.log(`📊 TRANSFER COUNT: ${agentName} -> ${event.item.targetAgent.name} = ${currentCalls}/${maxCalls}`); } // inject give up control instructions if needed (parent handing off to child) @@ -1094,13 +1291,14 @@ export async function* streamResponse( const newAgentName = event.item.targetAgent.name; - loopLogger.log(`switched to agent: ${newAgentName} || reason: handoff by ${agentName}`); + loopLogger.log(`🔄 AGENT SWITCH: ${agentName} -> ${newAgentName} (reason: handoff)`); // add current agent to stack only if new agent is internal - if (agentConfig[newAgentName]?.outputVisibility === 'internal') { + const newAgentConfig = agentConfig[newAgentName]; + if (newAgentConfig?.outputVisibility === 'internal' || newAgentConfig?.type === 'pipeline') { stack.push(agentName); - loopLogger.log(`-- pushed agent to stack: ${agentName} || reason: new agent ${newAgentName} is internal`); - loopLogger.log(`-- stack is now: ${JSON.stringify(stack)}`); + loopLogger.log(`📚 STACK PUSH: ${agentName} (new agent ${newAgentName} is internal/pipeline)`); + loopLogger.log(`📚 STACK NOW: [${stack.join(' -> ')}]`); } // set this as the new agent name @@ -1132,7 +1330,8 @@ export async function* streamResponse( event.item.rawItem.type === 'message' && event.item.rawItem.status === 'completed') { // check response visibility - const isInternal = agentConfig[agentName]?.outputVisibility === 'internal'; + const agentConfigObj = agentConfig[agentName]; + const isInternal = agentConfigObj?.outputVisibility === 'internal' || agentConfigObj?.type === 'pipeline'; for (const content of event.item.rawItem.content) { if (content.type === 'output_text') { // create message @@ -1151,40 +1350,71 @@ export async function* streamResponse( } } - // if this is an internal agent, switch to previous agent + // if this is an internal agent or pipeline agent, switch to previous agent if (isInternal) { const current = agentName; + const currentAgentConfig = agentConfig[agentName]; - // if the control type is relinquish_to_parent or retain, we need to pop the stack, else if the control type is relinquish_to_start, we need to use the start agent - if (agentConfig[agentName]?.controlType === 'relinquish_to_parent' || agentConfig[agentName]?.controlType === 'retain') { + // Check if this is a pipeline agent that needs to continue the pipeline + if (currentAgentConfig?.type === 'pipeline') { + const result = handlePipelineAgentExecution( + agents[current], // Use the correct agent from agents collection + current, + pipelineConfig, + stack, + loopLogger, + turnMsgs, + transferCounter, + createTransferEvents + ); + + // Emit transfer events if they exist + if (result.transferEvents) { + const [transferStart, transferComplete] = result.transferEvents; + yield* emitEvent(eventLogger, transferStart); + yield* emitEvent(eventLogger, transferComplete); + } + + if (result.shouldContinue) { + agentName = result.nextAgentName!; + // Run the turn from the next agent + continue turnLoop; + } + } + + // Check control type to determine next action for non-pipeline agents + if (currentAgentConfig?.controlType === 'relinquish_to_parent' || currentAgentConfig?.controlType === 'retain') { agentName = stack.pop()!; - loopLogger.log(`-- popped agent from stack: ${agentName} || reason: ${current} is an internal agent, it put out a message and it has a control type of ${agentConfig[agentName]?.controlType}, hence the flow of control needs to return to the previous agent`); - } else if (agentConfig[agentName]?.controlType === 'relinquish_to_start') { + loopLogger.log(`-- popped agent from stack: ${agentName} || reason: ${current} is an internal agent, it put out a message and it has a control type of ${currentAgentConfig?.controlType}, hence the flow of control needs to return to the previous agent`); + } else if (currentAgentConfig?.controlType === 'relinquish_to_start') { agentName = workflow.startAgent; - loopLogger.log(`-- using start agent: ${agentName} || reason: ${current} is an internal agent, it put out a message and it has a control type of ${agentConfig[agentName]?.controlType}, hence the flow of control needs to return to the start agent`); + loopLogger.log(`-- using start agent: ${agentName} || reason: ${current} is an internal agent, it put out a message and it has a control type of ${currentAgentConfig?.controlType}, hence the flow of control needs to return to the start agent`); } - loopLogger.log(`-- stack is now: ${JSON.stringify(stack)}`); + // Only emit transfer events if we're actually changing agents + if (agentName !== current) { + loopLogger.log(`-- stack is now: ${JSON.stringify(stack)}`); - // emit transfer tool call invocation - const [transferStart, transferComplete] = createTransferEvents(current, agentName); + // emit transfer tool call invocation + const [transferStart, transferComplete] = createTransferEvents(current, agentName); - // add messages to turn - turnMsgs.push(transferStart); - turnMsgs.push(transferComplete); + // add messages to turn + turnMsgs.push(transferStart); + turnMsgs.push(transferComplete); - // emit events - yield* emitEvent(eventLogger, transferStart); - yield* emitEvent(eventLogger, transferComplete); + // emit events + yield* emitEvent(eventLogger, transferStart); + yield* emitEvent(eventLogger, transferComplete); - // update transfer counter - transferCounter.increment(current, agentName); + // update transfer counter + transferCounter.increment(current, agentName); - // set this as the new agent name - loopLogger.log(`switched to agent: ${agentName} || reason: internal agent (${current}) put out a message`); + // set this as the new agent name + loopLogger.log(`switched to agent: ${agentName} || reason: internal agent (${current}) put out a message`); - // run the turn from the previous agent - continue turnLoop; + // run the turn from the previous agent + continue turnLoop; + } } break; } diff --git a/apps/rowboat/app/lib/components/atmentions.ts b/apps/rowboat/app/lib/components/atmentions.ts index 0d52adf1..f1b94712 100644 --- a/apps/rowboat/app/lib/components/atmentions.ts +++ b/apps/rowboat/app/lib/components/atmentions.ts @@ -8,29 +8,52 @@ interface CreateAtMentionsProps { agents: any[]; prompts: any[]; tools: any[]; + pipelines?: any[]; currentAgentName?: string; + currentAgent?: any; // Add current agent object to know its outputVisibility } -export function createAtMentions({ agents, prompts, tools, currentAgentName }: CreateAtMentionsProps): AtMentionItem[] { +export function createAtMentions({ agents, prompts, tools, pipelines = [], currentAgentName, currentAgent }: CreateAtMentionsProps): AtMentionItem[] { const atMentions: AtMentionItem[] = []; - // Add agents - for (const a of agents) { - if (a.disabled || a.name === currentAgentName) { - continue; + // For pipeline agents, only add tools and prompts - no agents or pipelines + const isCurrentAgentPipeline = currentAgent?.type === 'pipeline'; + + // Add agents (excluding pipeline agents and disabled agents) + // Also exclude ALL agents if current agent is a pipeline agent + if (!isCurrentAgentPipeline) { + for (const a of agents) { + if (a.disabled || a.name === currentAgentName || a.type === 'pipeline') { + continue; + } + const id = `agent:${a.name}`; + atMentions.push({ + id, + value: id, + label: `Agent: ${a.name}`, + denotationChar: "@", // Add required properties for Match type + link: id, + target: "_self" + }); } - const id = `agent:${a.name}`; - atMentions.push({ - id, - value: id, - label: `Agent: ${a.name}`, - denotationChar: "@", // Add required properties for Match type - link: id, - target: "_self" - }); } - // Add prompts + // Add pipelines (only if current agent is not a pipeline agent) + if (!isCurrentAgentPipeline) { + for (const pipeline of pipelines) { + const id = `pipeline:${pipeline.name}`; + atMentions.push({ + id, + value: id, + label: `Pipeline: ${pipeline.name}`, + denotationChar: "@", + link: id, + target: "_self" + }); + } + } + + // Add prompts (always allowed) for (const prompt of prompts) { const id = `prompt:${prompt.name}`; atMentions.push({ @@ -43,7 +66,7 @@ export function createAtMentions({ agents, prompts, tools, currentAgentName }: C }); } - // Add tools + // Add tools (always allowed) for (const tool of tools) { const id = `tool:${tool.name}`; atMentions.push({ diff --git a/apps/rowboat/app/lib/types/workflow_types.ts b/apps/rowboat/app/lib/types/workflow_types.ts index 2a8e5dbb..edd03a04 100644 --- a/apps/rowboat/app/lib/types/workflow_types.ts +++ b/apps/rowboat/app/lib/types/workflow_types.ts @@ -6,6 +6,7 @@ export const WorkflowAgent = z.object({ 'conversation', 'post_process', 'escalation', + 'pipeline', ]), description: z.string(), disabled: z.boolean().default(false).optional(), @@ -23,8 +24,32 @@ export const WorkflowAgent = z.object({ 'retain', 'relinquish_to_parent', 'relinquish_to_start', - ]).default('retain').describe('Whether this agent retains control after a turn, relinquishes to the parent agent, or relinquishes to the start agent'), + ]).optional().describe('Whether this agent retains control after a turn, relinquishes to the parent agent, or relinquishes to the start agent'), maxCallsPerParentAgent: z.number().default(3).describe('Maximum number of times this agent can be called by a parent agent in a single turn').optional(), +}).refine((data) => { + // Pipeline agents should have internal output visibility and relinquish_to_parent control type + if (data.type === 'pipeline' && data.outputVisibility !== 'internal') { + return false; + } + if (data.type === 'pipeline' && data.controlType !== 'relinquish_to_parent') { + return false; + } + // Internal agents should have relinquish_to_parent control type + if (data.outputVisibility === 'internal' && data.controlType !== 'relinquish_to_parent') { + return false; + } + // User-facing agents should not have relinquish_to_parent control type + if (data.outputVisibility === 'user_facing' && data.controlType === 'relinquish_to_parent') { + return false; + } + // All agents should have a control type + if (data.controlType === undefined) { + return false; + } + return true; +}, { + message: "Pipeline agents must have 'internal' output visibility and 'relinquish_to_parent' control type, while other agents must have appropriate control types", + path: ["controlType", "outputVisibility"] }); export const WorkflowPrompt = z.object({ name: z.string(), @@ -58,10 +83,19 @@ export const WorkflowTool = z.object({ logo: z.string(), // the logo for the Composio tool }).optional(), // the data for the Composio tool, if it is a Composio tool }); + +export const WorkflowPipeline = z.object({ + name: z.string(), + description: z.string().optional(), + agents: z.array(z.string()), // ordered list of agent names in the pipeline + order: z.number().int().optional(), +}); + export const Workflow = z.object({ agents: z.array(WorkflowAgent), prompts: z.array(WorkflowPrompt), tools: z.array(WorkflowTool), + pipelines: z.array(WorkflowPipeline).optional(), startAgent: z.string(), lastUpdatedAt: z.string().datetime(), mockTools: z.record(z.string(), z.string()).optional(), // a dict of toolName => mockInstructions @@ -76,7 +110,7 @@ export const WorkflowTemplate = Workflow }); export const ConnectedEntity = z.object({ - type: z.enum(['tool', 'prompt', 'agent']), + type: z.enum(['tool', 'prompt', 'agent', 'pipeline']), name: z.string(), }); @@ -86,13 +120,15 @@ export function sanitizeTextWithMentions( agents: z.infer[], tools: z.infer[], prompts: z.infer[], + pipelines?: z.infer[], }, + currentAgent?: z.infer, ): { sanitized: string; entities: z.infer[]; } { - // Regex to match [@type:name](#type:something) pattern where type is tool/prompt/agent - const mentionRegex = /\[@(tool|prompt|agent):([^\]]+)\]\(#mention\)/g; + // Regex to match [@type:name](#type:something) pattern where type is tool/prompt/agent/pipeline + const mentionRegex = /\[@(tool|prompt|agent|pipeline):([^\]]+)\]\(#mention\)/g; const seen = new Set(); // collect entities @@ -107,18 +143,28 @@ export function sanitizeTextWithMentions( }) .map(match => { return { - type: match[1] as 'tool' | 'prompt' | 'agent', + type: match[1] as 'tool' | 'prompt' | 'agent' | 'pipeline', name: match[2], }; }) .filter(entity => { seen.add(entity.name); + + // For pipeline agents, only allow tool and prompt mentions + if (currentAgent?.type === 'pipeline') { + return entity.type === 'tool' || entity.type === 'prompt'; + } + if (entity.type === 'agent') { - return workflow.agents.some(a => a.name === entity.name); + // Filter out pipeline agents - they should not be @ referenceable + const agent = workflow.agents.find(a => a.name === entity.name); + return agent && agent.type !== 'pipeline'; } else if (entity.type === 'tool') { return workflow.tools.some(t => t.name === entity.name); } else if (entity.type === 'prompt') { return workflow.prompts.some(p => p.name === entity.name); + } else if (entity.type === 'pipeline') { + return workflow.pipelines?.some(p => p.name === entity.name); } return false; }) diff --git a/apps/rowboat/app/projects/[projectId]/entities/agent_config.tsx b/apps/rowboat/app/projects/[projectId]/entities/agent_config.tsx index 4147b433..b3b8045a 100644 --- a/apps/rowboat/app/projects/[projectId]/entities/agent_config.tsx +++ b/apps/rowboat/app/projects/[projectId]/entities/agent_config.tsx @@ -3,7 +3,7 @@ import { WithStringId } from "../../../lib/types/types"; import { WorkflowPrompt, WorkflowAgent, Workflow, WorkflowTool } from "../../../lib/types/workflow_types"; import { DataSource } from "../../../lib/types/datasource_types"; import { z } from "zod"; -import { PlusIcon, Sparkles, X as XIcon, ChevronDown, ChevronRight, Trash2, Maximize2, Minimize2, StarIcon, DatabaseIcon, UserIcon, Settings } from "lucide-react"; +import { PlusIcon, Sparkles, X as XIcon, ChevronDown, ChevronRight, Trash2, Maximize2, Minimize2, StarIcon, DatabaseIcon, UserIcon, Settings, Info } from "lucide-react"; import { useState, useEffect, useRef } from "react"; import { usePreviewModal } from "../workflow/preview-modal"; import { Modal, ModalContent, ModalHeader, ModalBody, ModalFooter, Select, SelectItem, Chip, SelectSection } from "@heroui/react"; @@ -19,7 +19,7 @@ import clsx from "clsx"; import { InputField } from "@/app/lib/components/input-field"; import { USE_TRANSFER_CONTROL_OPTIONS } from "@/app/lib/feature_flags"; import { Input } from "@/components/ui/input"; -import { Info } from "lucide-react"; +import { Info as InfoIcon } from "lucide-react"; import { useCopilot } from "../copilot/use-copilot"; import { BillingUpgradeModal } from "@/components/common/billing-upgrade-modal"; import { ModelsResponse } from "@/app/lib/types/billing_types"; @@ -39,6 +39,7 @@ export function AgentConfig({ workflow, agent, usedAgentNames, + usedPipelineNames, agents, tools, prompts, @@ -54,6 +55,7 @@ export function AgentConfig({ workflow: z.infer, agent: z.infer, usedAgentNames: Set, + usedPipelineNames: Set, agents: z.infer[], tools: z.infer[], prompts: z.infer[], @@ -78,6 +80,9 @@ export function AgentConfig({ const [billingError, setBillingError] = useState(null); const [showSavedBanner, setShowSavedBanner] = useState(false); + // Check if this agent is a pipeline agent + const isPipelineAgent = agent.type === 'pipeline'; + const { start: startCopilotChat, } = useCopilot({ @@ -117,14 +122,31 @@ export function AgentConfig({ setShowRagCta(false); }; - // Add effect to handle control type update when transfer control is disabled or when internal agents have invalid control type + // Add effect to handle control type update to ensure agents have correct control types useEffect(() => { - if (!USE_TRANSFER_CONTROL_OPTIONS && agent.controlType !== 'retain') { - handleUpdate({ ...agent, controlType: 'retain' }); + let correctControlType: "retain" | "relinquish_to_parent" | "relinquish_to_start" | undefined = undefined; + + // Determine the correct control type based on agent type and output visibility + if (agent.type === "pipeline") { + correctControlType = "relinquish_to_parent"; + } else if (agent.outputVisibility === "internal") { + correctControlType = "relinquish_to_parent"; + } else if (agent.outputVisibility === "user_facing") { + correctControlType = "retain"; } - // For internal agents, "retain" is not a valid option, so change it to "relinquish_to_parent" - if (agent.outputVisibility === "internal" && agent.controlType === 'retain') { - handleUpdate({ ...agent, controlType: 'relinquish_to_parent' }); + + // Handle undefined control type + if (agent.controlType === undefined) { + if (agent.outputVisibility === "user_facing") { + correctControlType = "retain"; + } else { + correctControlType = "relinquish_to_parent"; + } + } + + // Update if the control type is incorrect + if (correctControlType && agent.controlType !== correctControlType) { + handleUpdate({ ...agent, controlType: correctControlType }); } }, [agent.controlType, agent.outputVisibility, agent, handleUpdate]); @@ -151,7 +173,12 @@ export function AgentConfig({ return false; } if (value !== agent.name && usedAgentNames.has(value)) { - setNameError("This name is already taken"); + setNameError("This name is already taken by another agent"); + return false; + } + // Check for conflicts with pipeline names + if (usedPipelineNames.has(value)) { + setNameError("This name is already taken by a pipeline"); return false; } if (!/^[a-zA-Z0-9_-\s]+$/.test(value)) { @@ -175,10 +202,12 @@ export function AgentConfig({ }; const atMentions = createAtMentions({ - agents, + agents: agents, prompts, tools, - currentAgentName: agent.name + pipelines: agent.type === "pipeline" ? [] : (workflow.pipelines || []), // Pipeline agents can't reference pipelines + currentAgentName: agent.name, + currentAgent: agent }); // Add local state for max calls input @@ -211,7 +240,7 @@ export function AgentConfig({
{/* Saved Banner */} {showSavedBanner && ( -
+
@@ -246,7 +275,7 @@ export function AgentConfig({
{/* Saved Banner for maximized instructions */} {showSavedBanner && ( -
+
@@ -368,7 +397,7 @@ export function AgentConfig({
{/* Saved Banner for maximized examples */} {showSavedBanner && ( -
+
@@ -500,20 +529,30 @@ export function AgentConfig({
- { - handleUpdate({ - ...agent, - outputVisibility: value as z.infer["outputVisibility"] - }); - showSavedMessage(); - }} - /> + {isPipelineAgent ? ( + // For pipeline agents, show read-only display +
+ + Pipeline Agent + +
+ ) : ( + // For non-pipeline agents, show dropdown without pipeline option + { + handleUpdate({ + ...agent, + outputVisibility: value as z.infer["outputVisibility"] + }); + showSavedMessage(); + }} + /> + )}
@@ -585,7 +624,7 @@ export function AgentConfig({ }
- {agent.outputVisibility === "internal" && ( + {agent.outputVisibility === "internal" && !isPipelineAgent && (
@@ -622,14 +661,18 @@ export function AgentConfig({
)} - {USE_TRANSFER_CONTROL_OPTIONS && ( + {USE_TRANSFER_CONTROL_OPTIONS && !isPipelineAgent && (
, + pipeline: z.infer, + usedPipelineNames: Set, + usedAgentNames: Set, + agents: z.infer[], + pipelines: z.infer[], + handleUpdate: (pipeline: z.infer) => void, + handleClose: () => void, +}) { + const [localName, setLocalName] = useState(pipeline.name); + const [nameError, setNameError] = useState(null); + const [showSavedBanner, setShowSavedBanner] = useState(false); + + // Function to show saved banner + const showSavedMessage = () => { + setShowSavedBanner(true); + setTimeout(() => setShowSavedBanner(false), 2000); + }; + + useEffect(() => { + setLocalName(pipeline.name); + }, [pipeline.name]); + + const validateName = (value: string) => { + if (value.length === 0) { + setNameError("Name cannot be empty"); + return false; + } + // Check for conflicts with other pipeline names + if (value !== pipeline.name && usedPipelineNames.has(value)) { + setNameError("This name is already taken by another pipeline"); + return false; + } + // Check for conflicts with agent names + if (usedAgentNames.has(value)) { + setNameError("This name is already taken by an agent"); + return false; + } + if (!/^[a-zA-Z0-9_-\s]+$/.test(value)) { + setNameError("Name must contain only letters, numbers, underscores, hyphens, and spaces"); + return false; + } + setNameError(null); + return true; + }; + + const handleNameChange = (value: string) => { + setLocalName(value); + + if (validateName(value)) { + handleUpdate({ + ...pipeline, + name: value + }); + } + showSavedMessage(); + }; + + return ( + +
+ {pipeline.name} +
+ + + +
+ } + > +
+ {/* Saved Banner */} + {showSavedBanner && ( +
+ + + + Changes saved +
+ )} + + {/* Pipeline Configuration */} +
+ {/* Identity Section Card */} + } + title="Identity" + labelWidth="md:w-32" + className="mb-1" + > +
+
+ +
+ +
+
+
+ +
+ { + handleUpdate({ ...pipeline, description: value }); + showSavedMessage(); + }} + multiline={true} + placeholder="Enter a description for this pipeline" + className="w-full" + /> +
+
+
+ + + {/* Pipeline Info */} + } + title="Behavior" + labelWidth="md:w-32" + className="mb-1" + > +
+
+
+ Agents in Pipeline: {pipeline.agents.length} +
+
+
How Pipelines Work:
+
    +
  • Agents execute sequentially in the order shown
    • +
  • Output from one agent flows as input to the next
    • +
  • Add agents to this pipeline from the agents panel
    • +
+
+
+
+ +
+
+ + ); +} \ No newline at end of file diff --git a/apps/rowboat/app/projects/[projectId]/entities/prompt_config.tsx b/apps/rowboat/app/projects/[projectId]/entities/prompt_config.tsx index a3ccdf9a..159bd7b9 100644 --- a/apps/rowboat/app/projects/[projectId]/entities/prompt_config.tsx +++ b/apps/rowboat/app/projects/[projectId]/entities/prompt_config.tsx @@ -79,7 +79,7 @@ export function PromptConfig({
{/* Saved Banner */} {showSavedBanner && ( -
+
diff --git a/apps/rowboat/app/projects/[projectId]/entities/tool_config.tsx b/apps/rowboat/app/projects/[projectId]/entities/tool_config.tsx index eab56e3d..a3822d93 100644 --- a/apps/rowboat/app/projects/[projectId]/entities/tool_config.tsx +++ b/apps/rowboat/app/projects/[projectId]/entities/tool_config.tsx @@ -358,7 +358,7 @@ export function ToolConfig({
{/* Saved Banner */} {showSavedBanner && ( -
+
diff --git a/apps/rowboat/app/projects/[projectId]/playground/components/messages.tsx b/apps/rowboat/app/projects/[projectId]/playground/components/messages.tsx index b0c576ae..80c10577 100644 --- a/apps/rowboat/app/projects/[projectId]/playground/components/messages.tsx +++ b/apps/rowboat/app/projects/[projectId]/playground/components/messages.tsx @@ -89,20 +89,21 @@ function InternalAssistantMessage({ content, sender, latency, delta, showJsonMod {sender ?? 'Assistant'} {(Boolean(showDebugMessages && typeof onFix === 'function' && !isFirstAssistant) || Boolean(showDebugMessages && typeof onExplain === 'function' && !isFirstAssistant) - || Boolean(isJsonContent && hasResponseKey)) && ( + || Boolean(isJsonContent)) && ( onFix(content, index) : () => {}} onExplain={onExplain ? () => onExplain('assistant', content, index) : () => {}} - onJson={() => {}} + onJson={() => setJsonMode(!jsonMode)} + jsonLabel={jsonMode ? 'View formatted content' : 'View complete JSON'} /> )}
- {isJsonContent && hasResponseKey && jsonMode && ( + {isJsonContent && jsonMode && (
)} - {isJsonContent && hasResponseKey && jsonMode ? ( + {isJsonContent && jsonMode ? ( 
 a.name === message.agentName);
+            const isInternalOrPipeline = message.responseType === 'internal' || 
+                                       (agentConfig && (agentConfig.outputVisibility === 'internal' || agentConfig.type === 'pipeline'));
+            
+            if (message.content && isInternalOrPipeline) {
+                // Skip internal/pipeline messages if debug mode is off
                 if (!showDebugMessages) {
                     return null;
                 }
diff --git a/apps/rowboat/app/projects/[projectId]/workflow/entity_list.tsx b/apps/rowboat/app/projects/[projectId]/workflow/entity_list.tsx
index db85b2fa..0aa1ab8c 100644
--- a/apps/rowboat/app/projects/[projectId]/workflow/entity_list.tsx
+++ b/apps/rowboat/app/projects/[projectId]/workflow/entity_list.tsx
@@ -1,6 +1,6 @@
 import React, { forwardRef, useImperativeHandle } from "react";
 import { z } from "zod";
-import { WorkflowPrompt, WorkflowAgent, WorkflowTool, Workflow } from "../../../lib/types/workflow_types";
+import { WorkflowPrompt, WorkflowAgent, WorkflowTool, WorkflowPipeline, Workflow } from "../../../lib/types/workflow_types";
 import { Project } from "../../../lib/types/project_types";
 import { DataSource } from "../../../lib/types/datasource_types";
 import { WithStringId } from "../../../lib/types/types";
@@ -46,25 +46,30 @@ interface EntityListProps {
     agents: z.infer[];
     tools: z.infer[];
     prompts: z.infer[];
+    pipelines: z.infer[];
     dataSources: WithStringId>[];
     workflow: z.infer;
     selectedEntity: {
-        type: "agent" | "tool" | "prompt" | "datasource" | "visualise";
+        type: "agent" | "tool" | "prompt" | "datasource" | "pipeline" | "visualise";
         name: string;
     } | null;
     startAgentName: string | null;
     onSelectAgent: (name: string) => void;
     onSelectTool: (name: string) => void;
     onSelectPrompt: (name: string) => void;
+    onSelectPipeline: (name: string) => void;
     onSelectDataSource?: (id: string) => void;
     onAddAgent: (agent: Partial>) => void;
     onAddTool: (tool: Partial>) => void;
     onAddPrompt: (prompt: Partial>) => void;
+    onAddPipeline: (pipeline: Partial>) => void;
+    onAddAgentToPipeline: (pipelineName: string) => void;
     onToggleAgent: (name: string) => void;
     onSetMainAgent: (name: string) => void;
     onDeleteAgent: (name: string) => void;
     onDeleteTool: (name: string) => void;
     onDeletePrompt: (name: string) => void;
+    onDeletePipeline: (name: string) => void;
     onShowVisualise: (name: string) => void;
     onProjectToolsUpdated?: () => void;
     onDataSourcesUpdated?: () => void;
@@ -72,6 +77,7 @@ interface EntityListProps {
     useRagUploads: boolean;
     useRagS3Uploads: boolean;
     useRagScraping: boolean;
+    onReorderPipelines: (pipelines: z.infer[]) => void;
 }
 
 interface EmptyStateProps {
@@ -174,7 +180,7 @@ interface ServerCardProps {
     serverName: string;
     tools: z.infer[];
     selectedEntity: {
-        type: "agent" | "tool" | "prompt" | "datasource" | "visualise";
+        type: "agent" | "tool" | "prompt" | "datasource" | "pipeline" | "visualise";
         name: string;
     } | null;
     onSelectTool: (name: string) => void;
@@ -258,6 +264,163 @@ type ComposioToolkit = {
     tools: z.infer[];
 }
 
+interface PipelineCardProps {
+    pipeline: z.infer;
+    agents: z.infer[];
+    selectedEntity: {
+        type: "agent" | "tool" | "prompt" | "datasource" | "pipeline" | "visualise";
+        name: string;
+    } | null;
+    onSelectPipeline: (name: string) => void;
+    onSelectAgent: (name: string) => void;
+    onDeletePipeline: (name: string) => void;
+    onDeleteAgent: (name: string) => void;
+    onAddAgentToPipeline: (pipelineName: string) => void;
+    selectedRef: React.RefObject;
+    startAgentName: string | null;
+    dragHandle?: React.ReactNode;
+}
+
+const PipelineCard = ({
+    pipeline,
+    agents,
+    selectedEntity,
+    onSelectPipeline,
+    onSelectAgent,
+    onDeletePipeline,
+    onDeleteAgent,
+    onAddAgentToPipeline,
+    selectedRef,
+    startAgentName,
+    dragHandle,
+}: PipelineCardProps) => {
+    // Get agents that belong to this pipeline
+    const pipelineAgents = pipeline.agents
+        .map(agentName => agents.find(agent => agent.name === agentName))
+        .filter(Boolean) as z.infer[];
+
+    // Check if any agent in this pipeline is currently selected
+    const hasSelectedAgent = selectedEntity?.type === "agent" && 
+        pipeline.agents.includes(selectedEntity.name);
+
+    // Track expansion state - allow manual override even when agent is selected
+    const [isExpanded, setIsExpanded] = useState(false);
+    const [lastSelectedAgent, setLastSelectedAgent] = useState(null);
+
+    // Auto-expand when a new agent in this pipeline is selected
+    useEffect(() => {
+        if (hasSelectedAgent && selectedEntity?.name !== lastSelectedAgent) {
+            setIsExpanded(true);
+            setLastSelectedAgent(selectedEntity?.name || null);
+        } else if (!hasSelectedAgent) {
+            setLastSelectedAgent(null);
+        }
+    }, [hasSelectedAgent, selectedEntity?.name, lastSelectedAgent]);
+
+    return (
+        

+            

+                {dragHandle}
+                {/* Chevron button for expand/collapse - only show when has agents and on hover */}
+                
+                
+                {/* Pipeline name button for configuration */}
+                
+                
+                {/* Pipeline menu */}
+                

+                    
+                        
+                            
+                        
+                         {
+                                if (key === 'delete') {
+                                    onDeletePipeline(pipeline.name);
+                                }
+                            }}
+                        >
+                            Delete Pipeline
+                        
+                    
+                

+            

+            
+            {isExpanded && (
+                

+                    {pipelineAgents.map((agent, index) => (
+                        

+                            
 onSelectAgent(agent.name)}>
+                                

+                                    
+                                        {index + 1}
+                                    
+                                

+                                {agent.name}
+                                {startAgentName === agent.name && (
+                                    

+                                        Start
+                                    

+                                )}
+                                

+                                    
+                                

+                            

+                        

+                    ))}
+                    {/* Add Agent option */}
+                    
+                

+            )}
+        

+    );
+};
+
 export const EntityList = forwardRef<
     { openDataSourcesModal: () => void },
     EntityListProps & { 
@@ -268,6 +431,7 @@ export const EntityList = forwardRef<
     agents,
     tools,
     prompts,
+    pipelines,
     dataSources,
     workflow,
     selectedEntity,
@@ -275,27 +439,33 @@ export const EntityList = forwardRef<
     onSelectAgent,
     onSelectTool,
     onSelectPrompt,
+    onSelectPipeline,
     onSelectDataSource,
     onAddAgent,
     onAddTool,
     onAddPrompt,
+    onAddPipeline,
+    onAddAgentToPipeline,
     onToggleAgent,
     onSetMainAgent,
     onDeleteAgent,
     onDeleteTool,
     onDeletePrompt,
+    onDeletePipeline,
     onProjectToolsUpdated,
     onDataSourcesUpdated,
     projectId,
     projectConfig,
     onReorderAgents,
+    onReorderPipelines,
     onShowVisualise,
     useRagUploads,
     useRagS3Uploads,
     useRagScraping,
 }: EntityListProps & { 
     projectId: string,
-    onReorderAgents: (agents: z.infer[]) => void 
+    onReorderAgents: (agents: z.infer[]) => void,
+    onReorderPipelines: (pipelines: z.infer[]) => void 
 }, ref) {
     const [showAgentTypeModal, setShowAgentTypeModal] = useState(false);
     const [showToolsModal, setShowToolsModal] = useState(false);
@@ -417,20 +587,44 @@ export const EntityList = forwardRef<
         const { active, over } = event;
         
         if (over && active.id !== over.id) {
-            const oldIndex = agents.findIndex(agent => agent.name === active.id);
-            const newIndex = agents.findIndex(agent => agent.name === over.id);
+            // Determine if we're dragging a pipeline or an agent
+            const isPipelineDrag = pipelines.some(pipeline => pipeline.name === active.id);
+            const isPipelineTarget = pipelines.some(pipeline => pipeline.name === over.id);
             
-            const newAgents = [...agents];
-            const [movedAgent] = newAgents.splice(oldIndex, 1);
-            newAgents.splice(newIndex, 0, movedAgent);
-            
-            // Update order numbers
-            const updatedAgents = newAgents.map((agent, index) => ({
-                ...agent,
-                order: index * 100
-            }));
-            
-            onReorderAgents(updatedAgents);
+            if (isPipelineDrag && isPipelineTarget) {
+                // Reordering pipelines
+                const oldIndex = pipelines.findIndex(pipeline => pipeline.name === active.id);
+                const newIndex = pipelines.findIndex(pipeline => pipeline.name === over.id);
+                
+                const newPipelines = [...pipelines];
+                const [movedPipeline] = newPipelines.splice(oldIndex, 1);
+                newPipelines.splice(newIndex, 0, movedPipeline);
+                
+                // Update order numbers
+                const updatedPipelines = newPipelines.map((pipeline, index) => ({
+                    ...pipeline,
+                    order: index * 100
+                }));
+                
+                onReorderPipelines(updatedPipelines);
+            } else if (!isPipelineDrag && !isPipelineTarget) {
+                // Reordering individual agents (not in pipelines)
+                const oldIndex = agents.findIndex(agent => agent.name === active.id);
+                const newIndex = agents.findIndex(agent => agent.name === over.id);
+                
+                const newAgents = [...agents];
+                const [movedAgent] = newAgents.splice(oldIndex, 1);
+                newAgents.splice(newIndex, 0, movedAgent);
+                
+                // Update order numbers
+                const updatedAgents = newAgents.map((agent, index) => ({
+                    ...agent,
+                    order: index * 100
+                }));
+                
+                onReorderAgents(updatedAgents);
+            }
+            // Note: We don't allow dragging between pipelines and agents
         }
     };
 
@@ -509,36 +703,83 @@ export const EntityList = forwardRef<
                         {expandedPanels.agents && (
                             

                                 

-                                    {agents.length > 0 ? (
-                                        
-                                             a.name)}
-                                                strategy={verticalListSortingStrategy}
-                                            >
-                                                

-                                                    {agents.map((agent) => (
-                                                         onSelectAgent(agent.name)}
-                                                            selectedRef={selectedEntity?.type === "agent" && selectedEntity.name === agent.name ? selectedRef : undefined}
-                                                            statusLabel={startAgentName === agent.name ?  : null}
-                                                            onToggle={onToggleAgent}
-                                                            onSetMainAgent={onSetMainAgent}
-                                                            onDelete={onDeleteAgent}
-                                                            isStartAgent={startAgentName === agent.name}
-                                                        />
-                                                    ))}
-                                                

-                                            
-                                        
+                                    {pipelines.length > 0 || agents.length > 0 ? (
+                                        

+                                            {/* Show pipelines first with drag-and-drop */}
+                                            {pipelines.length > 0 && (
+                                                
+                                                     p.name)}
+                                                        strategy={verticalListSortingStrategy}
+                                                    >
+                                                        {pipelines.map((pipeline) => (
+                                                            
+                                                        ))}
+                                                    
+                                                
+                                            )}
+                                            
+                                            {/* Show individual agents that are NOT part of any pipeline */}
+                                            {(() => {
+                                                // Get all agent names that are part of pipelines
+                                                const pipelineAgentNames = new Set(
+                                                    pipelines.flatMap(pipeline => pipeline.agents)
+                                                );
+                                                
+                                                // Filter agents that are not in any pipeline and are not pipeline agents
+                                                const individualAgents = agents.filter(
+                                                    agent => !pipelineAgentNames.has(agent.name) && agent.type !== 'pipeline'
+                                                );
+                                                
+                                                if (individualAgents.length === 0) return null;
+                                                
+                                                return (
+                                                    
+                                                         a.name)}
+                                                            strategy={verticalListSortingStrategy}
+                                                        >
+                                                            {individualAgents.map((agent) => (
+                                                                 onSelectAgent(agent.name)}
+                                                                    selectedRef={selectedEntity?.type === "agent" && selectedEntity.name === agent.name ? selectedRef : undefined}
+                                                                    statusLabel={startAgentName === agent.name ?  : null}
+                                                                    onToggle={onToggleAgent}
+                                                                    onSetMainAgent={onSetMainAgent}
+                                                                    onDelete={onDeleteAgent}
+                                                                    isStartAgent={startAgentName === agent.name}
+                                                                />
+                                                            ))}
+                                                        
+                                                    
+                                                );
+                                            })()}
+                                        

                                     ) : (
-                                        
+                                        
                                     )}
                                 

                             

@@ -925,6 +1166,10 @@ export const EntityList = forwardRef<
                 isOpen={showAgentTypeModal}
                 onClose={() => setShowAgentTypeModal(false)}
                 onConfirm={handleAddAgentWithType}
+                onCreatePipeline={() => {
+                    onAddPipeline({ name: `Pipeline ${pipelines.length + 1}` });
+                    setShowAgentTypeModal(false);
+                }}
             />
              void;
@@ -1210,7 +1455,7 @@ const ComposioCard = ({
                 

                     
+                }
+            />
+        

+    );
+};
+
 interface AgentTypeModalProps {
     isOpen: boolean;
     onClose: () => void;
     onConfirm: (agentType: 'internal' | 'user_facing') => void;
+    onCreatePipeline: () => void;
 }
 
-function AgentTypeModal({ isOpen, onClose, onConfirm }: AgentTypeModalProps) {
-    const [selectedType, setSelectedType] = useState<'internal' | 'user_facing'>('internal');
+function AgentTypeModal({ isOpen, onClose, onConfirm, onCreatePipeline }: AgentTypeModalProps) {
+    const [selectedType, setSelectedType] = useState<'internal' | 'user_facing' | 'pipeline'>('internal');
 
     const handleConfirm = () => {
-        onConfirm(selectedType);
+        if (selectedType === 'pipeline') {
+            onCreatePipeline();
+        } else {
+            onConfirm(selectedType);
+        }
         onClose();
     };
 
     return (
-        
-            
+        
+            
                 
                     

                         
-                        Create New Agent
+                        Create New Agent or Pipeline
                     

                 
-                
-                    

+                
+                    

                         

-                            Choose the type of agent you want to create:
+                            Choose what you want to create:
                         

-                        

+                        

                             {/* Task Agent (Internal) */}
                             
+
+                            {/* Pipeline */}
+                            
                         

                     

                 
-                
+                
                     
                 
             
diff --git a/apps/rowboat/app/projects/[projectId]/workflow/workflow_editor.tsx b/apps/rowboat/app/projects/[projectId]/workflow/workflow_editor.tsx
index 816a9292..55da3bc6 100644
--- a/apps/rowboat/app/projects/[projectId]/workflow/workflow_editor.tsx
+++ b/apps/rowboat/app/projects/[projectId]/workflow/workflow_editor.tsx
@@ -1,11 +1,12 @@
 "use client";
 import React, { useReducer, Reducer, useState, useCallback, useEffect, useRef, createContext, useContext } from "react";
 import { MCPServer, Message, WithStringId } from "../../../lib/types/types";
-import { Workflow, WorkflowTool, WorkflowPrompt, WorkflowAgent } from "../../../lib/types/workflow_types";
+import { Workflow, WorkflowTool, WorkflowPrompt, WorkflowAgent, WorkflowPipeline } from "../../../lib/types/workflow_types";
 import { DataSource } from "../../../lib/types/datasource_types";
 import { Project } from "../../../lib/types/project_types";
 import { produce, applyPatches, enablePatches, produceWithPatches, Patch } from 'immer';
 import { AgentConfig } from "../entities/agent_config";
+import { PipelineConfig } from "../entities/pipeline_config";
 import { ToolConfig } from "../entities/tool_config";
 import { App as ChatApp } from "../playground/app";
 import { z } from "zod";
@@ -25,7 +26,7 @@ import { publishWorkflow } from "@/app/actions/project_actions";
 import { saveWorkflow } from "@/app/actions/project_actions";
 import { updateProjectName } from "@/app/actions/project_actions";
 import { BackIcon, HamburgerIcon, WorkflowIcon } from "../../../lib/components/icons";
-import { CopyIcon, ImportIcon, Layers2Icon, RadioIcon, RedoIcon, ServerIcon, Sparkles, UndoIcon, RocketIcon, PenLine, AlertTriangle, DownloadIcon, XIcon, SettingsIcon, ChevronDownIcon, PhoneIcon, MessageCircleIcon } from "lucide-react";
+import { CopyIcon, ImportIcon, RadioIcon, RedoIcon, ServerIcon, Sparkles, UndoIcon, RocketIcon, PenLine, AlertTriangle, DownloadIcon, XIcon, SettingsIcon, ChevronDownIcon, PhoneIcon, MessageCircleIcon } from "lucide-react";
 import { EntityList } from "./entity_list";
 import { ProductTour } from "@/components/common/product-tour";
 import { ModelsResponse } from "@/app/lib/types/billing_types";
@@ -49,7 +50,7 @@ interface StateItem {
     workflow: z.infer;
     publishing: boolean;
     selection: {
-        type: "agent" | "tool" | "prompt" | "datasource" | "visualise";
+        type: "agent" | "tool" | "prompt" | "datasource" | "pipeline" | "visualise";
         name: string;
     } | null;
     saving: boolean;
@@ -83,18 +84,31 @@ export type Action = {
 } | {
     type: "add_prompt";
     prompt: Partial>;
+} | {
+    type: "add_pipeline";
+    pipeline: Partial>;
 } | {
     type: "select_agent";
     name: string;
 } | {
     type: "select_tool";
     name: string;
+} | {
+    type: "select_pipeline";
+    name: string;
 } | {
     type: "delete_agent";
     name: string;
 } | {
     type: "delete_tool";
     name: string;
+} | {
+    type: "delete_pipeline";
+    name: string;
+} | {
+    type: "update_pipeline";
+    name: string;
+    pipeline: Partial>;
 } | {
     type: "update_agent";
     name: string;
@@ -119,6 +133,8 @@ export type Action = {
     name: string;
 } | {
     type: "unselect_prompt";
+} | {
+    type: "unselect_pipeline";
 } | {
     type: "delete_prompt";
     name: string;
@@ -144,6 +160,9 @@ export type Action = {
 } | {
     type: "reorder_agents";
     agents: z.infer[];
+} | {
+    type: "reorder_pipelines";
+    pipelines: z.infer[];
 } | {
     type: "select_datasource";
     id: string;
@@ -235,6 +254,23 @@ function reducer(state: State, action: Action): State {
                 currentIndex: state.currentIndex + 1,
             };
         }
+        case "reorder_pipelines": {
+            const newState = produce(state.present, draft => {
+                draft.workflow.pipelines = action.pipelines;
+                draft.lastUpdatedAt = new Date().toISOString();
+            });
+            const [nextState, patches, inversePatches] = produceWithPatches(state.present, draft => {
+                draft.workflow.pipelines = action.pipelines;
+                draft.lastUpdatedAt = new Date().toISOString();
+            });
+            return {
+                ...state,
+                present: nextState,
+                patches: [...state.patches.slice(0, state.currentIndex), patches],
+                inversePatches: [...state.inversePatches.slice(0, state.currentIndex), inversePatches],
+                currentIndex: state.currentIndex + 1,
+            };
+        }
         case "show_visualise": {
             newState = produce(state, draft => {
                 draft.present.selection = { type: "visualise", name: "visualise" };
@@ -270,6 +306,12 @@ function reducer(state: State, action: Action): State {
                                 name: action.name
                             };
                             break;
+                        case "select_pipeline":
+                            draft.selection = {
+                                type: "pipeline",
+                                name: action.name
+                            };
+                            break;
                         case "select_datasource":
                             draft.selection = {
                                 type: "datasource",
@@ -280,6 +322,7 @@ function reducer(state: State, action: Action): State {
                         case "unselect_tool":
                         case "unselect_prompt":
                         case "unselect_datasource":
+                        case "unselect_pipeline":
                             draft.selection = null;
                             break;
                         case "add_agent": {
@@ -366,13 +409,97 @@ function reducer(state: State, action: Action): State {
                             draft.chatKey++;
                             break;
                         }
+                        case "add_pipeline": {
+                            if (isLive) {
+                                break;
+                            }
+                            let newPipelineName = "New pipeline";
+                            if (draft.workflow?.pipelines?.some((pipeline) => pipeline.name === newPipelineName)) {
+                                newPipelineName = `New pipeline ${(draft.workflow?.pipelines?.filter((pipeline) =>
+                                    pipeline.name.startsWith("New pipeline")).length || 0) + 1}`;
+                            }
+                            if (!draft.workflow.pipelines) {
+                                draft.workflow.pipelines = [];
+                            }
+                            
+                            // Create the first agent for this pipeline
+                            const firstAgentName = `${action.pipeline.name || newPipelineName} Step 1`;
+                            draft.workflow.agents.push({
+                                name: firstAgentName,
+                                type: "pipeline",
+                                description: "",
+                                disabled: false,
+                                instructions: "",
+                                model: "gpt-4o",
+                                locked: false,
+                                toggleAble: true,
+                                ragReturnType: "chunks",
+                                ragK: 3,
+                                controlType: "relinquish_to_parent",
+                                outputVisibility: "internal",
+                                maxCallsPerParentAgent: 3,
+                            });
+                            
+                            // Create the pipeline with the first agent
+                            draft.workflow.pipelines.push({
+                                name: newPipelineName,
+                                description: "",
+                                agents: [firstAgentName],
+                                ...action.pipeline
+                            });
+                            
+                            // Select the newly created agent to open it in agent_config
+                            draft.selection = {
+                                type: "agent",
+                                name: firstAgentName
+                            };
+                            draft.pendingChanges = true;
+                            draft.chatKey++;
+                            break;
+                        }
                         case "delete_agent":
                             if (isLive) {
                                 break;
                             }
+                            // Remove the agent
                             draft.workflow.agents = draft.workflow.agents.filter(
                                 (agent) => agent.name !== action.name
                             );
+                            
+                            // Update references to deleted agent in other agents' instructions
+                            draft.workflow.agents = draft.workflow.agents.map(agent => ({
+                                ...agent,
+                                instructions: agent.instructions.replace(
+                                    new RegExp(`\\[@agent:${action.name}\\]\\(#mention\\)`, 'g'),
+                                    ''
+                                )
+                            }));
+                            
+                            // Update references in prompts
+                            draft.workflow.prompts = draft.workflow.prompts.map(prompt => ({
+                                ...prompt,
+                                prompt: prompt.prompt.replace(
+                                    new RegExp(`\\[@agent:${action.name}\\]\\(#mention\\)`, 'g'),
+                                    ''
+                                )
+                            }));
+                            
+                            // Update references in pipelines
+                            if (draft.workflow.pipelines) {
+                                draft.workflow.pipelines = draft.workflow.pipelines.map(pipeline => ({
+                                    ...pipeline,
+                                    agents: pipeline.agents.filter(agentName => agentName !== action.name)
+                                }));
+                            }
+                            
+                            // Update start agent if it was the deleted agent
+                            if (draft.workflow.startAgent === action.name) {
+                                // Set to first available agent, or empty string if no agents left
+                                draft.workflow.startAgent = draft.workflow.agents.length > 0 
+                                    ? draft.workflow.agents[0].name 
+                                    : '';
+                            }
+                            
                             draft.selection = null;
                             draft.pendingChanges = true;
                             draft.chatKey++;
@@ -399,7 +526,80 @@ function reducer(state: State, action: Action): State {
                             draft.pendingChanges = true;
                             draft.chatKey++;
                             break;
-                        case "update_agent":
+                        case "delete_pipeline":
+                            if (isLive) {
+                                break;
+                            }
+                            if (draft.workflow.pipelines) {
+                                // Find the pipeline to get its associated agents
+                                const pipelineToDelete = draft.workflow.pipelines.find(
+                                    (pipeline) => pipeline.name === action.name
+                                );
+                                
+                                if (pipelineToDelete) {
+                                    // Remove all agents that belong to this pipeline
+                                    const agentsToDelete = pipelineToDelete.agents || [];
+                                    
+                                    // Check if startAgent is one of the agents being deleted
+                                    const startAgentBeingDeleted = agentsToDelete.includes(draft.workflow.startAgent);
+                                    
+                                    draft.workflow.agents = draft.workflow.agents.filter(
+                                        (agent) => !agentsToDelete.includes(agent.name)
+                                    );
+                                    
+                                    // Update references to deleted agents in other agents' instructions
+                                    agentsToDelete.forEach(agentName => {
+                                        draft.workflow.agents = draft.workflow.agents.map(agent => ({
+                                            ...agent,
+                                            instructions: agent.instructions.replace(
+                                                new RegExp(`\\[@agent:${agentName}\\]\\(#mention\\)`, 'g'),
+                                                ''
+                                            )
+                                        }));
+                                        
+                                        // Update references in prompts
+                                        draft.workflow.prompts = draft.workflow.prompts.map(prompt => ({
+                                            ...prompt,
+                                            prompt: prompt.prompt.replace(
+                                                new RegExp(`\\[@agent:${agentName}\\]\\(#mention\\)`, 'g'),
+                                                ''
+                                            )
+                                        }));
+                                    });
+                                    
+                                    // Update start agent if it was one of the deleted agents (same logic as regular agent deletion)
+                                    if (startAgentBeingDeleted) {
+                                        // Set to first available agent, or empty string if no agents left
+                                        draft.workflow.startAgent = draft.workflow.agents.length > 0 
+                                            ? draft.workflow.agents[0].name 
+                                            : '';
+                                    }
+                                }
+                                
+                                // Remove the pipeline itself
+                                draft.workflow.pipelines = draft.workflow.pipelines.filter(
+                                    (pipeline) => pipeline.name !== action.name
+                                );
+                            }
+                            draft.selection = null;
+                            draft.pendingChanges = true;
+                            draft.chatKey++;
+                            break;
+                        case "update_pipeline": {
+                            if (isLive) {
+                                break;
+                            }
+                            if (draft.workflow.pipelines) {
+                                draft.workflow.pipelines = draft.workflow.pipelines.map(pipeline =>
+                                    pipeline.name === action.name ? { ...pipeline, ...action.pipeline } : pipeline
+                                );
+                            }
+                            draft.selection = null;
+                            draft.pendingChanges = true;
+                            draft.chatKey++;
+                            break;
+                        }
+                        case "update_agent": {
                             if (isLive) {
                                 break;
                             }
@@ -432,6 +632,16 @@ function reducer(state: State, action: Action): State {
                                     )
                                 }));
 
+                                // update pipeline references if this agent is part of any pipeline
+                                if (draft.workflow.pipelines) {
+                                    draft.workflow.pipelines = draft.workflow.pipelines.map(pipeline => ({
+                                        ...pipeline,
+                                        agents: pipeline.agents.map(agentName => 
+                                            agentName === action.name ? action.agent.name! : agentName
+                                        )
+                                    }));
+                                }
+
                                 // update the selection pointer if this is the selected agent
                                 if (draft.selection?.type === "agent" && draft.selection.name === action.name) {
                                     draft.selection = {
@@ -449,6 +659,7 @@ function reducer(state: State, action: Action): State {
                             draft.pendingChanges = true;
                             draft.chatKey++;
                             break;
+                        }
                         case "update_tool":
                             if (isLive) {
                                 break;
@@ -785,10 +996,59 @@ export function WorkflowEditor({
         dispatch({ type: "add_prompt", prompt });
     }
 
+    function handleSelectPipeline(name: string) {
+        dispatch({ type: "select_pipeline", name });
+    }
+
+    function handleAddPipeline(pipeline: Partial> = {}) {
+        dispatch({ type: "add_pipeline", pipeline });
+    }
+
+    function handleDeletePipeline(name: string) {
+        if (window.confirm(`Are you sure you want to delete the pipeline "${name}"?`)) {
+            dispatch({ type: "delete_pipeline", name });
+        }
+    }
+
+    function handleAddAgentToPipeline(pipelineName: string) {
+        // Create a pipeline agent and add it to the specified pipeline
+        const newAgentName = `${pipelineName} Step ${(state.present.workflow.pipelines?.find(p => p.name === pipelineName)?.agents.length || 0) + 1}`;
+        
+        const agentWithModel = {
+            name: newAgentName,
+            type: 'pipeline' as const,
+            outputVisibility: 'internal' as const,
+            model: defaultModel || "gpt-4o"
+        };
+        
+        // First add the agent
+        dispatch({ type: "add_agent", agent: agentWithModel });
+        
+        // Then add it to the pipeline
+        const pipeline = state.present.workflow.pipelines?.find(p => p.name === pipelineName);
+        if (pipeline) {
+            dispatch({ 
+                type: "update_pipeline", 
+                name: pipelineName, 
+                pipeline: { 
+                    ...pipeline, 
+                    agents: [...pipeline.agents, newAgentName] 
+                } 
+            });
+        }
+        
+        // Select the newly created agent to open it in agent_config
+        dispatch({ type: "select_agent", name: newAgentName });
+    }
+
     function handleUpdateAgent(name: string, agent: Partial>) {
         dispatch({ type: "update_agent", name, agent });
     }
 
+    function handleUpdatePipeline(name: string, pipeline: Partial>) {
+        dispatch({ type: "update_pipeline", name, pipeline });
+    }
+
     function handleDeleteAgent(name: string) {
         if (window.confirm(`Are you sure you want to delete the agent "${name}"?`)) {
             dispatch({ type: "delete_agent", name });
@@ -835,6 +1095,18 @@ export function WorkflowEditor({
         dispatch({ type: "reorder_agents", agents });
     }
 
+    function handleReorderPipelines(pipelines: z.infer[]) {
+        // Save order to localStorage
+        const orderMap = pipelines.reduce((acc, pipeline, index) => {
+            acc[pipeline.name] = index;
+            return acc;
+        }, {} as Record);
+        const mode = isLive ? 'live' : 'draft';
+        localStorage.setItem(`${mode}_workflow_${projectId}_pipeline_order`, JSON.stringify(orderMap));
+        
+        dispatch({ type: "reorder_pipelines", pipelines });
+    }
+
     async function handlePublishWorkflow() {
         await publishWorkflow(projectId, state.present.workflow);
         onChangeMode('live');
@@ -1110,6 +1382,7 @@ export function WorkflowEditor({
                                 agents={state.present.workflow.agents}
                                 tools={state.present.workflow.tools}
                                 prompts={state.present.workflow.prompts}
+                                pipelines={state.present.workflow.pipelines || []}
                                 dataSources={dataSources}
                                 workflow={state.present.workflow}
                                 selectedEntity={
@@ -1117,7 +1390,8 @@ export function WorkflowEditor({
                                     (state.present.selection.type === "agent" ||
                                      state.present.selection.type === "tool" ||
                                      state.present.selection.type === "prompt" ||
-                                     state.present.selection.type === "datasource")
+                                     state.present.selection.type === "datasource" ||
+                                     state.present.selection.type === "pipeline")
                                       ? state.present.selection
                                       : null
                                 }
@@ -1125,21 +1399,26 @@ export function WorkflowEditor({
                                 onSelectAgent={handleSelectAgent}
                                 onSelectTool={handleSelectTool}
                                 onSelectPrompt={handleSelectPrompt}
+                                onSelectPipeline={handleSelectPipeline}
                                 onSelectDataSource={handleSelectDataSource}
                                 onAddAgent={handleAddAgent}
                                 onAddTool={handleAddTool}
                                 onAddPrompt={handleAddPrompt}
+                                onAddPipeline={handleAddPipeline}
+                                onAddAgentToPipeline={handleAddAgentToPipeline}
                                 onToggleAgent={handleToggleAgent}
                                 onSetMainAgent={handleSetMainAgent}
                                 onDeleteAgent={handleDeleteAgent}
                                 onDeleteTool={handleDeleteTool}
                                 onDeletePrompt={handleDeletePrompt}
+                                onDeletePipeline={handleDeletePipeline}
                                 onShowVisualise={handleShowVisualise}
                                 projectId={projectId}
                                 onProjectToolsUpdated={onProjectToolsUpdated}
                                 onDataSourcesUpdated={onDataSourcesUpdated}
                                 projectConfig={projectConfig}
                                 onReorderAgents={handleReorderAgents}
+                                onReorderPipelines={handleReorderPipelines}
                                 useRagUploads={useRagUploads}
                                 useRagS3Uploads={useRagS3Uploads}
                                 useRagScraping={useRagScraping}
@@ -1168,6 +1447,7 @@ export function WorkflowEditor({
                             workflow={state.present.workflow}
                             agent={state.present.workflow.agents.find((agent) => agent.name === state.present.selection!.name)!}
                             usedAgentNames={new Set(state.present.workflow.agents.filter((agent) => agent.name !== state.present.selection!.name).map((agent) => agent.name))}
+                            usedPipelineNames={new Set((state.present.workflow.pipelines || []).map((pipeline) => pipeline.name))}
                             agents={state.present.workflow.agents}
                             tools={state.present.workflow.tools}
                             prompts={state.present.workflow.prompts}
@@ -1209,6 +1489,18 @@ export function WorkflowEditor({
                             handleClose={() => dispatch({ type: "unselect_datasource" })}
                             onDataSourceUpdate={onDataSourcesUpdated}
                         />}
+                        {state.present.selection?.type === "pipeline" &&  pipeline.name === state.present.selection!.name)!}
+                            usedPipelineNames={new Set((state.present.workflow.pipelines || []).filter((pipeline) => pipeline.name !== state.present.selection!.name).map((pipeline) => pipeline.name))}
+                            usedAgentNames={new Set(state.present.workflow.agents.map((agent) => agent.name))}
+                            agents={state.present.workflow.agents}
+                            pipelines={state.present.workflow.pipelines || []}
+                            handleUpdate={handleUpdatePipeline.bind(null, state.present.selection.name)}
+                            handleClose={() => dispatch({ type: "unselect_pipeline" })}
+                        />}
                         {state.present.selection?.type === "visualise" && (