importflow

package
v2.54.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 18, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package importflow imports external structured data (CSV, MySQL/PG SQL dumps) into CortexDB, building RAG (vector/FTS5) and knowledge-graph (RDF triple) foundations in a single pass. AI assistance (mapping inference, triple extraction, text refinement) is injected via interfaces and always optional; this package imports no LLM SDK.

pkg/importflow/importer.go

pkg/importflow/infer.go

pkg/importflow/refine.go

pkg/importflow/sink_kg.go

pkg/importflow/sink_rag.go

pkg/importflow/source.go

pkg/importflow/source_csv.go

pkg/importflow/source_dump.go

pkg/importflow/toolbox.go

pkg/importflow/types.go

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func MappingFromDDL added in v2.25.0

func MappingFromDDL(ddl string, opts DDLMappingOptions) (MappingPlan, []DDLTable, error)

MappingFromDDL parses DDL and derives a deterministic MappingPlan: tables -> entities (PK -> id), foreign keys -> referenced entities + relations, non-key columns -> RAG content + entity props. It also returns the parsed tables.

func MappingFromDDLWithLLM added in v2.25.0

func MappingFromDDLWithLLM(ctx context.Context, ddl string, gen graphflow.JSONGenerator, opts DDLMappingOptions) (MappingPlan, []DDLTable, bool, error)

MappingFromDDLWithLLM parses DDL, builds the deterministic baseline plan, and asks the LLM to refine it (semantic naming, implicit relations, free-text TextExtract, junction-table collapse). It returns the refined plan, the parsed tables, and llmUsed=false (with the baseline plan) when the LLM is unavailable or returns an unusable result. err is non-nil only on a hard failure (bad DDL or nil generator).

func NewMCPServer added in v2.20.1

func NewMCPServer(im *Importer, opts MCPServerOptions) (*mcp.Server, error)

NewMCPServer returns an MCP server that exposes the importflow tool surface (importflow_plan, importflow_run) backed by the given Importer.

func RunMCPStdio added in v2.20.1

func RunMCPStdio(ctx context.Context, im *Importer, opts MCPServerOptions) error

RunMCPStdio runs the importflow MCP server over stdio until the context is done.

Types

type CSVOptions

type CSVOptions struct {
	Table      string // logical table name; default "csv"
	Delimiter  rune   // default ','
	SampleSize int    // sample rows in Schemas(); default 5
}

CSVOptions configures a CSVSource.

type CSVSource

type CSVSource struct {
	// contains filtered or unexported fields
}

CSVSource is an in-memory Source over a single CSV stream (header required).

func NewCSVSource

func NewCSVSource(r io.Reader, opts CSVOptions) (*CSVSource, error)

NewCSVSource eagerly reads the whole CSV into memory and infers column types.

func (*CSVSource) Close

func (s *CSVSource) Close() error

func (*CSVSource) Records

func (s *CSVSource) Records(ctx context.Context, fn func(Record) error) error

func (*CSVSource) Schemas

func (s *CSVSource) Schemas(_ context.Context) ([]Schema, error)

type Column

type Column struct {
	Name string
	Type string // "integer","number","text","timestamp","" (unknown)
}

Column is one source column with a best-effort type label.

type DDLColumn added in v2.25.0

type DDLColumn struct {
	Name string
	Type string // normalized: text/integer/number/timestamp/"" (unknown)
}

DDLColumn is one declared column.

type DDLForeignKey added in v2.25.0

type DDLForeignKey struct {
	Column    string
	RefTable  string
	RefColumn string
}

DDLForeignKey is one foreign key: Column references RefTable(RefColumn).

type DDLMappingOptions added in v2.25.0

type DDLMappingOptions struct {
	// RelationStyle picks the relation predicate from a foreign key:
	//   "" / "column": strip a trailing _id from the FK column (customer_id -> "customer"),
	//                  fallback "references_<reftable>".
	//   "reftable":    predicate = "references_<reftable>".
	RelationStyle string
	IncludeRAG    *bool // default true
	IncludeKG     *bool // default true
}

DDLMappingOptions tunes MappingFromDDL.

type DDLTable added in v2.25.0

type DDLTable struct {
	Name        string
	Columns     []DDLColumn
	PrimaryKey  []string
	ForeignKeys []DDLForeignKey
}

DDLTable is one parsed CREATE TABLE statement.

func ParseDDL added in v2.25.0

func ParseDDL(ddl string) ([]DDLTable, error)

ParseDDL parses a practical subset of Postgres/MySQL CREATE TABLE statements (columns, PRIMARY KEY, FOREIGN KEY/REFERENCES). Statements it cannot parse are skipped; it errors only when no CREATE TABLE was found at all.

type Dialect

type Dialect string

Dialect selects dump-specific parsing rules.

const (
	DialectAuto     Dialect = "auto"
	DialectMySQL    Dialect = "mysql"
	DialectPostgres Dialect = "postgres"
)

type DumpOptions

type DumpOptions struct {
	Dialect    Dialect // default DialectAuto
	SampleSize int     // sample rows per table in Schemas(); default 5
}

DumpOptions configures a SQLDumpSource.

type EntityMap

type EntityMap struct {
	Ref       string   `json:"ref"`     // local handle, e.g. "customer"
	Type      string   `json:"type"`    // entity class
	IDTmpl    string   `json:"id_tmpl"` // "{customer_id}"
	LabelTmpl string   `json:"label_tmpl,omitempty"`
	Props     []string `json:"props,omitempty"`
}

EntityMap maps columns to one entity per row.

type Goal

type Goal struct {
	BuildRAG bool
	BuildKG  bool
	Hint     string // domain hint passed to the AI inferer
}

Goal declares what the import should build.

type Importer

type Importer struct {
	// contains filtered or unexported fields
}

Importer orchestrates Source -> MappingPlan -> RAG + KG sinks.

func New

func New(db *cortexdb.DB, opts ...Option) *Importer

New constructs an Importer over an open cortexdb.DB.

func (*Importer) AutoImport

func (im *Importer) AutoImport(ctx context.Context, src Source, goal Goal) (*Report, error)

AutoImport runs Plan then Run in one step.

func (*Importer) Plan

func (im *Importer) Plan(ctx context.Context, src Source, goal Goal) (MappingPlan, error)

Plan asks the configured MappingInferer to propose a plan from the source schemas.

func (*Importer) Run

func (im *Importer) Run(ctx context.Context, src Source, plan MappingPlan) (*Report, error)

Run streams every record through the plan into the RAG and KG sinks.

type KGPlan

type KGPlan struct {
	Entities    []EntityMap   `json:"entities,omitempty"`
	Relations   []RelationMap `json:"relations,omitempty"`
	TextExtract []TextExtract `json:"text_extract,omitempty"`
}

KGPlan describes how a row becomes entities and relations.

type LLMInferer

type LLMInferer struct {
	Client graphflow.JSONGenerator
}

LLMInferer is the default MappingInferer, backed by a graphflow.JSONGenerator.

func (LLMInferer) InferPlan

func (l LLMInferer) InferPlan(ctx context.Context, schemas []Schema, goal Goal) (MappingPlan, error)

type LLMRefiner

type LLMRefiner struct {
	Client graphflow.JSONGenerator
}

LLMRefiner is the default TextRefiner, backed by a graphflow.JSONGenerator.

func (LLMRefiner) Refine

func (l LLMRefiner) Refine(ctx context.Context, table, column, raw string) (string, error)

type MCPServerOptions added in v2.20.1

type MCPServerOptions struct {
	Implementation *mcp.Implementation
	Instructions   string
	Logger         *slog.Logger
}

MCPServerOptions configures the importflow MCP server wrapper.

type MappingInferer

type MappingInferer interface {
	InferPlan(ctx context.Context, schemas []Schema, goal Goal) (MappingPlan, error)
}

MappingInferer proposes a MappingPlan from table schemas and a goal.

type MappingPlan

type MappingPlan struct {
	Tables map[string]TablePlan `json:"tables"`
}

MappingPlan declares, per table, how source rows route to RAG and KG.

type Option

type Option func(*Importer)

Option configures an Importer.

func WithBatchSize

func WithBatchSize(n int) Option

WithBatchSize sets the sink batch size (default 500).

func WithExtractor

func WithExtractor(e graphflow.Extractor) Option

WithExtractor sets the graphflow extractor used for KGPlan.TextExtract columns.

func WithMappingInferer

func WithMappingInferer(m MappingInferer) Option

WithMappingInferer sets the AI mapping inferer used by Plan/AutoImport.

func WithStrictMode

func WithStrictMode() Option

WithStrictMode aborts the run on the first row error instead of collecting it.

func WithTextRefiner

func WithTextRefiner(r TextRefiner) Option

WithTextRefiner sets the AI text refiner used when RAGPlan.Refine is true.

type RAGPlan

type RAGPlan struct {
	Namespace   string   `json:"namespace,omitempty"`
	ContentTmpl string   `json:"content_tmpl"`        // "{title}\n\n{body}"
	IDColumn    string   `json:"id_column,omitempty"` // default synthesized "table:row"
	Metadata    []string `json:"metadata,omitempty"`  // columns copied into metadata
	Refine      bool     `json:"refine,omitempty"`    // run TextRefiner before embedding
}

RAGPlan describes how a row becomes a retrievable text chunk.

type Record

type Record struct {
	Table  string
	Values map[string]string // column name -> string value
	Nulls  map[string]bool   // column name -> is NULL
	Row    int               // 0-based row index within the table
}

Record is one normalized source row.

func (Record) Get

func (r Record) Get(col string) (string, bool)

Get returns the value for col and false when the column is missing or NULL.

type RelationMap

type RelationMap struct {
	Subject   string `json:"subject"`
	Predicate string `json:"predicate"`
	Object    string `json:"object"`
}

RelationMap connects two entity refs with a predicate.

type Report

type Report struct {
	RowsRead           int
	ChunksIndexed      int
	TriplesCreated     int
	Skipped            int
	UnparsedStatements []string
	Errors             []error
}

Report summarizes an import run. Per repo "no silent caps" rule, dropped / unparsed input is surfaced here rather than discarded silently.

type SQLDumpSource

type SQLDumpSource struct {
	// contains filtered or unexported fields
}

SQLDumpSource parses a common subset of MySQL/PG dumps: CREATE TABLE (for column order), INSERT INTO ... VALUES (...), and PG COPY ... \. blocks. Unrecognized statements are recorded in Unparsed() instead of being dropped.

func NewSQLDumpSource

func NewSQLDumpSource(r io.Reader, opts DumpOptions) (*SQLDumpSource, error)

NewSQLDumpSource parses the entire dump eagerly.

func (*SQLDumpSource) Close

func (s *SQLDumpSource) Close() error

func (*SQLDumpSource) Records

func (s *SQLDumpSource) Records(ctx context.Context, fn func(Record) error) error

func (*SQLDumpSource) Schemas

func (s *SQLDumpSource) Schemas(_ context.Context) ([]Schema, error)

func (*SQLDumpSource) Unparsed

func (s *SQLDumpSource) Unparsed() []string

Unparsed returns statements the parser could not handle.

type Schema

type Schema struct {
	Table   string
	Columns []Column
	Sample  []Record
}

Schema describes a table plus sample rows, used for AI mapping inference.

type Source

type Source interface {
	// Schemas returns table schemas with up to a few sample rows each.
	Schemas(ctx context.Context) ([]Schema, error)
	// Records streams every record; fn returning an error aborts iteration.
	Records(ctx context.Context, fn func(Record) error) error
	// Close releases any underlying resources.
	Close() error
}

Source yields table schemas (for planning) and a stream of records (for import). Implementations should make Schemas() cheap enough to call before Records().

type TablePlan

type TablePlan struct {
	Skip bool     `json:"skip,omitempty"`
	RAG  *RAGPlan `json:"rag,omitempty"`
	KG   *KGPlan  `json:"kg,omitempty"`
}

TablePlan is the per-table routing decision.

type TextExtract

type TextExtract struct {
	Column string   `json:"column"`
	Types  []string `json:"types,omitempty"`
}

TextExtract names a free-text column for AI triple extraction.

type TextRefiner

type TextRefiner interface {
	Refine(ctx context.Context, table, column, raw string) (string, error)
}

TextRefiner cleans/summarizes a raw column value before embedding.

type Toolbox

type Toolbox struct {
	// contains filtered or unexported fields
}

Toolbox exposes importflow as agent-callable tools.

func NewToolbox

func NewToolbox(im *Importer) *Toolbox

NewToolbox wraps an Importer in a tool surface.

func (*Toolbox) Call

func (t *Toolbox) Call(ctx context.Context, name string, input json.RawMessage) (any, error)

Call dispatches a tool by name.

func (*Toolbox) Definitions

func (t *Toolbox) Definitions() []cortexdb.ToolDefinition

Definitions returns the tool definitions (mirrors graphflow's toolbox shape).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL