---

name: ingestion-pipelines description: "Operate M-flow ingestion, loader selection, memorization stages, procedural learning, and custom pipeline workflows." disable-model-invocation: true metadata: disco-role: operating license: Apache 2.0

M-flow ingestion pipelines

Use this sub-skill when the task is about getting data into M-flow memory or assembling ingestion/memorization pipelines:

• choose input shapes, loaders, preferred_loaders, chunking, and content_type;
• run one-step ingest(), staged add() then memorize(), or advanced run_custom_pipeline() workflows;
• configure memorization layers: episodic, procedural, content routing, episode routing, precise mode, and conflict handling;
• extract procedural memory later with learn() from already-built episodic memory.

Route elsewhere:

• basic CLI/API quickstarts, dataset CRUD, and first-time examples → core-memory-api;
• RecallMode, search(), query(), graph/vector adapter tuning, and retrieval ranking → retrieval-graph-search;
• running API servers, UI, MCP, Docker, and service integration topology → service-integrations.

First decisions

1. Separate ingestion from memorization when you need control. Use ingest() for the common add+memorize path. Use add() then memorize() when you need to validate stored data first, use different loader/chunking settings, skip/query later, or recover from a failed memorize stage.
2. Declare content shape before LLM routing. With content routing enabled, pass content_type=ContentType.TEXT for articles/notes/code-like prose and content_type=ContentType.DIALOG for chats, interviews, meeting logs, or transcripts. For mixed article + transcript data, prefer separate calls/datasets per content type instead of one mixed call.
3. Choose memory layers intentionally. Episodic memory is enabled by default. Procedural extraction is opt-in during memorize(enable_procedural=True) or can be run later with learn().
4. Use conflict_mode="error" for production jobs. It prevents accidental concurrent memorize() runs over the same dataset. Use "warn" for interactive work and "ignore" only when you accept duplicate/inconsistent graph risk.
5. Inspect the runtime before advising. Run scripts/pipeline_stage_inspector.py to list the installed loader registry and public signatures without executing a pipeline.

Common recipes

One-step ingestion with explicit routing

import m_flow
from m_flow import ContentType

result = await m_flow.ingest(
    data=["meeting-notes.txt", "transcript.txt"],
    dataset_name="team-memory",
    content_type=ContentType.DIALOG,
    enable_content_routing=True,
    enable_procedural=True,
    conflict_mode="error",
)

if result.needs_retry():
    # add() succeeded but memorize() failed; retry only the dataset's memorize stage.
    await m_flow.memorize(datasets=[result.dataset_name], content_type=ContentType.DIALOG)

Staged add then memorize

import m_flow
from m_flow import ContentType

await m_flow.add(
    data=["docs/overview.md", "docs/reference.pdf"],
    dataset_name="docs",
    preferred_loaders=[
        {"advanced_pdf_loader": {"strategy": "hi_res"}},
        "pypdf_loader",
        "text_loader",
    ],
)

await m_flow.memorize(
    datasets=["docs"],
    chunk_size=1200,
    chunks_per_batch=20,
    content_type=ContentType.TEXT,
    enable_content_routing=True,
    precise_mode=True,
    conflict_mode="error",
)

Learn procedures from existing episodic memory

import m_flow

# Prefer dataset-wide learning for duplicate prevention: learn() skips Episodes
# that already have derived_procedure edges when it discovers Episodes itself.
summary = await m_flow.learn(datasets=["team-memory"])
print(summary)

learn(run_in_background=True) is accepted by the public signature but currently logs a warning and runs synchronously so it can create derived_procedure edges after procedure persistence.

Reference map

• references/data-formats.md: accepted data shapes, loader registry, optional loader extras, preferred_loaders, and content-type decisions.
• references/pipeline-reference.md: memorize stage order, toggles, learn(), Stage, run_custom_pipeline(), and validation commands.
• references/troubleshooting.md: loader, path, routing, concurrency, procedural, and custom-pipeline failure signals.