How to Ingest Messages and Trigger the LI Pipeline Using the Conversation API in TencentDB Agent Memory

TLDR: Send a POST request to /v3/skill/conversation/add with your chat messages; the TencentDB-Agent-Memory service buffers the turn, automatically archives when thresholds are met, and triggers the downstream LI (Long-term-Inference) pipeline.

The TencentDB-Agent-Memory repository provides a per-turn incremental ingest flow for conversational AI applications. When you submit messages through the conversation API, the system stores them as a conversation buffer and launches the LI pipeline when archive criteria are satisfied. This article explains the exact request format, automatic triggering logic, and manual override options based on the source code implementation.

Conversation API Endpoint and Request Structure

The Core Endpoint: POST /v3/skill/conversation/add

The primary entry point for ingesting messages is implemented in MemoryCore/src/gateway/skill-handlers.ts. The handleConversationAdd function validates requests against conversationAddRequestSchema and resolves the per-instance ConversationAdd wiring.

from tencentdb_agent_memory.v3 import SkillClient

client = SkillClient(
    endpoint="https://your-instance.tencentyun.com",
    api_key="YOUR_API_KEY",
)

resp = client.conversation_add(
    session_id="sess-12345",
    user_id="u001",
    team_id="teamA",
    agent_id="agentX",
    messages=[
        {"role": "user", "content": "Hello, how are you?"},
        {"role": "assistant", "content": "I'm fine, thanks!"},
    ],
)

The Python SDK implementation resides in sdk/memory-core/python/tencentdb_agent_memory/v3/skill_client.py at lines 571–589.

Required and Optional Parameters

According to MemoryCore/src/gateway/skill-schemas.ts, the conversationAddRequestSchema defines the contract:

Required fields:

• session_id – unique conversation identifier
• user_id – end-user identifier
• team_id – organizational boundary
• agent_id – specific agent instance
• messages – array of role/content objects

Common optional fields:

• space_id – defaults to team_id
• task_id – for task-scoped memory
• metadata – arbitrary key-value pairs

How the Automatic Archive Trigger Works

The Three-Stage Processing Flow

The ingestion and LI pipeline triggering follows this architecture as implemented in the source code:

Stage 1: Gateway Validation (handleConversationAdd)

• Validates request schema (lines 19–31 in skill-handlers.ts)
• Resolves per-instance wiring configuration

Stage 2: Handler Processing (SkillConversationAddHandler)

• Located in core/skill/conversation-add/wire.js
• Packs incoming messages with compression/oversize handling
• Evaluates archive thresholds

Stage 3: Archive Trigger (wired.trigger.archive)

• Writes buffered segment to COS (Cloud Object Storage)
• Emits a task consumed by the LI pipeline
• Invoked at lines 68–78 in skill-handlers.ts

Archive Thresholds Configuration

The automatic archive is driven by two thresholds defined in skillCfg.extraction.chunkMaxBytes:

Threshold	Trigger Condition
Size threshold	Buffered size exceeds chunkMaxBytes (default varies by deployment)
Tool-call count	Number of tool-call messages ≥ 10

When either condition is met, the handler compresses the payload and invokes trigger.archive, which starts the LI pipeline.

Raw HTTP and Manual Trigger Examples

Using curl for Direct API Access

curl -X POST https://your-instance.tencentyun.com/v3/skill/conversation/add \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "session_id":"sess-12345",
        "user_id":"u001",
        "team_id":"teamA",
        "agent_id":"agentX",
        "messages":[
          {"role":"user","content":"Hello, how are you?"},
          {"role":"assistant","content":"I'm fine, thanks!"}
        ]
      }'

Force-Archive for Immediate LI Processing

When you need to trigger the pipeline before automatic thresholds are hit, use the force-archive endpoint implemented at lines 400–418 in skill-handlers.ts:

curl -X POST https://your-instance.tencentyun.com/v3/skill/conversation/force-archive \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "session_id":"sess-12345",
        "user_id":"u001",
        "team_id":"teamA",
        "agent_id":"agentX",
        "space_id":"teamA",
        "reason":"manual_trigger"
      }'

The handleConversationForceArchive function (lines 400–410) bypasses threshold checks and immediately archives the buffered conversation.

LI Pipeline Consumption and Processing

The LI pipeline runs as a separate worker process that:

1. Watches the COS archive bucket for new objects
2. Loads archived segments via MemoryProxy/src/injection/pipeline.ts
3. Executes Long-term-Inference processing
4. Writes results back to the memory store

This decoupled architecture ensures that ingestion remains low-latency while heavy inference runs asynchronously.

Key Source Files Reference

File	Purpose
MemoryCore/src/gateway/skill-handlers.ts	handleConversationAdd, handleConversationForceArchive implementations
MemoryCore/src/gateway/skill-schemas.ts	conversationAddRequestSchema definition
core/skill/conversation-add/wire.js	SkillConversationAddHandler – merge, compress, archive logic
sdk/memory-core/python/tencentdb_agent_memory/v3/skill_client.py	Python SDK wrapper
MemoryProxy/src/injection/pipeline.ts	LI pipeline consumer from COS

Summary

• Primary endpoint: POST /v3/skill/conversation/add ingests message turns and buffers them
• Automatic triggering: Archive and LI pipeline launch when size or tool-call thresholds are met
• Manual override: POST /v3/skill/conversation/force-archive triggers immediate archival
• Implementation core: skill-handlers.ts gateway, wire.js handler logic, pipeline.ts LI consumer

Frequently Asked Questions

What is the difference between conversation/add and force-archive?

The conversation/add endpoint buffers messages and only archives when configured thresholds are reached. The force-archive endpoint immediately writes the current buffer to COS and triggers the LI pipeline regardless of size or message count, as implemented in handleConversationForceArchive (lines 400–418).

How do I know if my messages triggered the LI pipeline?

The conversation/add response includes an archived boolean field. When true, the archive was written and the LI pipeline task was emitted. Check the response: {"status": "ok", "archived": true, ...}.

What happens if my message payload exceeds chunkMaxBytes?

The SkillConversationAddHandler in core/skill/conversation-add/wire.js applies compression and oversize handling. If the payload still exceeds limits after compression, the handler splits or truncates according to the skill configuration before archiving.

Can I customize the archive thresholds per session?

Thresholds are defined in skillCfg.extraction.chunkMaxBytes at the skill configuration level, not per-session. To apply different thresholds, create separate skill configurations or use force-archive for sessions requiring immediate processing.

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