Configure the Pipeline for Memory Extraction and Processing in TencentDB Agent Memory

You configure the memory extraction pipeline by instantiating the PipelineFactoryOptions interface in MemoryCore/src/utils/pipeline-factory.ts, which wires together L1 extraction, L2 enrichment, L3 consolidation, and storage backends, then launching the MemoryPipelineManager scheduler.

TencentDB Agent Memory progressively refines conversational data from raw logs (L0) into structured memory representations (L1→L2→L3) using a factory-based pipeline architecture. The entire workflow is orchestrated through the pipeline factory located at MemoryCore/src/utils/pipeline-factory.ts, which constructs a MemoryPipelineManager to handle batch processing, back-pressure, and persistence.

Understanding the Memory Extraction Pipeline Architecture

The pipeline transforms data through four distinct stages, each implemented as a configurable runner:

• L1 Extraction – Pulls raw conversation rows from L0 storage and feeds them to an LLM (host-neutral with enableTools = false) to generate initial memory blocks (persona, scene, etc.). Implemented in extractL1Memories within core/record/l1-extractor.ts.
• L2 Processing – Resolves memory prompts and runs additional LLM calls (tool-enabled) to enrich each block with provenance metadata. Created via createL2Runner in the factory.
• L3 Consolidation – Persists blocks to a vector store, updates profile scopes, and writes a manifest for downstream services. Created via createL3Runner.
• Persistence Layer – Stores vectors, embeddings, and metadata using either LocalStorageBackend for SQLite or a COS-backed remote store via createStoreBundle.

Pipeline Factory and Core Components

The entry point for all configuration is the createPipeline function exported from MemoryCore/src/utils/pipeline-factory.ts (lines 73–88 define the core interface). This factory aggregates multiple subsystems into a single MemoryPipelineManager instance that manages the lifecycle of memory extraction.

Key internal components include:

• Session Filtering – Optional SessionFilter class to limit processing to specific conversation sessions.
• Profile Scoping – Isolation functions (buildProfileL2Key, scopedStorage, scopedDataDir) that ensure multi-tenant data separation.
• Storage Adapters – Abstractions over local SQLite and remote COS storage, selected via the configuration object.

Configuring PipelineFactoryOptions

The PipelineFactoryOptions interface controls every aspect of pipeline behavior. You must provide these fields when calling createPipeline:

Required Configuration Fields

Field	Purpose	Example Value
pluginDataDir	Root directory containing raw L0 data (conversation/), interim records, and vector stores	/var/lib/tdai/data
cfg	Parsed MemoryTdaiConfig (loaded from YAML) containing global settings, batch sizes, and feature toggles	Loaded from tdai-gateway.yaml
openclawConfig	Configuration for the OpenClaw LLM service; required for L1 extraction	cfg.openclaw
logger	Logger implementation conforming to the Logger type from core/types.js	console or Pino instance

Optional Customization

You can further refine behavior with these optional parameters:

• sessionFilter – Instance of SessionFilter to whitelist or blacklist specific session IDs.
• l1LlmRunner – Custom LLM runner for L1 extraction if you need to override the default OpenClaw client.
• l2l3LlmRunner – Custom runner for L2/L3 stages when you require different model endpoints or parameters.

Step-by-Step Configuration Guide

Follow these steps to initialize and run the pipeline:

1. Create the YAML configuration file (e.g., tdai-gateway.yaml):

pluginDataDir: /var/lib/tdai/data
storage:
  backend: local   # or "cos"

  path: vectors.db
l1:
  batchProcess: 10          # matches L1_BATCH_PROCESS constant

  llm:
    model: qwen-7b-chat
    temperature: 0.7
l2l3:
  llm:
    model: qwen-14b-chat
    enableTools: true
profileScope: default

2. Load the configuration in your entry script:

import { loadConfig } from "./config.js";
const cfg = await loadConfig("/etc/tdai/tdai-gateway.yaml");

3. Instantiate the pipeline using the factory:

import { createPipeline } from "./utils/pipeline-factory.js";

const pipeline = await createPipeline({
  pluginDataDir: cfg.pluginDataDir,
  cfg,
  openclawConfig: cfg.openclaw,
  logger: console,
  sessionFilter: new SessionFilter(), // optional
});

4. Start the scheduler to begin periodic processing:

await pipeline.scheduler.start();

The MemoryPipelineManager handles periodic L1 runs, idle timers, and back-pressure automatically.

5. Adjust batch sizes globally (optional):

import { L1_BATCH_PROCESS, L1_BATCH_QUERY } from "./utils/pipeline-factory.js";

// Must be done before createPipeline is called
(global as any).L1_BATCH_PROCESS = 20;
(global as any).L1_BATCH_QUERY = 40;

Profile Isolation and Multi-Tenancy

TencentDB Agent Memory supports multi-tenant deployments through profile scoping. When you set a non-default profileScope in your configuration, the pipeline automatically prefixes all vector store keys and on-disk directories using encoding functions like buildProfileL2Key and scopedStorage. This ensures complete data isolation between users or tenants without requiring separate code paths.

To enable isolation, set the profileScope field in your YAML configuration:

profileScope: user-tenant-123

All subsequent storage operations in LocalStorageBackend or COS will use scoped paths derived from scopedDataDir.

Manual Execution and Debugging

For debugging or one-off processing, you can run extraction stages manually without the scheduler:

Run a single L1 extraction batch:

import { extractL1Memories } from "../core/record/l1-extractor.js";
import { readConversationMessagesGroupedBySessionId } from "../core/conversation/l0-recorder.js";

async function runOneBatch() {
  const msgs = await readConversationMessagesGroupedBySessionId("/data/conversation");
  const l1Blocks = await extractL1Memories(msgs, { /* LLM runner options */ });
  console.log("Generated", l1Blocks.length, "L1 memory blocks");
}
runOneBatch();

Apply a custom session filter:

import { SessionFilter } from "./utils/session-filter.js";

const filter = new SessionFilter();
filter.addIncludePattern(/^user-123/);

await createPipeline({
  pluginDataDir: "/data",
  cfg,
  openclawConfig: cfg.openclaw,
  logger: console,
  sessionFilter: filter,
});

Summary

• The pipeline factory at MemoryCore/src/utils/pipeline-factory.ts constructs the complete extraction workflow via createPipeline.
• Configuration is driven by the PipelineFactoryOptions interface, requiring pluginDataDir, cfg, openclawConfig, and a logger.
• The pipeline processes data through L1 (extraction), L2 (enrichment), and L3 (consolidation) stages, persisting results to SQLite or COS storage.
• Profile scoping enables multi-tenancy by automatically prefixing storage keys and directories.
• Batch sizes and session filters can be customized to control throughput and data selection.

Frequently Asked Questions

What is the difference between L1, L2, and L3 memory processing?

L1 extraction generates initial memory blocks from raw conversations using a host-neutral LLM without tools. L2 processing enriches these blocks with additional metadata and provenance using tool-enabled LLM calls. L3 consolidation persists the enriched blocks to vector storage and updates the user profile manifest.

How do I enable multi-tenant isolation in the pipeline?

Set a unique profileScope value in your MemoryTdaiConfig (e.g., profileScope: tenant-abc). The pipeline automatically applies functions like buildProfileL2Key and scopedStorage to encode the scope into all vector keys and file paths, ensuring data separation between tenants.

Can I use a custom LLM provider instead of OpenClaw?

Yes. Pass custom runners via the l1LlmRunner and l2l3LlmRunner fields in PipelineFactoryOptions. These replace the default OpenClaw clients for their respective stages, allowing you to integrate alternative LLM services while maintaining the same extraction logic.

How do I debug or manually trigger a single extraction batch?

Import extractL1Memories from core/record/l1-extractor.ts and readConversationMessagesGroupedBySessionId from core/conversation/l0-recorder.ts. Load your conversation data and call extractL1Memories directly with your LLM options to generate blocks without starting the full scheduler.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}