diff --git a/app/api/v1/flows/route.ts b/app/api/v1/flows/route.ts index d1976ec..9e7a96d 100644 --- a/app/api/v1/flows/route.ts +++ b/app/api/v1/flows/route.ts @@ -20,10 +20,16 @@ export async function GET() { } export async function POST(request: Request) { - const { name } = await request.json() as { name?: string } + const { name, json } = await request.json() as { + name?: string + json?: { nodes: unknown[]; edges: unknown[] } + } const [flow] = await db .insert(flows) - .values({ name: name ?? 'Untitled flow', json: { nodes: [], edges: [] } }) + .values({ + name: name ?? 'Untitled flow', + json: json ?? { nodes: [], edges: [] }, + }) .returning() return Response.json(flow, { status: 201 }) } diff --git a/app/api/v1/vectors/route.ts b/app/api/v1/vectors/route.ts new file mode 100644 index 0000000..81b9306 --- /dev/null +++ b/app/api/v1/vectors/route.ts @@ -0,0 +1,106 @@ +// GET /api/v1/vectors?flowId=&nodeId=&limit= +// +// Lists rows in the `embeddings` table for a single VectorNode, scoped by +// (flowId, nodeId). Embeddings are streamed back with a small numeric preview +// (first 8 dims) instead of the full vector so payloads stay manageable for +// 1536-dim models. + +import { and, asc, eq } from 'drizzle-orm' +import { db } from '@/lib/db/client' +import { embeddings, vectorStores } from '@/lib/db/schema' + +export const runtime = 'nodejs' + +const DEFAULT_LIMIT = 50 +const MAX_LIMIT = 500 +const PREVIEW_DIMS = 8 + +const SCHEMA = [ + { column: 'id', type: 'uuid', note: 'primary key' }, + { column: 'store_id', type: 'uuid', note: 'FK → vector_stores.id' }, + { column: 'flow_id', type: 'uuid', note: 'FK → flows.id' }, + { column: 'chunk_index', type: 'integer', note: 'order within source document' }, + { column: 'content', type: 'text', note: 'raw chunk text' }, + { column: 'embedding', type: 'jsonb', note: 'number[] — model dimensions' }, + { column: 'metadata', type: 'jsonb', note: '{ source, page, … }' }, + { column: 'created_at', type: 'timestamp', note: 'insert time' }, +] as const + +const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i + +export async function GET(request: Request) { + const url = new URL(request.url) + const flowId = url.searchParams.get('flowId') + const nodeId = url.searchParams.get('nodeId') + const limit = Math.min(MAX_LIMIT, Math.max(1, Number(url.searchParams.get('limit') ?? DEFAULT_LIMIT))) + + if (!flowId || !nodeId) { + return Response.json({ error: 'flowId and nodeId are required' }, { status: 400 }) + } + // Drizzle/Postgres rejects non-UUID values on a uuid column with a low-level + // error. Treat unsaved/local flow ids as "no store yet" so the sidebar shows + // an empty state instead of bubbling a 500 to the client. + if (!UUID_RE.test(flowId)) { + return Response.json({ schema: SCHEMA, store: null, rows: [], total: 0 }) + } + + try { + const [store] = await db + .select() + .from(vectorStores) + .where(and(eq(vectorStores.flowId, flowId), eq(vectorStores.nodeId, nodeId))) + .limit(1) + + if (!store) { + return Response.json({ schema: SCHEMA, store: null, rows: [], total: 0 }) + } + + const rows = await db + .select({ + id: embeddings.id, + chunkIndex: embeddings.chunkIndex, + content: embeddings.content, + embedding: embeddings.embedding, + metadata: embeddings.metadata, + createdAt: embeddings.createdAt, + }) + .from(embeddings) + .where(eq(embeddings.storeId, store.id)) + .orderBy(asc(embeddings.chunkIndex)) + .limit(limit) + + const shaped = rows.map((r) => { + const vec = (r.embedding as unknown as number[]) ?? [] + return { + id: r.id, + chunkIndex: r.chunkIndex, + content: r.content, + preview: vec.slice(0, PREVIEW_DIMS), + dims: vec.length, + metadata: r.metadata, + createdAt: r.createdAt, + } + }) + + return Response.json({ + schema: SCHEMA, + store: { + id: store.id, + name: store.name, + provider: store.provider, + model: store.model, + dimensions: store.dimensions, + indexType: store.indexType, + metric: store.metric, + createdAt: store.createdAt, + }, + rows: shaped, + total: shaped.length, + limit, + }) + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + console.error('[vectors] list failed', err) + return Response.json({ error: message, schema: SCHEMA, store: null, rows: [], total: 0 }, { status: 500 }) + } +} diff --git a/app/api/v1/workflows/clone-sample/route.ts b/app/api/v1/workflows/clone-sample/route.ts new file mode 100644 index 0000000..32283a3 --- /dev/null +++ b/app/api/v1/workflows/clone-sample/route.ts @@ -0,0 +1,26 @@ +import { db } from '@/lib/db/client' +import { flows } from '@/lib/db/schema' +import { getSampleWorkflow } from '@/lib/sampleWorkflows' + +export const runtime = 'nodejs' + +export async function POST(request: Request) { + const { sampleId } = await request.json() as { sampleId?: string } + if (!sampleId) { + return Response.json({ error: 'sampleId is required' }, { status: 400 }) + } + const sample = getSampleWorkflow(sampleId) + if (!sample) { + return Response.json({ error: `Unknown sample: ${sampleId}` }, { status: 404 }) + } + + const [created] = await db + .insert(flows) + .values({ + name: sample.name, + json: { nodes: sample.graph.nodes, edges: sample.graph.edges }, + }) + .returning() + + return Response.json(created, { status: 201 }) +} diff --git a/app/canvas/page.tsx b/app/canvas/page.tsx index 1150668..0f78129 100644 --- a/app/canvas/page.tsx +++ b/app/canvas/page.tsx @@ -1,6 +1,7 @@ 'use client' -import { useEffect, useState } from 'react' +import { Suspense, useEffect, useState } from 'react' +import { useSearchParams } from 'next/navigation' import Canvas from '@/components/canvas/Canvas' import Toolbar from '@/components/canvas/Toolbar' import NodeSidebar from '@/components/canvas/NodeSidebar' @@ -14,26 +15,45 @@ interface FlowRow { createdAt: string } -export default function CanvasPage() { +function CanvasInner() { + const params = useSearchParams() + const requestedId = params.get('flowId') const [flowId, setFlowId] = useState(null) const [flowName, setFlowName] = useState() - const loadGraph = useStore((s) => s.loadGraph) + const loadGraph = useStore((s) => s.loadGraph) + const setStoreFlow = useStore((s) => s.setFlowId) useFlowPersist(flowId) useEffect(() => { - fetch('/api/v1/flows') - .then((r) => r.json()) - .then((rows: FlowRow[]) => { - const flow = rows[0] - if (!flow) return + async function load() { + if (requestedId) { + const r = await fetch(`/api/v1/flows/${requestedId}`) + if (!r.ok) { + console.error('Failed to load flow', requestedId, r.status) + return + } + const flow = await r.json() as FlowRow setFlowId(flow.id) + setStoreFlow(flow.id) setFlowName(flow.name) // eslint-disable-next-line @typescript-eslint/no-explicit-any loadGraph(flow.json.nodes as any, flow.json.edges as any) - }) - .catch(console.error) - }, [loadGraph]) + return + } + + const r = await fetch('/api/v1/flows') + const rows = await r.json() as FlowRow[] + const flow = rows[0] + if (!flow) return + setFlowId(flow.id) + setStoreFlow(flow.id) + setFlowName(flow.name) + // eslint-disable-next-line @typescript-eslint/no-explicit-any + loadGraph(flow.json.nodes as any, flow.json.edges as any) + } + load().catch(console.error) + }, [requestedId, loadGraph, setStoreFlow]) if (!flowId) { return ( @@ -53,3 +73,17 @@ export default function CanvasPage() { ) } + +export default function CanvasPage() { + return ( + +
+
+ } + > + + + ) +} diff --git a/app/saved-workflows/page.tsx b/app/saved-workflows/page.tsx index 2db0704..ac00f72 100644 --- a/app/saved-workflows/page.tsx +++ b/app/saved-workflows/page.tsx @@ -1,81 +1,138 @@ import Link from 'next/link' -import { getSession } from '@/lib/auth' +import { desc } from 'drizzle-orm' import { redirect } from 'next/navigation' +import { getSession } from '@/lib/auth' import { db } from '@/lib/db/client' import { flows } from '@/lib/db/schema' +import { SAMPLE_WORKFLOWS } from '@/lib/sampleWorkflows' +import WorkflowCard from '@/components/workflows/WorkflowCard' + +interface FlowJson { + nodes?: { type?: string; data?: { nodeType?: string } }[] + edges?: unknown[] +} + +const ANNOTATION_TYPES = new Set(['shape', 'text', 'drawing', 'arrow']) + +function summarize(json: unknown): { nodeCount: number; nodeTypes: string[] } { + const j = (json ?? {}) as FlowJson + const nodes = Array.isArray(j.nodes) ? j.nodes : [] + const runnable = nodes.filter((n) => !ANNOTATION_TYPES.has((n.type ?? '') as string)) + const nodeTypes = runnable + .map((n) => (n.data?.nodeType ?? n.type ?? '') as string) + .filter(Boolean) + return { nodeCount: runnable.length, nodeTypes } +} export default async function SavedWorkflowsPage() { const session = await getSession() if (!session) redirect('/') - // Fetch all flows (schema doesn't have user-level filtering yet) - const userFlows = await db.select().from(flows) + const userFlows = await db.select().from(flows).orderBy(desc(flows.createdAt)) return (
-
+
-
-
-

Saved Workflows

- +
+
+
+

Saved Workflows

+

+ Open a workflow to keep editing, or start from a template. +

+
+ + + New workflow + +
+ + {/* ── Templates section ─────────────────────────────────────── */} +
+
+

Templates

+ + Working examples you can clone and customize + +
+
+ {SAMPLE_WORKFLOWS.map((s) => { + const { nodeCount, nodeTypes } = summarize(s.graph) + return ( + + ) + })} +
+
+ + {/* ── User workflows ────────────────────────────────────────── */} +
+
+

Your workflows

+ + {userFlows.length} {userFlows.length === 1 ? 'workflow' : 'workflows'} + +
+ {userFlows.length === 0 ? ( -
-

You don't have any saved workflows yet.

+
+

You haven't saved any workflows yet.

+

+ Open the canvas to build one, or start from a template above. +

- → Create your first workflow + → Open canvas
) : ( -
- {userFlows.map((flow) => ( -
-
-
-

- {flow.name || 'Untitled Workflow'} -

-

- Created {new Date(flow.createdAt).toLocaleDateString()} -

-
- - Open - -
-
- ))} +
+ {userFlows.map((flow) => { + const { nodeCount, nodeTypes } = summarize(flow.json) + return ( + + ) + })}
)} - - {/* Back button */} -
- - ← Back to Canvas - -
-
+
) diff --git a/components/canvas/Canvas.tsx b/components/canvas/Canvas.tsx index 9110629..df0e181 100644 --- a/components/canvas/Canvas.tsx +++ b/components/canvas/Canvas.tsx @@ -39,6 +39,8 @@ import LLMNode from '@/components/nodes/LLMNode' import ToolNode from '@/components/nodes/ToolNode' import MemoryNode from '@/components/nodes/MemoryNode' import DatabaseNode from '@/components/nodes/DatabaseNode' +import EmbeddingNode from '@/components/nodes/EmbeddingNode' +import VectorNode from '@/components/nodes/VectorNode' import OutputNode from '@/components/nodes/OutputNode' import ShapeNode from '@/components/nodes/ShapeNode' import TextNode from '@/components/nodes/TextNode' @@ -54,17 +56,19 @@ import { // Module-scope — new object on every render = infinite loop const NODE_TYPES: NodeTypes = { - input: InputNode, - prompt: PromptNode, - llm: LLMNode, - tool: ToolNode, - memory: MemoryNode, - database: DatabaseNode, - output: OutputNode, - shape: ShapeNode, - text: TextNode, - drawing: DrawingNode, - arrow: ArrowNode, + input: InputNode, + prompt: PromptNode, + llm: LLMNode, + tool: ToolNode, + memory: MemoryNode, + database: DatabaseNode, + embedding: EmbeddingNode, + vector: VectorNode, + output: OutputNode, + shape: ShapeNode, + text: TextNode, + drawing: DrawingNode, + arrow: ArrowNode, } as const const DEFAULT_EDGE_OPTIONS: DefaultEdgeOptions = { @@ -81,13 +85,15 @@ const PORT_COMPAT: Record = { function nodeColor(data: NodeData): string { const colors: Record = { - input: '#6366f1', - prompt: '#8b5cf6', - llm: '#3b82f6', - tool: '#f59e0b', - memory: '#14b8a6', - database: '#22d3ee', - output: '#22c55e', + input: '#6366f1', + prompt: '#8b5cf6', + llm: '#3b82f6', + tool: '#f59e0b', + memory: '#14b8a6', + database: '#22d3ee', + embedding: '#e879f9', + vector: '#a78bfa', + output: '#22c55e', shape: '#94a3b8', text: '#94a3b8', drawing: '#94a3b8', @@ -115,9 +121,11 @@ function nodeInfoText(n: AgentNode): string { case 'tool': return `${cfg.method ?? 'GET'} · ${(cfg.url as string) || 'no url'}` case 'prompt': return 'Prompt template' case 'memory': return `Buffer · k=${cfg.k ?? 10}` - case 'database': return `DB · ${(cfg.driver as string) ?? 'postgres'} · ${(cfg.mode as string) ?? 'query'}` - case 'input': return `Input · ${cfg.inputType ?? 'text'}` - case 'output': return 'Output sink' + case 'database': return `DB · ${(cfg.driver as string) ?? 'postgres'} · ${(cfg.mode as string) ?? 'query'}` + case 'embedding': return `Embed · ${(cfg.provider as string) ?? 'openai'} · ${(cfg.model as string) ?? 'text-embedding-3-small'}` + case 'vector': return `Vector · ${(cfg.indexType as string) ?? 'flat'} · top_k=${(cfg.topK as number) ?? 5}` + case 'input': return `Input · ${cfg.inputType ?? 'text'}` + case 'output': return 'Output sink' default: return n.data.nodeType } } diff --git a/components/canvas/NodePalette.tsx b/components/canvas/NodePalette.tsx index 3239d4f..18434cf 100644 --- a/components/canvas/NodePalette.tsx +++ b/components/canvas/NodePalette.tsx @@ -98,6 +98,38 @@ const PALETTE: PaletteItem[] = [ ), }, + { + kind: 'embedding', + label: 'Embedding', + dot: 'bg-fuchsia-400', + hover: 'group-hover:border-fuchsia-400/40 group-hover:bg-fuchsia-400/5', + icon: ( + + + + + + + + + + + ), + }, + { + kind: 'vector', + label: 'Vector', + dot: 'bg-violet-400', + hover: 'group-hover:border-violet-400/40 group-hover:bg-violet-400/5', + icon: ( + + + + + + + ), + }, { kind: 'output', label: 'Output', diff --git a/components/canvas/NodeSidebar.tsx b/components/canvas/NodeSidebar.tsx index b15c8fd..3d8dc32 100644 --- a/components/canvas/NodeSidebar.tsx +++ b/components/canvas/NodeSidebar.tsx @@ -24,7 +24,7 @@ import { import { useStore, type AgentNodeKind, type RunHistoryEntry } from '@/store' import { useSessionKeys } from '@/store/sessionKeys' import type { DBColumn, DBDriver, DBMode, DBSchema, DBTable, ToolRunSnapshot } from '@/lib/types' -import type { ProviderId, ProviderModel } from '@/lib/providers/registry' +import { isProviderId, type ProviderId, type ProviderModel } from '@/lib/providers/registry' const MonacoEditor = dynamic(() => import('@monaco-editor/react'), { ssr: false }) @@ -136,13 +136,15 @@ const MONACO_OPTIONS_PROMPT = { // ─── Design tokens ──────────────────────────────────────────────────────────── const NODE_META: Record = { - input: { label: 'Input Node', color: 'text-indigo-400', dot: 'bg-indigo-400' }, - prompt: { label: 'Prompt Node', color: 'text-purple-400', dot: 'bg-purple-400' }, - llm: { label: 'LLM Node', color: 'text-blue-400', dot: 'bg-blue-400' }, - tool: { label: 'Tool Node', color: 'text-amber-400', dot: 'bg-amber-400' }, - memory: { label: 'Memory Node', color: 'text-teal-400', dot: 'bg-teal-400' }, - database: { label: 'Database Node', color: 'text-cyan-400', dot: 'bg-cyan-400' }, - output: { label: 'Output Node', color: 'text-green-400', dot: 'bg-green-400' }, + input: { label: 'Input Node', color: 'text-indigo-400', dot: 'bg-indigo-400' }, + prompt: { label: 'Prompt Node', color: 'text-purple-400', dot: 'bg-purple-400' }, + llm: { label: 'LLM Node', color: 'text-blue-400', dot: 'bg-blue-400' }, + tool: { label: 'Tool Node', color: 'text-amber-400', dot: 'bg-amber-400' }, + memory: { label: 'Memory Node', color: 'text-teal-400', dot: 'bg-teal-400' }, + database: { label: 'Database Node', color: 'text-cyan-400', dot: 'bg-cyan-400' }, + embedding: { label: 'Embedding Node', color: 'text-fuchsia-400', dot: 'bg-fuchsia-400' }, + vector: { label: 'Vector Node', color: 'text-violet-400', dot: 'bg-violet-400' }, + output: { label: 'Output Node', color: 'text-green-400', dot: 'bg-green-400' }, } // ─── Reusable field row ─────────────────────────────────────────────────────── @@ -2418,6 +2420,687 @@ function OutputHistory({ history }: { history: RunHistoryEntry[] }) { ) } +// ─── Provider key block (reusable) ─────────────────────────────────────────── +// Same UX as the LLM form's key block — session vs DB persistence, validation +// via /api/v1/keys. Self-contained so any provider-bound node can drop it in. + +type PersistMode = 'session' | 'db' + +interface ProviderKeyBlockProps { + provider: ProviderId + label?: string + accent?: 'blue' | 'fuchsia' | 'violet' +} + +const KEY_ACCENTS = { + blue: { ring: 'focus-visible:ring-blue-500/30', btn: 'bg-blue-500/20 hover:bg-blue-500/30 text-blue-300 border-blue-500/30 hover:border-blue-400/50', pill: 'bg-blue-500/15 text-blue-300 border-blue-500/25', tab: 'bg-blue-500/20 text-blue-300' }, + fuchsia: { ring: 'focus-visible:ring-fuchsia-500/30', btn: 'bg-fuchsia-500/20 hover:bg-fuchsia-500/30 text-fuchsia-300 border-fuchsia-500/30 hover:border-fuchsia-400/50', pill: 'bg-fuchsia-500/15 text-fuchsia-300 border-fuchsia-500/25', tab: 'bg-fuchsia-500/20 text-fuchsia-300' }, + violet: { ring: 'focus-visible:ring-violet-500/30', btn: 'bg-violet-500/20 hover:bg-violet-500/30 text-violet-300 border-violet-500/30 hover:border-violet-400/50', pill: 'bg-violet-500/15 text-violet-300 border-violet-500/25', tab: 'bg-violet-500/20 text-violet-300' }, +} as const + +function ProviderKeyBlock({ provider, label, accent = 'blue' }: ProviderKeyBlockProps) { + const setSessionKey = useSessionKeys((s) => s.setKey) + const clearSessionKey = useSessionKeys((s) => s.clearKey) + const sessionKeyMap = useSessionKeys((s) => s.keys) + + const [persistedProviders, setPersistedProviders] = useState([]) + const refreshPersisted = useCallback(() => { + fetch('/api/v1/keys') + .then((r) => r.json()) + .then((data: { keys: { provider: ProviderId }[] }) => { + setPersistedProviders(data.keys.map((k) => k.provider)) + }) + .catch(() => {}) + }, []) + useEffect(() => { refreshPersisted() }, [refreshPersisted]) + + const hasSessionKey = Boolean(sessionKeyMap[provider]) + const hasPersistedKey = persistedProviders.includes(provider) + + const [keyInput, setKeyInput] = useState('') + const [persistMode, setPersistMode] = useState('session') + const [keyStatus, setKeyStatus] = useState<{ kind: 'idle' | 'validating' | 'ok' | 'error'; message?: string }>({ kind: 'idle' }) + + // Reset input when switching providers + useEffect(() => { setKeyInput(''); setKeyStatus({ kind: 'idle' }) }, [provider]) + + async function handleSaveKey() { + const trimmed = keyInput.trim() + if (trimmed.length < 8) { + setKeyStatus({ kind: 'error', message: 'Key looks too short' }) + return + } + setKeyStatus({ kind: 'validating' }) + try { + const res = await fetch('/api/v1/keys', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ provider, apiKey: trimmed, persist: persistMode }), + }) + const data = await res.json().catch(() => ({})) as { error?: string; ok?: boolean } + if (!res.ok || !data.ok) { + setKeyStatus({ kind: 'error', message: data.error ?? 'Validation failed' }) + return + } + if (persistMode === 'session') { + setSessionKey(provider, trimmed) + } else { + clearSessionKey(provider) + refreshPersisted() + } + setKeyInput('') + setKeyStatus({ kind: 'ok', message: persistMode === 'db' ? 'Saved & encrypted' : 'Active for this session' }) + } catch (err) { + setKeyStatus({ kind: 'error', message: err instanceof Error ? err.message : String(err) }) + } + } + + async function handleClearKey() { + clearSessionKey(provider) + if (hasPersistedKey) { + await fetch(`/api/v1/keys/${provider}`, { method: 'DELETE' }).catch(() => {}) + refreshPersisted() + } + setKeyStatus({ kind: 'idle' }) + } + + const a = KEY_ACCENTS[accent] + + return ( +
+
+
+

+ {label ?? provider} API Key +

+

+ {hasSessionKey && 'Active for this session'} + {!hasSessionKey && hasPersistedKey && 'Saved in database (encrypted)'} + {!hasSessionKey && !hasPersistedKey && 'No key configured — runs will fail'} +

+
+ {(hasSessionKey || hasPersistedKey) && ( + + {hasPersistedKey ? 'DB' : 'Session'} + + )} +
+ +
+ setKeyInput(e.target.value)} + placeholder={hasSessionKey || hasPersistedKey ? '•••• replace key' : 'sk-…'} + className={`bg-[#1a1a1e] border-white/10 text-white/80 placeholder:text-white/20 font-mono text-xs h-9 ${a.ring}`} + /> + +
+ {(['session', 'db'] as const).map((mode) => ( + + ))} +
+ +
+ + {(hasSessionKey || hasPersistedKey) && ( + + )} +
+ + {keyStatus.message && ( +

+ {keyStatus.message} +

+ )} +
+
+ ) +} + +// ─── Embedding form ────────────────────────────────────────────────────────── +// Source → chunker → embedding model. The "Source" dropdown drives which slot +// of the upstream context the handler reads from (auto / output / dbRows / …). + +const EMBEDDING_PROVIDERS: { value: 'openai' | 'google' | 'openrouter'; label: string }[] = [ + { value: 'openai', label: 'OpenAI' }, + { value: 'google', label: 'Google' }, + { value: 'openrouter', label: 'OpenRouter' }, +] + +const EMBEDDING_MODELS: Record = { + openai: [ + { id: 'text-embedding-3-small', label: 'text-embedding-3-small', dims: 1536 }, + { id: 'text-embedding-3-large', label: 'text-embedding-3-large', dims: 3072 }, + { id: 'text-embedding-ada-002', label: 'ada-002 (legacy)', dims: 1536 }, + ], + google: [ + { id: 'text-embedding-004', label: 'text-embedding-004', dims: 768 }, + ], + openrouter: [ + { id: 'openai/text-embedding-3-small', label: 'OpenAI 3-small (via OR)', dims: 1536 }, + ], +} + +const EMBEDDING_SOURCES = [ + { value: 'auto', label: 'Auto', description: 'DB rows → upstream output → input' }, + { value: 'output', label: 'Output', description: 'Upstream node output' }, + { value: 'input', label: 'User input', description: 'Original user input' }, + { value: 'dbRows', label: 'Database rows', description: 'Forwarded DB query rows' }, +] as const + +function EmbeddingForm({ + config, + onSave, +}: { + config: Record + onSave: (v: Record) => void +}) { + const [provider, setProvider] = useState((config.provider as string) ?? 'openai') + const [model, setModel] = useState((config.model as string) ?? 'text-embedding-3-small') + const [dimensions, setDimensions] = useState((config.dimensions as number) ?? 1536) + const [chunkSize, setChunkSize] = useState((config.chunkSize as number) ?? 512) + const [chunkOverlap, setChunkOverlap] = useState((config.chunkOverlap as number) ?? 64) + const [sourceField, setSourceField] = useState((config.sourceField as string) ?? 'auto') + + const models = EMBEDDING_MODELS[provider] ?? [] + + // When provider changes, snap to the first model + its dimensions + useEffect(() => { + if (models.length === 0) return + if (!models.some((m) => m.id === model)) { + setModel(models[0].id) + setDimensions(models[0].dims) + } + }, [provider, models, model]) + + function selectModel(id: string) { + setModel(id) + const m = models.find((x) => x.id === id) + if (m) setDimensions(m.dims) + } + + function handleSave() { + onSave({ provider, model, dimensions, chunkSize, chunkOverlap, sourceField }) + } + + return ( +
+ + + + + + + + + + + + + {isProviderId(provider) && ( + p.value === provider)?.label} + accent="fuchsia" + /> + )} + + + setChunkSize(v)} + className="[&_[role=slider]]:bg-fuchsia-400 [&_[role=slider]]:border-fuchsia-400 [&_.bg-primary]:bg-fuchsia-400" + /> + + + + setChunkOverlap(v)} + className="[&_[role=slider]]:bg-fuchsia-400 [&_[role=slider]]:border-fuchsia-400 [&_.bg-primary]:bg-fuchsia-400" + /> + + +

+ API key for the selected provider is read from the LLM node config or your saved keys. + Anthropic does not currently expose embedding models. +

+ + +
+ ) +} + +// ─── Vector form ───────────────────────────────────────────────────────────── +// Index/query mode + retriever tuning. Persists per (flowId, nodeId) once +// embeddings are written by the runtime. + +const VECTOR_MODES = [ + { value: 'auto', label: 'Auto', description: 'Index when chunks arrive, query otherwise' }, + { value: 'index', label: 'Index', description: 'Always upsert into the store' }, + { value: 'query', label: 'Query', description: 'Always retrieve top-K' }, +] as const + +const INDEX_TYPES = [ + { value: 'flat', label: 'Flat (exact)', description: 'Brute-force; best recall' }, + { value: 'hnsw', label: 'HNSW', description: 'Fast ANN, higher RAM' }, + { value: 'ivfflat', label: 'IVFFlat', description: 'Cluster-based, lower recall' }, +] as const + +const METRICS = [ + { value: 'cosine', label: 'Cosine' }, + { value: 'dot', label: 'Dot' }, + { value: 'l2', label: 'L2' }, +] as const + +const INJECT_TARGETS = [ + { value: 'context', label: 'Context (prompt)', description: 'Prepend retrieved text + question' }, + { value: 'messages', label: 'System message', description: 'Inject as system; pass user through' }, + { value: 'output', label: 'Raw output', description: 'Replace output with hits only' }, +] as const + +interface VectorRowView { + id: string + chunkIndex: number + content: string + preview: number[] + dims: number + metadata: Record + createdAt: string +} + +interface VectorStoreView { + id: string + name: string + provider: string + model: string + dimensions: number + indexType: string + metric: string + createdAt: string +} + +interface VectorSchemaCol { + column: string + type: string + note: string +} + +interface VectorListResponse { + schema: VectorSchemaCol[] + store: VectorStoreView | null + rows: VectorRowView[] + total: number + limit?: number + error?: string +} + +function StoredVectorsViewer({ flowId, nodeId }: { flowId: string | null; nodeId: string | null }) { + const [data, setData] = useState(null) + const [loading, setLoading] = useState(false) + const [error, setError] = useState(null) + const [open, setOpen] = useState(null) + + const load = useCallback(async () => { + if (!flowId || !nodeId) return + setLoading(true) + setError(null) + try { + const res = await fetch(`/api/v1/vectors?flowId=${encodeURIComponent(flowId)}&nodeId=${encodeURIComponent(nodeId)}&limit=200`) + const json = await res.json() as VectorListResponse + if (!res.ok || json.error) throw new Error(json.error ?? `HTTP ${res.status}`) + setData(json) + } catch (err) { + setError(err instanceof Error ? err.message : String(err)) + } finally { + setLoading(false) + } + }, [flowId, nodeId]) + + useEffect(() => { load() }, [load]) + + const schema = data?.schema ?? [] + const rows = data?.rows ?? [] + const store = data?.store ?? null + + return ( +
+ {/* Header */} +
+ Stored Vectors + +
+ + {/* Schema */} +
+
+ embeddings + schema +
+
+ {schema.map((c) => ( +
+ {c.column} + {c.type} + {c.note} +
+ ))} +
+
+ + {/* Store metadata */} + {store && ( +
+
+ {store.name} + + {store.indexType.toUpperCase()} · {store.metric} + +
+
+ {store.provider} · {store.model} · {store.dimensions}d +
+
+ )} + + {/* Status */} + {error && ( +
+ {error} +
+ )} + {!flowId && ( +

Save the flow before viewing stored vectors.

+ )} + + {/* Rows */} + {rows.length === 0 && !loading && !error && flowId && ( +
+ No vectors stored yet. + Run the flow with Mode = index. +
+ )} + + {rows.length > 0 && ( +
+
+ {rows.length} row{rows.length === 1 ? '' : 's'} + first {rows[0]?.preview.length ?? 0} dims shown +
+ {rows.map((r) => { + const isOpen = open === r.id + return ( +
+ + {isOpen && ( +
+
+ content + 
+                        {r.content}
+
+
+
+ embedding (preview) + 
+                        [{r.preview.map((v) => v.toFixed(6)).join(', ')}{r.preview.length < r.dims ? `, … ${r.dims - r.preview.length} more` : ''}]
+
+
+ {Object.keys(r.metadata ?? {}).length > 0 && ( +
+ metadata + 
+                          {JSON.stringify(r.metadata, null, 2)}
+
+
+ )} +
+ {new Date(r.createdAt).toLocaleString()} +
+
+ )} +
+ ) + })} +
+ )} +
+ ) +} + +function VectorForm({ + config, + onSave, + flowId, + nodeId, +}: { + config: Record + onSave: (v: Record) => void + flowId: string | null + nodeId: string | null +}) { + const [storeName, setStoreName] = useState((config.storeName as string) ?? 'default') + const [mode, setMode] = useState((config.mode as string) ?? 'auto') + const [indexType, setIndexType] = useState((config.indexType as string) ?? 'flat') + const [metric, setMetric] = useState((config.metric as string) ?? 'cosine') + const [topK, setTopK] = useState((config.topK as number) ?? 5) + const [topP, setTopP] = useState((config.topP as number) ?? 0) + const [injectInto, setInjectInto] = useState((config.injectInto as string) ?? 'context') + const [replace, setReplace] = useState(Boolean(config.replace)) + + function handleSave() { + onSave({ storeName, mode, indexType, metric, topK, topP, injectInto, replace }) + } + + return ( +
+ + setStoreName(e.target.value || 'default')} + placeholder="default" + className="bg-[#1a1a1e] border-white/10 text-white/80 placeholder:text-white/20 font-mono text-xs h-9 focus-visible:ring-violet-500/30" + /> + + + + + + + + + + + + + + + + setTopK(v)} + className="[&_[role=slider]]:bg-violet-400 [&_[role=slider]]:border-violet-400 [&_.bg-primary]:bg-violet-400" + /> + + + + setTopP(v)} + className="[&_[role=slider]]:bg-violet-400 [&_[role=slider]]:border-violet-400 [&_.bg-primary]:bg-violet-400" + /> + + + + + + + + + + + +
+ ) +} + // ─── NodeSidebar ────────────────────────────────────────────────────────────── export default function NodeSidebar() { @@ -2428,6 +3111,7 @@ export default function NodeSidebar() { }))) const nodes = useStore(useShallow((s) => s.nodes)) + const flowId = useStore((s) => s.flowId) const updateNodeData = useStore((s) => s.updateNodeData) const [saved, setSaved] = useState(false) @@ -2505,6 +3189,17 @@ export default function NodeSidebar() { {node?.data.nodeType === 'database' && ( )} + {node?.data.nodeType === 'embedding' && ( + + )} + {node?.data.nodeType === 'vector' && ( + + )} {node?.data.nodeType === 'output' && ( )} diff --git a/components/nodes/EmbeddingNode.tsx b/components/nodes/EmbeddingNode.tsx new file mode 100644 index 0000000..79051b3 --- /dev/null +++ b/components/nodes/EmbeddingNode.tsx @@ -0,0 +1,82 @@ +'use client' + +import { memo } from 'react' +import { Handle, Position, type NodeProps } from '@xyflow/react' +import type { NodeData, RunStatus } from '@/store' + +const statusBorder: Record = { + idle: 'border-white/10', + running: 'border-[#00ff88] shadow-[0_0_18px_rgba(0,255,136,0.45)] ring-1 ring-[#00ff88]/30', + done: 'border-[#00ff88]/50', + error: 'border-red-500', +} + +function EmbeddingNode({ data, selected }: NodeProps) { + const d = data as NodeData + const status = d.runStatus ?? 'idle' + const border = statusBorder[status] + const selRing = selected ? 'ring-1 ring-fuchsia-400/60' : '' + const provider = (d.config?.provider as string | undefined) ?? 'openai' + const model = (d.config?.model as string | undefined) ?? 'text-embedding-3-small' + const dims = (d.config?.dimensions as number | undefined) ?? 1536 + const meta = d.runMeta + + return ( +
+ + + + {/* Header */} +
+ + Embedding + + {provider} + +
+ + {/* Model + dims */} +
+

{model}

+ {dims}d +
+ + {/* Running */} + {status === 'running' && ( +
+ + Embedding… +
+ )} + + {/* Done */} + {status === 'done' && ( +
+ + {(meta?.vectorCount ?? 0)} vec{meta?.vectorCount === 1 ? '' : 's'} + + + {meta?.durationMs != null ? `${meta.durationMs}ms` : ''} + +
+ )} + + {/* Error */} + {status === 'error' && ( +
+ + {meta?.errorMsg ?? 'Embedding failed'} + +
+ )} +
+ ) +} + +export default memo(EmbeddingNode, (prev, next) => + (prev.data as NodeData).runStatus === (next.data as NodeData).runStatus && + (prev.data as NodeData).runOutput === (next.data as NodeData).runOutput && + (prev.data as NodeData).runMeta === (next.data as NodeData).runMeta && + (prev.data as NodeData).config === (next.data as NodeData).config && + prev.selected === next.selected +) diff --git a/components/nodes/VectorNode.tsx b/components/nodes/VectorNode.tsx new file mode 100644 index 0000000..b9085eb --- /dev/null +++ b/components/nodes/VectorNode.tsx @@ -0,0 +1,110 @@ +'use client' + +import { memo } from 'react' +import { Handle, Position, type NodeProps } from '@xyflow/react' +import type { NodeData, RunStatus } from '@/store' + +const statusBorder: Record = { + idle: 'border-white/10', + running: 'border-[#00ff88] shadow-[0_0_18px_rgba(0,255,136,0.45)] ring-1 ring-[#00ff88]/30', + done: 'border-[#00ff88]/50', + error: 'border-red-500', +} + +const indexLabel: Record = { + flat: 'FLAT', + hnsw: 'HNSW', + ivfflat: 'IVF', +} + +const modeColor: Record = { + index: 'bg-amber-500/15 text-amber-300 border-amber-500/20', + query: 'bg-emerald-500/15 text-emerald-300 border-emerald-500/20', + auto: 'bg-violet-500/15 text-violet-300 border-violet-500/20', +} + +function VectorNode({ data, selected }: NodeProps) { + const d = data as NodeData + const status = d.runStatus ?? 'idle' + const border = statusBorder[status] + const selRing = selected ? 'ring-1 ring-violet-400/60' : '' + const storeName = (d.config?.storeName as string | undefined) ?? 'default' + const indexType = (d.config?.indexType as string | undefined) ?? 'flat' + const metric = (d.config?.metric as string | undefined) ?? 'cosine' + const topK = (d.config?.topK as number | undefined) ?? 5 + const mode = (d.config?.mode as string | undefined) ?? 'auto' + const meta = d.runMeta + const runMode = meta?.mode ?? mode + + return ( +
+ + + + + {/* Header */} +
+ + Vector + + {runMode} + +
+ + {/* Store + index */} +
+

{storeName}

+ + {indexLabel[indexType] ?? indexType.toUpperCase()} · {metric} + +
+ + {/* Hyperparams */} +
+

+ top_k={topK} + {(d.config?.topP as number | undefined) ? ` · top_p=${d.config?.topP}` : ''} +

+
+ + {/* Running */} + {status === 'running' && ( +
+ + + {runMode === 'index' ? 'Indexing…' : 'Querying…'} + +
+ )} + + {/* Done */} + {status === 'done' && ( +
+ + {(meta?.vectorCount ?? 0)} {runMode === 'index' ? 'stored' : 'matched'} + + + {meta?.durationMs != null ? `${meta.durationMs}ms` : ''} + +
+ )} + + {/* Error */} + {status === 'error' && ( +
+ + {meta?.errorMsg ?? 'Vector op failed'} + +
+ )} +
+ ) +} + +export default memo(VectorNode, (prev, next) => + (prev.data as NodeData).runStatus === (next.data as NodeData).runStatus && + (prev.data as NodeData).runOutput === (next.data as NodeData).runOutput && + (prev.data as NodeData).runMeta === (next.data as NodeData).runMeta && + (prev.data as NodeData).config === (next.data as NodeData).config && + prev.selected === next.selected +) diff --git a/components/workflows/WorkflowCard.tsx b/components/workflows/WorkflowCard.tsx new file mode 100644 index 0000000..0a8973d --- /dev/null +++ b/components/workflows/WorkflowCard.tsx @@ -0,0 +1,191 @@ +'use client' + +import { useRouter } from 'next/navigation' +import { useState } from 'react' +import { toast } from 'sonner' +import { Button } from '@/components/ui/button' + +interface NodeShape { type?: string; data?: { nodeType?: string } } + +export interface WorkflowCardProps { + id: string + name: string + description?: string + tags?: string[] + createdAt?: string | Date + nodeCount?: number + /** Node-type counts so the card can render small icon-bands */ + nodeTypes?: string[] + variant: 'user' | 'sample' +} + +const TYPE_BADGE: Record = { + input: { label: 'Input', className: 'bg-emerald-500/15 text-emerald-300 border-emerald-500/30' }, + prompt: { label: 'Prompt', className: 'bg-amber-500/15 text-amber-300 border-amber-500/30' }, + llm: { label: 'LLM', className: 'bg-indigo-500/15 text-indigo-300 border-indigo-500/30' }, + tool: { label: 'Tool', className: 'bg-cyan-500/15 text-cyan-300 border-cyan-500/30' }, + memory: { label: 'Memory', className: 'bg-pink-500/15 text-pink-300 border-pink-500/30' }, + database: { label: 'Database', className: 'bg-slate-500/15 text-slate-300 border-slate-500/30' }, + embedding: { label: 'Embedding', className: 'bg-violet-500/15 text-violet-300 border-violet-500/30' }, + vector: { label: 'Vector', className: 'bg-fuchsia-500/15 text-fuchsia-300 border-fuchsia-500/30' }, + output: { label: 'Output', className: 'bg-rose-500/15 text-rose-300 border-rose-500/30' }, +} + +const RUNNABLE_TYPES = new Set(Object.keys(TYPE_BADGE)) + +function formatDate(d?: string | Date): string { + if (!d) return '' + const date = typeof d === 'string' ? new Date(d) : d + return date.toLocaleDateString(undefined, { month: 'short', day: 'numeric', year: 'numeric' }) +} + +export default function WorkflowCard(props: WorkflowCardProps) { + const router = useRouter() + const [busy, setBusy] = useState<'idle' | 'opening' | 'cloning' | 'deleting'>('idle') + + const usedTypes = (props.nodeTypes ?? []) + .filter((t) => RUNNABLE_TYPES.has(t)) + // de-dupe while preserving order + .filter((t, i, a) => a.indexOf(t) === i) + + async function handleOpen() { + setBusy('opening') + router.push(`/canvas?flowId=${props.id}`) + } + + async function handleClone() { + setBusy('cloning') + try { + const r = await fetch(`/api/v1/workflows/clone-sample`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ sampleId: props.id }), + }) + if (!r.ok) throw new Error(`HTTP ${r.status}`) + const flow = await r.json() as { id: string; name: string } + toast.success('Template added', { description: flow.name }) + router.push(`/canvas?flowId=${flow.id}`) + } catch (err) { + console.error(err) + toast.error('Could not clone template') + setBusy('idle') + } + } + + async function handleDelete() { + if (!confirm(`Delete "${props.name}"? This cannot be undone.`)) return + setBusy('deleting') + try { + const r = await fetch(`/api/v1/flows/${props.id}`, { method: 'DELETE' }) + if (!r.ok) throw new Error(`HTTP ${r.status}`) + toast.success('Workflow deleted') + router.refresh() + } catch (err) { + console.error(err) + toast.error('Could not delete workflow') + setBusy('idle') + } + } + + const isSample = props.variant === 'sample' + + return ( +
+ {isSample && ( + + Template + + )} + +
+
+

{props.name}

+ {props.description && ( +

+ {props.description} +

+ )} +
+
+ + {usedTypes.length > 0 && ( +
+ {usedTypes.map((t) => { + const meta = TYPE_BADGE[t] + return ( + + {meta.label} + + ) + })} +
+ )} + + {props.tags && props.tags.length > 0 && ( +
+ {props.tags.map((t) => ( + + {t} + + ))} +
+ )} + +
+ + {typeof props.nodeCount === 'number' + ? `${props.nodeCount} ${props.nodeCount === 1 ? 'node' : 'nodes'}` + : ' '} + + {formatDate(props.createdAt)} +
+ +
+ {isSample ? ( + + ) : ( + <> + + + + )} +
+
+ ) +} + +export type { NodeShape } diff --git a/lib/db/schema.ts b/lib/db/schema.ts index 8fe70e4..bb7f34c 100644 --- a/lib/db/schema.ts +++ b/lib/db/schema.ts @@ -158,10 +158,55 @@ export const apiKeys = pgTable( (t) => [uniqueIndex('api_keys_user_provider_idx').on(t.userId, t.provider)], ) +// ─── Vector stores ──────────────────────────────────────────────────────────── +// One row per VectorNode on a flow. Holds the model + index hyperparams the +// node was configured with so the runtime can recreate query behaviour later. +// Embeddings themselves live in `embeddings`, FK'd by `storeId`. + +export const vectorStores = pgTable( + 'vector_stores', + { + id: uuid('id').primaryKey().defaultRandom(), + flowId: uuid('flow_id').references(() => flows.id, { onDelete: 'cascade' }).notNull(), + nodeId: text('node_id').notNull(), // canvas node id + name: text('name').notNull().default('default'), + provider: text('provider').notNull(), // openai | google | … + model: text('model').notNull(), // e.g. text-embedding-3-small + dimensions: integer('dimensions').notNull(), + indexType: text('index_type').notNull().default('flat'), // flat | hnsw | ivfflat + metric: text('metric').notNull().default('cosine'), // cosine | l2 | dot + config: jsonb('config').notNull().default({}), // hyperparams (topK, topP, …) + createdAt: timestamp('created_at').notNull().defaultNow(), + updatedAt: timestamp('updated_at').notNull().defaultNow().$onUpdate(() => new Date()), + }, + (t) => [uniqueIndex('vector_stores_flow_node_idx').on(t.flowId, t.nodeId)], +) + +// ─── Embeddings ─────────────────────────────────────────────────────────────── +// Vector + chunk content. `embedding` is a portable jsonb number[] so the table +// works without the pgvector extension; the service layer scores in-process. +// When pgvector is enabled the column can be migrated to vector(d) without a +// data shape change. + +export const embeddings = pgTable('embeddings', { + id: uuid('id').primaryKey().defaultRandom(), + storeId: uuid('store_id').references(() => vectorStores.id, { onDelete: 'cascade' }).notNull(), + flowId: uuid('flow_id').references(() => flows.id, { onDelete: 'cascade' }).notNull(), + chunkIndex: integer('chunk_index').notNull().default(0), + content: text('content').notNull(), + embedding: jsonb('embedding').notNull(), // number[] + metadata: jsonb('metadata').notNull().default({}), // { source, page, … } + createdAt: timestamp('created_at').notNull().defaultNow(), +}) + // ─── Inferred row types ─────────────────────────────────────────────────────── -export type User = typeof users.$inferSelect -export type NewUser = typeof users.$inferInsert -export type Profile = typeof profiles.$inferSelect -export type NewProfile = typeof profiles.$inferInsert -export type ApiKey = typeof apiKeys.$inferSelect +export type User = typeof users.$inferSelect +export type NewUser = typeof users.$inferInsert +export type Profile = typeof profiles.$inferSelect +export type NewProfile = typeof profiles.$inferInsert +export type ApiKey = typeof apiKeys.$inferSelect +export type VectorStore = typeof vectorStores.$inferSelect +export type NewVectorStore = typeof vectorStores.$inferInsert +export type Embedding = typeof embeddings.$inferSelect +export type NewEmbedding = typeof embeddings.$inferInsert diff --git a/lib/nodeFactory.ts b/lib/nodeFactory.ts index 5be5c01..62000ec 100644 --- a/lib/nodeFactory.ts +++ b/lib/nodeFactory.ts @@ -15,22 +15,42 @@ const AGENT_DEFAULTS: Record> = { forwardSchema: true, forwardRows: true, }, + embedding: { + provider: 'openai', + model: 'text-embedding-3-small', + dimensions: 1536, + chunkSize: 512, + chunkOverlap: 64, + sourceField: 'auto', // 'auto' | 'output' | 'dbRows' | custom field name + }, + vector: { + storeName: 'default', + indexType: 'flat', // flat | hnsw | ivfflat + metric: 'cosine', // cosine | l2 | dot + mode: 'auto', // 'auto' (index when chunks provided, else query) | 'index' | 'query' + topK: 5, + topP: 0.0, // similarity threshold (0 = disabled) + injectInto: 'context', // 'context' | 'messages' | 'output' + }, output: {}, } const PORT_MAP: Record = { - input: { inputs: [], outputs: [{ id: 'text-out', type: 'string' }] }, - prompt: { inputs: [{ id: 'vars-in', type: 'any' }], outputs: [{ id: 'text-out', type: 'string' }] }, - llm: { inputs: [{ id: 'messages-in', type: 'messages' }, { id: 'system-in', type: 'string' }], outputs: [{ id: 'messages-out', type: 'messages' }, { id: 'text-out', type: 'string' }] }, - tool: { inputs: [{ id: 'trigger-in', type: 'any' }], outputs: [{ id: 'json-out', type: 'json' }] }, - memory: { inputs: [{ id: 'messages-in', type: 'messages' }], outputs: [{ id: 'messages-out',type: 'messages' }] }, - database: { inputs: [{ id: 'trigger-in', type: 'any' }], outputs: [{ id: 'json-out', type: 'json' }] }, - output: { inputs: [{ id: 'text-in', type: 'string' }], outputs: [] }, + input: { inputs: [], outputs: [{ id: 'text-out', type: 'string' }] }, + prompt: { inputs: [{ id: 'vars-in', type: 'any' }], outputs: [{ id: 'text-out', type: 'string' }] }, + llm: { inputs: [{ id: 'messages-in', type: 'messages' }, { id: 'system-in', type: 'string' }], outputs: [{ id: 'messages-out', type: 'messages' }, { id: 'text-out', type: 'string' }] }, + tool: { inputs: [{ id: 'trigger-in', type: 'any' }], outputs: [{ id: 'json-out', type: 'json' }] }, + memory: { inputs: [{ id: 'messages-in', type: 'messages' }], outputs: [{ id: 'messages-out',type: 'messages' }] }, + database: { inputs: [{ id: 'trigger-in', type: 'any' }], outputs: [{ id: 'json-out', type: 'json' }] }, + embedding: { inputs: [{ id: 'source-in', type: 'any' }], outputs: [{ id: 'vectors-out', type: 'json' }] }, + vector: { inputs: [{ id: 'vectors-in', type: 'any' }], outputs: [{ id: 'context-out', type: 'string' }, { id: 'json-out', type: 'json' }] }, + output: { inputs: [{ id: 'text-in', type: 'string' }], outputs: [] }, } const AGENT_LABELS: Record = { input: '+ Input', prompt: '+ Prompt', llm: '+ LLM', - tool: '+ Tool', memory: '+ Memory', database: '+ Database', output: '+ Output', + tool: '+ Tool', memory: '+ Memory', database: '+ Database', + embedding: '+ Embedding', vector: '+ Vector', output: '+ Output', } export const AGENT_NODE_LABELS = AGENT_LABELS diff --git a/lib/runtime/execute.ts b/lib/runtime/execute.ts index 3fa7042..f5c0165 100644 --- a/lib/runtime/execute.ts +++ b/lib/runtime/execute.ts @@ -6,8 +6,10 @@ import { handleLLM } from './handlers/llm' import { handleOutput } from './handlers/output' import { handleTool } from './handlers/tool' import { handleDatabase } from './handlers/database' +import { handleEmbedding } from './handlers/embedding' +import { handleVector } from './handlers/vector' -const ANNOTATION_TYPES = new Set(['shape', 'text', 'drawing']) +const ANNOTATION_TYPES = new Set(['shape', 'text', 'drawing', 'arrow']) export async function execute( graph: FlowGraph, @@ -40,7 +42,8 @@ export async function execute( const inContext: NodeContext = parentContexts.reduce( (acc, ctx) => ({ ...acc, ...ctx }), { - input: userInput, + input: userInput, + flowId: graph.id, ...(fileData ? { fileData } : {}), ...(sessionKeys ? { sessionKeys } : {}), }, @@ -70,6 +73,14 @@ export async function execute( outContext = await handleDatabase(node.id, node.data.config as unknown as DBNodeConfig, inContext, emit) break + case 'embedding': + outContext = await handleEmbedding(node.id, node.data.config as Record, inContext, emit) + break + + case 'vector': + outContext = await handleVector(node.id, node.data.config as Record, inContext, emit) + break + case 'output': outContext = await handleOutput(node.id, undefined, inContext, emit) break diff --git a/lib/runtime/handlers/embedding.ts b/lib/runtime/handlers/embedding.ts new file mode 100644 index 0000000..f8ebd9c --- /dev/null +++ b/lib/runtime/handlers/embedding.ts @@ -0,0 +1,144 @@ +import type { EmitFn, EmbeddingPayload, NodeContext, SessionKeys } from '@/lib/types' +import { isProviderId, type ProviderId } from '@/lib/providers/registry' +import { currentUserId } from '@/lib/auth' +import { chunkText, embedQuery, embedTexts } from '@/services/vector-store' + +interface EmbeddingConfig { + provider?: string + model?: string + dimensions?: number + chunkSize?: number + chunkOverlap?: number + sourceField?: string // 'auto' | 'output' | 'input' | 'dbRows' | named context key +} + +/** + * Pulls the text payload to embed off the upstream context. + * - 'auto' (default): prefer dbRows → output → input. dbRows are stringified + * row-by-row so each row becomes its own chunk. + * - named field: read context[name] and stringify if needed. + */ +function gatherSourceTexts(ctx: NodeContext, sourceField: string): { texts: string[]; queryText: string | null } { + const out = ctx.output ?? '' + const inp = ctx.input ?? '' + + if (sourceField !== 'auto') { + const v = ctx[sourceField] + if (v === undefined || v === null) return { texts: [], queryText: null } + if (Array.isArray(v)) return { texts: v.map((r) => (typeof r === 'string' ? r : JSON.stringify(r))), queryText: null } + if (typeof v === 'string') return { texts: [v], queryText: v } + return { texts: [JSON.stringify(v)], queryText: null } + } + + // auto: rows from a Database node split into one chunk per row, otherwise + // treat the streamed output as a document to be re-chunked. + if (Array.isArray(ctx.dbRows) && ctx.dbRows.length > 0) { + return { texts: ctx.dbRows.map((r) => JSON.stringify(r)), queryText: out || inp || null } + } + if (out) return { texts: [out], queryText: out } + if (inp) return { texts: [inp], queryText: inp } + return { texts: [], queryText: null } +} + +export async function handleEmbedding( + nodeId: string, + config: EmbeddingConfig | undefined, + context: NodeContext, + emit: EmitFn, +): Promise { + const cfg: Required = { + provider: config?.provider ?? 'openai', + model: config?.model ?? 'text-embedding-3-small', + dimensions: config?.dimensions ?? 1536, + chunkSize: config?.chunkSize ?? 512, + chunkOverlap: config?.chunkOverlap ?? 64, + sourceField: config?.sourceField ?? 'auto', + } + + if (!isProviderId(cfg.provider)) { + throw Object.assign(new Error(`Unknown embedding provider: ${cfg.provider}`), { + errorMeta: { code: 'unknown', message: `Unknown embedding provider: ${cfg.provider}` }, + }) + } + const provider: ProviderId = cfg.provider + + const t0 = Date.now() + const { texts, queryText } = gatherSourceTexts(context, cfg.sourceField) + + // Re-chunk single long documents so we get useful retrieval granularity. + // Pre-chunked sources (DB rows) flow through as-is. + const inputCameAsRows = Array.isArray(context.dbRows) && context.dbRows.length > 0 + const chunks = inputCameAsRows + ? texts + : texts.flatMap((t) => chunkText(t, cfg.chunkSize, cfg.chunkOverlap)) + + if (chunks.length === 0) { + emit({ type: 'node-replace', nodeId, output: '{ "vectors": 0, "note": "no source text" }' }) + return { ...context, embeddings: { provider, model: cfg.model, dimensions: cfg.dimensions, chunks: [] } } + } + + const userId = await currentUserId() + const sessionKeys = context.sessionKeys as SessionKeys | undefined + + let vectors: number[][] + try { + vectors = await embedTexts({ + provider, + model: cfg.model, + texts: chunks, + userId, + sessionKeys, + }) + } catch (err) { + const msg = err instanceof Error ? err.message : String(err) + emit({ type: 'node-replace', nodeId, output: JSON.stringify({ error: msg }, null, 2) }) + throw Object.assign(new Error(msg), { + errorMeta: { code: 'unknown', message: msg, provider, model: cfg.model }, + }) + } + + const dims = vectors[0]?.length ?? cfg.dimensions + const payload: EmbeddingPayload = { + provider, + model: cfg.model, + dimensions: dims, + chunks: chunks.map((content, i) => ({ + content, + embedding: vectors[i] ?? [], + metadata: { chunkIndex: i, source: cfg.sourceField }, + })), + } + + // When the upstream looks like a question (single short text, no rows), also + // embed it as a query so a downstream Vector node can switch into query mode. + if (queryText && !inputCameAsRows && chunks.length <= 2) { + try { + const qVec = chunks.length === 1 && queryText === chunks[0] + ? vectors[0] + : await embedQuery({ provider, model: cfg.model, text: queryText, userId, sessionKeys }) + payload.query = { text: queryText, embedding: qVec } + } catch { + // query embed failure shouldn't fail the whole index step + } + } + + const summary = { + provider, + model: cfg.model, + dimensions: dims, + vectors: payload.chunks.length, + sample: payload.chunks.slice(0, 1).map((c) => ({ + content: c.content.length > 200 ? c.content.slice(0, 200) + '…' : c.content, + preview: c.embedding.slice(0, 4), + })), + } + emit({ type: 'node-replace', nodeId, output: JSON.stringify(summary, null, 2) }) + + const durationMs = Date.now() - t0 + return { + ...context, + output: JSON.stringify(summary), + embeddings: payload, + runMeta: { ...(context.runMeta ?? {}), vectorCount: payload.chunks.length, dimensions: dims, durationMs }, + } as NodeContext +} diff --git a/lib/runtime/handlers/vector.ts b/lib/runtime/handlers/vector.ts new file mode 100644 index 0000000..235c9ac --- /dev/null +++ b/lib/runtime/handlers/vector.ts @@ -0,0 +1,224 @@ +import type { EmitFn, NodeContext, RagContext, SessionKeys } from '@/lib/types' +import { isProviderId, type ProviderId } from '@/lib/providers/registry' +import { currentUserId } from '@/lib/auth' +import { + embedQuery, + queryStore, + upsertEmbeddings, + type IndexType, + type Metric, +} from '@/services/vector-store' + +interface VectorConfig { + storeName?: string + indexType?: IndexType + metric?: Metric + mode?: 'auto' | 'index' | 'query' + topK?: number + topP?: number // similarity threshold (0..1 for cosine, 0 = disabled) + injectInto?: 'context' | 'messages' | 'output' + replace?: boolean // when indexing, drop existing rows first +} + +const CONTEXT_PREFIX = 'Use the following retrieved context to answer. Cite specifics; do not invent details.\n\n' + +function formatHits(hits: RagContext['hits']): string { + return hits + .map((h, i) => { + const score = (h.score * 100).toFixed(1) + return `[${i + 1}] (score=${score}) ${h.content}` + }) + .join('\n\n') +} + +export async function handleVector( + nodeId: string, + config: VectorConfig | undefined, + context: NodeContext, + emit: EmitFn, +): Promise { + const cfg: Required = { + storeName: config?.storeName ?? 'default', + indexType: config?.indexType ?? 'flat', + metric: config?.metric ?? 'cosine', + mode: config?.mode ?? 'auto', + topK: config?.topK ?? 5, + topP: config?.topP ?? 0, + injectInto: config?.injectInto ?? 'context', + replace: config?.replace ?? false, + } + + const flowId = context.flowId + if (!flowId) { + throw Object.assign(new Error('Vector node requires a flow id (run from a saved flow).'), { + errorMeta: { code: 'unknown', message: 'Vector node requires a flow id' }, + }) + } + + const t0 = Date.now() + const payload = context.embeddings + const userId = await currentUserId() + const sessionKeys = context.sessionKeys as SessionKeys | undefined + + // ── Resolve mode ────────────────────────────────────────────────────────── + // index: chunks present and either explicit, or auto without an embedded query + // query: embedded query present, OR plain text input + an existing store + const wantIndex = + cfg.mode === 'index' || + (cfg.mode === 'auto' && payload?.chunks.length && !payload.query) + const wantQuery = + cfg.mode === 'query' || + (cfg.mode === 'auto' && (payload?.query || (!payload && (context.output || context.input)))) + + // ── Index path ──────────────────────────────────────────────────────────── + if (wantIndex && payload && payload.chunks.length > 0) { + if (!isProviderId(payload.provider)) { + throw Object.assign(new Error(`Unknown provider on embedding payload: ${payload.provider}`), { + errorMeta: { code: 'unknown', message: 'Unknown provider on embedding payload' }, + }) + } + try { + const { inserted } = await upsertEmbeddings({ + flowId, + nodeId, + storeName: cfg.storeName, + provider: payload.provider as ProviderId, + model: payload.model, + dimensions: payload.dimensions, + indexType: cfg.indexType, + metric: cfg.metric, + records: payload.chunks, + replace: cfg.replace, + config: { topK: cfg.topK, topP: cfg.topP }, + }) + + const summary = { + mode: 'index', + store: cfg.storeName, + index: cfg.indexType, + metric: cfg.metric, + stored: inserted, + dims: payload.dimensions, + } + emit({ type: 'node-replace', nodeId, output: JSON.stringify(summary, null, 2) }) + const durationMs = Date.now() - t0 + return { + ...context, + output: JSON.stringify(summary), + runMeta: { ...(context.runMeta ?? {}), mode: 'index', vectorCount: inserted, dimensions: payload.dimensions, durationMs }, + } as NodeContext + } catch (err) { + const msg = err instanceof Error ? err.message : String(err) + emit({ type: 'node-replace', nodeId, output: JSON.stringify({ error: msg }, null, 2) }) + throw Object.assign(new Error(msg), { errorMeta: { code: 'unknown', message: msg } }) + } + } + + // ── Query path ──────────────────────────────────────────────────────────── + if (!wantQuery) { + const summary = { mode: 'noop', note: 'no embeddings to index and no query text' } + emit({ type: 'node-replace', nodeId, output: JSON.stringify(summary, null, 2) }) + return { ...context, output: JSON.stringify(summary) } + } + + // Build the query embedding. Either it rode in on the embedding payload, + // or we embed the upstream text using the same model the store was built with. + let queryEmbedding = payload?.query?.embedding + if (!queryEmbedding) { + if (!payload) { + // No embedding payload at all — we need to know what model the store + // uses. The vector store row carries that, but the embed call needs a + // provider+model up front. For now, default to OpenAI text-embedding-3-small + // and surface an error if the user hasn't configured a key. + const text = context.output ?? context.input ?? '' + if (!text) { + const summary = { mode: 'query', note: 'no query text' } + emit({ type: 'node-replace', nodeId, output: JSON.stringify(summary, null, 2) }) + return { ...context, output: JSON.stringify(summary) } + } + try { + queryEmbedding = await embedQuery({ + provider: 'openai', + model: 'text-embedding-3-small', + text, + userId, + sessionKeys, + }) + } catch (err) { + const msg = err instanceof Error ? err.message : String(err) + emit({ type: 'node-replace', nodeId, output: JSON.stringify({ error: msg }, null, 2) }) + throw Object.assign(new Error(msg), { errorMeta: { code: 'unknown', message: msg } }) + } + } else { + const text = context.output ?? context.input ?? '' + try { + queryEmbedding = await embedQuery({ + provider: payload.provider as ProviderId, + model: payload.model, + text, + userId, + sessionKeys, + }) + } catch (err) { + const msg = err instanceof Error ? err.message : String(err) + emit({ type: 'node-replace', nodeId, output: JSON.stringify({ error: msg }, null, 2) }) + throw Object.assign(new Error(msg), { errorMeta: { code: 'unknown', message: msg } }) + } + } + } + + const { hits } = await queryStore({ + flowId, + nodeId, + storeName: cfg.storeName, + queryEmbedding, + topK: cfg.topK, + threshold: cfg.topP, + }) + + const rag: RagContext = { storeName: cfg.storeName, hits } + const formatted = formatHits(hits) + + // Inject retrieved context into the downstream payload according to config: + // - 'context': prepend a primer block to `output` (LLM uses as user prompt) + // - 'messages': add a system message ahead of any existing messages + // - 'output': replace output with the raw retrieved chunks + const next: NodeContext = { ...context, rag } + const incomingMessages = context.messages ?? [] + + if (cfg.injectInto === 'output') { + next.output = formatted + } else if (cfg.injectInto === 'messages') { + const existingUser = context.output ?? context.input ?? '' + next.output = existingUser + next.messages = [ + { role: 'system', content: CONTEXT_PREFIX + formatted }, + ...incomingMessages, + ] + } else { + const existingUser = context.output ?? context.input ?? '' + next.output = `${CONTEXT_PREFIX}${formatted}\n\n---\n\nQuestion: ${existingUser}` + next.messages = [{ role: 'user', content: next.output }] + } + + const summary = { + mode: 'query', + store: cfg.storeName, + metric: cfg.metric, + topK: cfg.topK, + topP: cfg.topP, + matched: hits.length, + hits: hits.map((h, i) => ({ + rank: i + 1, + score: Number(h.score.toFixed(4)), + preview: h.content.length > 160 ? h.content.slice(0, 160) + '…' : h.content, + })), + } + emit({ type: 'node-replace', nodeId, output: JSON.stringify(summary, null, 2) }) + + const durationMs = Date.now() - t0 + return { + ...next, + runMeta: { ...(next.runMeta ?? {}), mode: 'query', vectorCount: hits.length, topK: cfg.topK, durationMs }, + } as NodeContext +} diff --git a/lib/sampleWorkflows.ts b/lib/sampleWorkflows.ts new file mode 100644 index 0000000..da4fbbf --- /dev/null +++ b/lib/sampleWorkflows.ts @@ -0,0 +1,288 @@ +import type { AgentNode, AgentEdge } from '@/store' + +export interface SampleWorkflow { + id: string + name: string + tagline: string + description: string + tags: string[] + graph: { nodes: AgentNode[]; edges: AgentEdge[] } +} + +// Stable IDs so the cloned graph round-trips and is easier to read in storage. +const ids = { + noteHeader: 'note-header', + noteIndex: 'note-index', + noteQuery: 'note-query', + fileInput: 'input-csv', + fileEmbed: 'embed-index', + vecIndex: 'vector-index', + textInput: 'input-question', + queryEmbed: 'embed-query', + vecQuery: 'vector-query', + llm: 'llm-answer', + output: 'output-final', +} as const + +const STORE_NAME = 'csv-knowledge-base' + +const csvRagFlow = { + nodes: [ + // ── Header annotation ───────────────────────────────────────────────── + { + id: ids.noteHeader, + type: 'text', + position: { x: 80, y: 40 }, + width: 560, + height: 72, + selectable: true, + draggable: true, + data: { + label: 'Text', + nodeType: 'text', + config: { + text: + 'RAG over CSV — vectorize a CSV with an embedding model, then ' + + 'query the same store with the same model to ground the LLM.', + color: '#facc15', + fontSize: 14, + }, + text: + 'RAG over CSV — vectorize a CSV with an embedding model, then ' + + 'query the same store with the same model to ground the LLM.', + color: '#facc15', + fontSize: 14, + inputs: [], + outputs: [], + }, + }, + + // ── Indexing path ──────────────────────────────────────────────────── + { + id: ids.noteIndex, + type: 'text', + position: { x: 80, y: 150 }, + width: 280, + height: 48, + selectable: true, + draggable: true, + data: { + label: 'Text', + nodeType: 'text', + config: { + text: '① Index — upload a CSV file as input', + color: '#a3e635', + fontSize: 13, + }, + text: '① Index — upload a CSV file as input', + color: '#a3e635', + fontSize: 13, + inputs: [], + outputs: [], + }, + }, + { + id: ids.fileInput, + type: 'input', + position: { x: 80, y: 220 }, + data: { + label: 'CSV File', + nodeType: 'input', + config: { + inputType: 'file', + maxSizeKB: 5120, + allowedExtensions: ['.csv', '.tsv', '.txt'], + }, + inputs: [], + outputs: [{ id: 'text-out', type: 'string' }], + }, + }, + { + id: ids.fileEmbed, + type: 'embedding', + position: { x: 380, y: 220 }, + data: { + label: 'Embed CSV', + nodeType: 'embedding', + config: { + provider: 'openai', + model: 'text-embedding-3-small', + dimensions: 1536, + chunkSize: 512, + chunkOverlap: 64, + sourceField: 'auto', + }, + inputs: [{ id: 'source-in', type: 'any' }], + outputs: [{ id: 'vectors-out', type: 'json' }], + }, + }, + { + id: ids.vecIndex, + type: 'vector', + position: { x: 680, y: 220 }, + data: { + label: 'Index Store', + nodeType: 'vector', + config: { + storeName: STORE_NAME, + indexType: 'flat', + metric: 'cosine', + mode: 'index', + topK: 5, + topP: 0, + injectInto: 'context', + replace: false, + }, + inputs: [{ id: 'vectors-in', type: 'any' }], + outputs: [ + { id: 'context-out', type: 'string' }, + { id: 'json-out', type: 'json' }, + ], + }, + }, + + // ── Query path ─────────────────────────────────────────────────────── + { + id: ids.noteQuery, + type: 'text', + position: { x: 80, y: 380 }, + width: 320, + height: 48, + selectable: true, + draggable: true, + data: { + label: 'Text', + nodeType: 'text', + config: { + text: '② Ask — type a question to retrieve & answer', + color: '#60a5fa', + fontSize: 13, + }, + text: '② Ask — type a question to retrieve & answer', + color: '#60a5fa', + fontSize: 13, + inputs: [], + outputs: [], + }, + }, + { + id: ids.textInput, + type: 'input', + position: { x: 80, y: 450 }, + data: { + label: 'Question', + nodeType: 'input', + config: { inputType: 'text' }, + inputs: [], + outputs: [{ id: 'text-out', type: 'string' }], + }, + }, + { + id: ids.queryEmbed, + type: 'embedding', + position: { x: 380, y: 450 }, + data: { + label: 'Embed Query', + nodeType: 'embedding', + config: { + provider: 'openai', + model: 'text-embedding-3-small', + dimensions: 1536, + chunkSize: 512, + chunkOverlap: 64, + sourceField: 'auto', + }, + inputs: [{ id: 'source-in', type: 'any' }], + outputs: [{ id: 'vectors-out', type: 'json' }], + }, + }, + { + id: ids.vecQuery, + type: 'vector', + position: { x: 680, y: 450 }, + data: { + label: 'Retrieve', + nodeType: 'vector', + config: { + storeName: STORE_NAME, + indexType: 'flat', + metric: 'cosine', + mode: 'query', + topK: 5, + topP: 0, + injectInto: 'context', + }, + inputs: [{ id: 'vectors-in', type: 'any' }], + outputs: [ + { id: 'context-out', type: 'string' }, + { id: 'json-out', type: 'json' }, + ], + }, + }, + { + id: ids.llm, + type: 'llm', + position: { x: 980, y: 450 }, + data: { + label: 'LLM', + nodeType: 'llm', + config: { + provider: 'anthropic', + model: 'claude-sonnet-4-6', + temperature: 0.2, + maxTokens: 1000, + systemPrompt: + 'You answer using only the retrieved CSV context. ' + + 'Cite specific row values; if the answer is not in context, say so.', + }, + inputs: [ + { id: 'messages-in', type: 'messages' }, + { id: 'system-in', type: 'string' }, + ], + outputs: [ + { id: 'messages-out', type: 'messages' }, + { id: 'text-out', type: 'string' }, + ], + }, + }, + { + id: ids.output, + type: 'output', + position: { x: 1280, y: 450 }, + data: { + label: 'Answer', + nodeType: 'output', + config: {}, + inputs: [{ id: 'text-in', type: 'string' }], + outputs: [], + }, + }, + ] satisfies AgentNode[], + + edges: [ + { id: 'e1', source: ids.fileInput, sourceHandle: 'text-out', target: ids.fileEmbed, targetHandle: 'source-in' }, + { id: 'e2', source: ids.fileEmbed, sourceHandle: 'vectors-out', target: ids.vecIndex, targetHandle: 'vectors-in' }, + { id: 'e3', source: ids.textInput, sourceHandle: 'text-out', target: ids.queryEmbed, targetHandle: 'source-in' }, + { id: 'e4', source: ids.queryEmbed, sourceHandle: 'vectors-out', target: ids.vecQuery, targetHandle: 'vectors-in' }, + { id: 'e5', source: ids.vecQuery, sourceHandle: 'context-out', target: ids.llm, targetHandle: 'messages-in' }, + { id: 'e6', source: ids.llm, sourceHandle: 'text-out', target: ids.output, targetHandle: 'text-in' }, + ] satisfies AgentEdge[], +} + +export const SAMPLE_WORKFLOWS: SampleWorkflow[] = [ + { + id: 'rag-csv-embedding', + name: 'RAG over CSV', + tagline: 'Vectorize a CSV, then ask questions grounded in the data', + description: + 'Upload a CSV to embed and store as vectors, then ask natural-language ' + + 'questions. The same embedding model is used for indexing and querying so ' + + 'similarity scoring is consistent end-to-end.', + tags: ['RAG', 'Embedding', 'Vector Store', 'CSV'], + graph: csvRagFlow, + }, +] + +export function getSampleWorkflow(id: string): SampleWorkflow | undefined { + return SAMPLE_WORKFLOWS.find((w) => w.id === id) +} diff --git a/lib/types.ts b/lib/types.ts index 52b64fb..28109e5 100644 --- a/lib/types.ts +++ b/lib/types.ts @@ -4,7 +4,8 @@ import type { DBRunSnapshot, DBSchema, DBQueryResult } from '@/types/db' export type NodeType = | 'input' | 'prompt' | 'llm' | 'tool' - | 'memory' | 'mcp' | 'rag' | 'guardrail' | 'database' | 'output' + | 'memory' | 'mcp' | 'rag' | 'guardrail' | 'database' + | 'embedding' | 'vector' | 'output' export interface GraphNode { id: string @@ -124,16 +125,41 @@ export interface TokenUsage { firstTokenMs?: number // ms from LLM request start to first token } +export interface EmbeddingChunk { + content: string + embedding: number[] + metadata?: Record +} + +export interface EmbeddingPayload { + provider: string + model: string + dimensions: number + chunks: EmbeddingChunk[] + /** Last embedded query, when produced for retrieval rather than indexing. */ + query?: { text: string; embedding: number[] } +} + +export interface RagContext { + storeName: string + hits: Array<{ content: string; score: number; metadata: Record }> +} + export interface NodeContext { input?: string output?: string messages?: ModelMessage[] fileData?: FileData usage?: TokenUsage + // Run scope — set by execute() so handlers can persist per-flow state + flowId?: string // Database-node propagated state for downstream nodes dbSchema?: DBSchema dbRows?: DBQueryResult['rows'] db?: DBRunSnapshot + // Embedding / vector pipeline state + embeddings?: EmbeddingPayload + rag?: RagContext [key: string]: unknown } diff --git a/services/vector-store.ts b/services/vector-store.ts new file mode 100644 index 0000000..bae59c1 --- /dev/null +++ b/services/vector-store.ts @@ -0,0 +1,342 @@ +// Vector store service. Wraps: +// • embedding generation via the AI SDK (`embedMany` / `embed`) +// • upsert + similarity query against the `embeddings` / `vector_stores` tables +// +// Storage shape is portable JSON-backed (jsonb embedding column). Similarity +// scoring runs in-process so the same code path works whether or not the +// pgvector extension is enabled. When pgvector is later added, the query path +// can be swapped for a server-side `<=> ` operator without changing callers. + +import { and, eq } from 'drizzle-orm' +import { embed, embedMany } from 'ai' +import { createOpenAI } from '@ai-sdk/openai' +import { db } from '@/lib/db/client' +import { embeddings, vectorStores } from '@/lib/db/schema' +import { LLMManager } from '@/lib/llm-manager' +import { PROVIDERS, type ProviderId } from '@/lib/providers/registry' +import type { SessionKeys } from '@/lib/types' + +export type IndexType = 'flat' | 'hnsw' | 'ivfflat' +export type Metric = 'cosine' | 'l2' | 'dot' + +export interface EmbedTextsArgs { + provider: ProviderId + model: string + texts: string[] + userId: string + sessionKeys?: SessionKeys +} + +export interface ChunkRecord { + content: string + embedding: number[] + metadata?: Record +} + +export interface UpsertArgs { + flowId: string + nodeId: string + storeName: string + provider: ProviderId + model: string + dimensions: number + indexType: IndexType + metric: Metric + config?: Record + records: ChunkRecord[] + /** When true, deletes any existing rows in the store before inserting. */ + replace?: boolean +} + +export interface QueryArgs { + flowId: string + nodeId: string + storeName?: string + queryEmbedding: number[] + topK: number + /** Min similarity score (0..1 for cosine). 0 disables thresholding. */ + threshold?: number +} + +export interface QueryHit { + id: string + content: string + score: number + metadata: Record +} + +// ─── Embedding ──────────────────────────────────────────────────────────────── + +/** + * Resolve an embedding model instance for the user's configured provider. + * Only OpenAI-compatible providers expose `textEmbeddingModel`; others throw. + * The user-facing error is what surfaces in the node UI. + */ +async function resolveEmbeddingModel( + provider: ProviderId, + modelId: string, + userId: string, + sessionKeys?: SessionKeys, +) { + const apiKey = await LLMManager.forUser(userId).getKey(provider, sessionKeys) + if (!apiKey) { + throw new Error(`No API key configured for "${provider}" — open the Embedding node and add one.`) + } + const baseURL = PROVIDERS[provider].baseUrl + if (provider === 'anthropic') { + throw new Error('Anthropic does not expose embedding models. Use OpenAI or Google for embeddings.') + } + const openai = createOpenAI({ apiKey, ...(baseURL ? { baseURL } : {}) }) + return openai.textEmbeddingModel(modelId) +} + +export async function embedTexts(args: EmbedTextsArgs): Promise { + const model = await resolveEmbeddingModel(args.provider, args.model, args.userId, args.sessionKeys) + if (args.texts.length === 0) return [] + if (args.texts.length === 1) { + const { embedding } = await embed({ model, value: args.texts[0] }) + return [embedding] + } + const { embeddings: vecs } = await embedMany({ model, values: args.texts }) + return vecs +} + +export async function embedQuery(args: Omit & { text: string }): Promise { + const model = await resolveEmbeddingModel(args.provider, args.model, args.userId, args.sessionKeys) + const { embedding } = await embed({ model, value: args.text }) + return embedding +} + +// ─── Store lifecycle ────────────────────────────────────────────────────────── + +/** + * Returns the vector store row for `(flowId, nodeId)`, creating it on first + * call. Subsequent calls update mutable config fields so the store always + * reflects the node's current configuration. + */ +export async function ensureStore(args: { + flowId: string + nodeId: string + storeName: string + provider: ProviderId + model: string + dimensions: number + indexType: IndexType + metric: Metric + config?: Record +}) { + const [existing] = await db + .select() + .from(vectorStores) + .where(and(eq(vectorStores.flowId, args.flowId), eq(vectorStores.nodeId, args.nodeId))) + .limit(1) + + if (existing) { + if ( + existing.model !== args.model || + existing.provider !== args.provider || + existing.dimensions !== args.dimensions || + existing.indexType !== args.indexType || + existing.metric !== args.metric || + existing.name !== args.storeName + ) { + const [updated] = await db + .update(vectorStores) + .set({ + name: args.storeName, + provider: args.provider, + model: args.model, + dimensions: args.dimensions, + indexType: args.indexType, + metric: args.metric, + config: args.config ?? {}, + }) + .where(eq(vectorStores.id, existing.id)) + .returning() + return updated + } + return existing + } + + const [created] = await db + .insert(vectorStores) + .values({ + flowId: args.flowId, + nodeId: args.nodeId, + name: args.storeName, + provider: args.provider, + model: args.model, + dimensions: args.dimensions, + indexType: args.indexType, + metric: args.metric, + config: args.config ?? {}, + }) + .returning() + return created +} + +export async function upsertEmbeddings(args: UpsertArgs): Promise<{ storeId: string; inserted: number }> { + const store = await ensureStore({ + flowId: args.flowId, + nodeId: args.nodeId, + storeName: args.storeName, + provider: args.provider, + model: args.model, + dimensions: args.dimensions, + indexType: args.indexType, + metric: args.metric, + config: args.config, + }) + + if (args.replace) { + await db.delete(embeddings).where(eq(embeddings.storeId, store.id)) + } + + if (args.records.length === 0) return { storeId: store.id, inserted: 0 } + + await db.insert(embeddings).values( + args.records.map((r, i) => ({ + storeId: store.id, + flowId: args.flowId, + chunkIndex: i, + content: r.content, + embedding: r.embedding, + metadata: r.metadata ?? {}, + })), + ) + return { storeId: store.id, inserted: args.records.length } +} + +// ─── Query ──────────────────────────────────────────────────────────────────── + +function dot(a: number[], b: number[]): number { + const n = Math.min(a.length, b.length) + let s = 0 + for (let i = 0; i < n; i++) s += a[i] * b[i] + return s +} + +function norm(v: number[]): number { + let s = 0 + for (const x of v) s += x * x + return Math.sqrt(s) +} + +function cosine(a: number[], b: number[]): number { + const denom = norm(a) * norm(b) + return denom === 0 ? 0 : dot(a, b) / denom +} + +function l2Distance(a: number[], b: number[]): number { + const n = Math.min(a.length, b.length) + let s = 0 + for (let i = 0; i < n; i++) { + const d = a[i] - b[i] + s += d * d + } + return Math.sqrt(s) +} + +function score(metric: Metric, a: number[], b: number[]): number { + if (metric === 'cosine') return cosine(a, b) + if (metric === 'dot') return dot(a, b) + // l2: convert distance to similarity in (0, 1] + return 1 / (1 + l2Distance(a, b)) +} + +export async function queryStore(args: QueryArgs): Promise<{ storeId: string | null; hits: QueryHit[] }> { + // Resolve the store row. Primary lookup is (flowId, nodeId) — that row exists + // when the same node both indexed and is now querying. When a storeName is + // provided we also fall back to (flowId, storeName) so a separate Vector node + // (or another node that previously seeded the store) can be reached by name. + let [store] = await db + .select() + .from(vectorStores) + .where(and(eq(vectorStores.flowId, args.flowId), eq(vectorStores.nodeId, args.nodeId))) + .limit(1) + + if ((!store || (args.storeName && store.name !== args.storeName)) && args.storeName) { + const named = await db + .select() + .from(vectorStores) + .where(and(eq(vectorStores.flowId, args.flowId), eq(vectorStores.name, args.storeName))) + .limit(1) + if (named[0]) store = named[0] + } + + if (!store) return { storeId: null, hits: [] } + if (args.storeName && store.name !== args.storeName) return { storeId: store.id, hits: [] } + + const rows = await db + .select({ + id: embeddings.id, + content: embeddings.content, + embedding: embeddings.embedding, + metadata: embeddings.metadata, + }) + .from(embeddings) + .where(eq(embeddings.storeId, store.id)) + + const metric = (store.metric ?? 'cosine') as Metric + const scored = rows.map((r) => ({ + id: r.id, + content: r.content, + metadata: (r.metadata as Record) ?? {}, + score: score(metric, args.queryEmbedding, r.embedding as unknown as number[]), + })) + + scored.sort((a, b) => b.score - a.score) + const threshold = args.threshold ?? 0 + const filtered = threshold > 0 ? scored.filter((h) => h.score >= threshold) : scored + return { storeId: store.id, hits: filtered.slice(0, Math.max(1, args.topK)) } +} + +// ─── Chunking ───────────────────────────────────────────────────────────────── +// Recursive character splitter: paragraphs → sentences → words. Token counts +// are approximated by chars/4 (good enough for chunk sizing without bringing +// in a tokenizer). + +const SEPARATORS = ['\n\n', '\n', '. ', ' ', ''] + +function splitRecursive(text: string, sepIdx: number, target: number): string[] { + if (text.length <= target) return [text] + const sep = SEPARATORS[sepIdx] ?? '' + if (sep === '') { + const out: string[] = [] + for (let i = 0; i < text.length; i += target) out.push(text.slice(i, i + target)) + return out + } + const parts = text.split(sep) + const out: string[] = [] + let buf = '' + for (const p of parts) { + const candidate = buf ? buf + sep + p : p + if (candidate.length <= target) { + buf = candidate + continue + } + if (buf) out.push(buf) + if (p.length > target) { + out.push(...splitRecursive(p, sepIdx + 1, target)) + buf = '' + } else { + buf = p + } + } + if (buf) out.push(buf) + return out +} + +export function chunkText(text: string, chunkSize = 512, overlap = 64): string[] { + // chunkSize is in tokens (~4 chars/token); convert to chars + const targetChars = Math.max(64, chunkSize * 4) + const overlapChars = Math.max(0, Math.min(overlap * 4, targetChars - 1)) + const pieces = splitRecursive(text.trim(), 0, targetChars).filter((p) => p.trim().length > 0) + if (overlapChars === 0 || pieces.length <= 1) return pieces + const out: string[] = [pieces[0]] + for (let i = 1; i < pieces.length; i++) { + const tail = out[out.length - 1].slice(-overlapChars) + out.push(tail + pieces[i]) + } + return out +} diff --git a/store/index.ts b/store/index.ts index f222c5b..fc71dc2 100644 --- a/store/index.ts +++ b/store/index.ts @@ -15,14 +15,14 @@ import type { TokenUsage, ToolRunSnapshot } from '@/lib/types' // ─── Port & Node types ──────────────────────────────────────────────────────── export type PortType = 'messages' | 'string' | 'json' | 'any' -export type AgentNodeKind = 'input' | 'prompt' | 'llm' | 'tool' | 'memory' | 'database' | 'output' +export type AgentNodeKind = 'input' | 'prompt' | 'llm' | 'tool' | 'memory' | 'database' | 'embedding' | 'vector' | 'output' export type AnnotationKind = 'shape' | 'text' | 'drawing' | 'arrow' export type NodeKind = AgentNodeKind | AnnotationKind export type RunStatus = 'idle' | 'running' | 'done' | 'error' export type DrawingTool = 'select' | 'rectangle' | 'ellipse' | 'pen' | 'text' | 'arrow' -export const AGENT_NODE_KINDS: readonly AgentNodeKind[] = ['input', 'prompt', 'llm', 'tool', 'memory', 'database', 'output'] as const +export const AGENT_NODE_KINDS: readonly AgentNodeKind[] = ['input', 'prompt', 'llm', 'tool', 'memory', 'database', 'embedding', 'vector', 'output'] as const export function isAnnotationKind(k: string | undefined): k is AnnotationKind { return k === 'shape' || k === 'text' || k === 'drawing' || k === 'arrow' @@ -58,6 +58,11 @@ export interface RunMeta { command?: string // SQL command verb (SELECT | INSERT | …) truncated?: boolean // result set was clipped to rowLimit errorMsg?: string // generic error message (DB and similar) + // Embedding / Vector node fields + vectorCount?: number // number of vectors written or returned + dimensions?: number // embedding dimensions + topK?: number // retriever top-k actually used + mode?: string // vector op: 'index' | 'query' | embedding mode label } export interface NodeData { @@ -82,6 +87,8 @@ interface State { // graph nodes: AgentNode[] edges: AgentEdge[] + // flow + flowId: string | null // run runId: string | null // ui @@ -99,6 +106,8 @@ interface Actions { addNode: (node: AgentNode) => void updateNodeData: (id: string, partial: Partial) => void loadGraph: (nodes: AgentNode[], edges: AgentEdge[]) => void + // flow + setFlowId: (id: string | null) => void // run setRunId: (id: string | null) => void setRunStatus: (nodeId: string, status: RunStatus, output?: string, meta?: RunMeta) => void @@ -120,6 +129,7 @@ export const useStore = create()((set) => ({ // ── initial state ────────────────────────────────────────────────────────── nodes: [], edges: [], + flowId: null, runId: null, selectedNodeId: null, sidebarOpen: false, @@ -154,6 +164,9 @@ export const useStore = create()((set) => ({ loadGraph: (nodes, edges) => set({ nodes, edges }), + // ── flow actions ─────────────────────────────────────────────────────────── + setFlowId: (id) => set({ flowId: id }), + // ── run actions ──────────────────────────────────────────────────────────── setRunId: (id) => set({ runId: id }),