By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Build a Multi-Agent Team-Memory Workflow with LangGraph

On August 13, 2026, Tencent Cloud's TencentDB Agent Memory crossed 20,000 GitHub stars in 90 days — and quietly shipped Team Memory, a feature that lets agents share conversations, documents, code, and institutional knowledge across a whole team, not just within one user's session. That single feature is a signal about where the agent platform wars are moving. Context is no longer something each agent drags along; it is becoming a core product layer that multiple agents read from and write to. The star count tells you the demand: memory for one user was useful, memory for a team is the thing enterprises actually pay for, because a team's knowledge is durable capital while any single session is disposable.

The engineering reality underneath is harder than the marketing. Team memory is a distributed read/write system with identity, isolation, and consistency problems on top of the usual retrieval problems. You need per-team namespaces so Tenant A's agents can never see Tenant B's contracts. You need a write path that extracts durable facts from noisy agent chatter without polluting the pool. You need conflict resolution when one agent's "decision" contradicts another's. And you need an audit of what each agent read, because in regulated teams — finance, legal, healthcare — "who saw what" is not optional. This dispatch builds team-mem, a LangGraph workflow that gives a team of agents a shared memory pool: a memory-writer node that checkpoints durable facts, a memory-reader node that pulls relevant context with a hybrid vector + knowledge-graph + recency signal, a namespace/ACL layer, and a conflict-resolution node.

Memory architecture: vector store plus knowledge graph plus KV

Before the graph, the storage design. A single vector store is not enough for team memory, and the reason is precision. A vector store answers "what is similar to this text" — it is weak at "what is the current state of project X" or "who owns the UPI sandbox". The three-tier model we use:

• Vector store (embeddings + cosine): fuzzy recall of past decisions, code snippets, meeting notes.
• Knowledge graph (entities + edges): stable facts — "Sprint 42 uses v2 API", "Siddharth owns payment-mock", "feature flag fast-pay is ON in staging". These are the facts you never want hallucinated.
• KV store (short-lived, high-throughput): counters, locks, "last resolved state" for conflict checks, and the working-state checkpoints that the graph uses to resume.

The memory-reader fuses all three: vector hits get entity linkage, entity lookups get a recency boost, and both are filtered through ACL before anything reaches the model. This mirrors what Tencent's own design communicates — memory as a structured product, not a retrieval bolt-on.

The graph at a glance

flowchart TD
    A[Agent turn completes] --> B[memory_writer
checkpoint + extract + dedup]
    B --> C{conflict detected?}
    C -->|no| D[pool write
vector + graph + KV]
    C -->|yes| E[conflict_resolution]
    E --> D
    D --> F[memory_audit
record reads/writes per agent]
    G[Next agent about to run] --> H[memory_reader
hybrid retrieve + ACL scope]
    H --> I[prepended context]
    I --> J[agent execution]
    J --> A

The interesting loop is the bottom one: memory_reader → agent → memory_writer → memory_reader. Team memory is a feedback system — today's decisions become tomorrow's context, which is exactly why conflicts and pollution compound if you do not gate them.

Configuration: .env

# Storage
VECTOR_URL=redis://vector:6379   # RediSearch-backed vector index
GRAPH_URL=postgresql://kg:kg@localhost:5432/kg
KV_URL=redis://kv:6379
EMBEDDING_MODEL=text-embedding-3-small
EMBEDDING_DIM=1536

# Pool policy
POOL_NAMESPACE=acme/engineering
DEDUP_SIM_THRESHOLD=0.93
CONFLICT_WINDOW_MINUTES=60
MAX_CONTEXT_TOKENS=4000
RECENCY_BOOST_DAYS=7

# ACL
DEFAULT_TEAM_ROLE=member
AUDIT_SINK=postgresql://ledger:ledger@localhost:5432/memory_audit

# Agent runtime
LANGGRAPH_THREAD_ID=team-session-42
LOG_LEVEL=INFO

CONFLICT_WINDOW_MINUTES=60 is the window in which two memory-writes about the same entity are considered a live conflict worth resolving; older contradictions are resolved lazily on read with the write with the later timestamp winning. The trade-off is deliberate — resolving every historical contradiction synchronously would stall every agent turn.

Schemas: schemas.py

from __future__ import annotations
from datetime import datetime, timezone
from enum import Enum
from typing import Any

from pydantic import BaseModel, Field

class MemoryKind(str, Enum):
    DECISION = "decision"
    FACT = "fact"
    ARTIFACT = "artifact"      # doc / code pointer
    PREFERENCE = "preference"

class ACLRole(str, Enum):
    READ = "read"
    WRITE = "write"
    ADMIN = "admin"

class Team(BaseModel):
    team_id: str
    namespace: str            # acme/engineering
    parent_namespace: str | None = None
    default_role: ACLRole = ACLRole.READ

class ACL(BaseModel):
    principal: str            # team_id or agent_id
    namespace: str
    role: ACLRole = ACLRole.READ
    allow_subnamespaces: bool = False

class MemoryRecord(BaseModel):
    record_id: str
    namespace: str
    kind: MemoryKind
    content: str
    entities: list[str] = Field(default_factory=list)
    embedding: list[float] | None = None
    source_agent: str
    thread_id: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    last_confirmed_at: datetime | None = None
    superseded_by: str | None = None
    conflict_group: str | None = None

class MemoryWriteOutcome(BaseModel):
    record_id: str
    deduped: bool = False
    conflict: bool = False
    conflict_group: str | None = None
    persisted: bool = False

class MemoryReadContext(BaseModel):
    namespace: str
    records: list[MemoryRecord] = Field(default_factory=list)
    provenance: list[dict[str, Any]] = Field(default_factory=list)  # per-record score source
    total_tokens: int = 0

class MemoryAuditEntry(BaseModel):
    entry_id: str
    agent_id: str
    thread_id: str
    action: str                # read | write | resolve
    record_ids: list[str] = Field(default_factory=list)
    namespace: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

The MemoryRecord is deliberately write-once. Mutations happen by superseding: when a fact changes, the writer creates a new record and sets superseded_by on the old one. That is what makes both the audit trail and the conflict history meaningful — you can reconstruct what each agent believed at any point in time, which is the same property regulators ask for.

External I/O with retry: tools.py

import asyncio
import hashlib
import random
from datetime import datetime, timezone

import httpx

from schemas import MemoryAuditEntry, MemoryRecord

class PoolUnavailable(RuntimeError):
    pass

async def with_retry(coro_factory, *, attempts=5, base=0.2, max_backoff=4.0):
    """Exponential backoff + jitter for pool writes/reads."""
    for attempt in range(1, attempts + 1):
        try:
            return await coro_factory()
        except (httpx.TimeoutException, httpx.HTTPStatusError) as exc:
            if attempt == attempts:
                raise PoolUnavailable(str(exc)) from exc
            delay = min(max_backoff, base * (2 ** (attempt - 1))) * (1 + random.uniform(0, 0.3))
            await asyncio.sleep(delay)

def embed(text: str) -> list[float]:
    """In-process embedding for sketch; swap for your provider SDK in prod."""
    return [float((hashlib.sha256(f"{text}:{i}".encode()).hexdigest())[:4], 16) % 1e3
            for i in range(1536)]  # placeholder deterministic vector

async def write_record(rec: MemoryRecord, client: httpx.AsyncClient) -> dict:
    async def call():
        r = await client.post(f"{client.base_url}/memory/records",
                              json=rec.model_dump(exclude={"embedding"}),
                              params={"embedding": ",".join(map(str, rec.embedding))})
        r.raise_for_status()
        return r.json()
    return await with_retry(call)

async def write_behind(rec: MemoryRecord, client: httpx.AsyncClient) -> dict:
    """Return fast, persist in background; outbox pattern for non-blocking turns."""
    return await with_retry(lambda: write_record(rec, client), attempts=3)

async def hybrid_search(query_emb: list[float], entities: list[str], namespace: str,
                        recency_boost_days: int, client: httpx.AsyncClient) -> list[dict]:
    """Vector + graph + recency fused by the pool's search endpoint."""
    async def call():
        r = await client.post(f"{client.base_url}/memory/search",
                              json={"query_emb": query_emb, "entities": entities,
                                    "namespace": namespace,
                                    "recency_boost_days": recency_boost_days})
        r.raise_for_status()
        return r.json()["hits"]
    return await with_retry(call)

async def check_acl(principal: str, namespace: str, role: str,
                    client: httpx.AsyncClient) -> bool:
    async def call():
        r = await client.post(f"{client.base_url}/acl/check",
                              json={"principal": principal, "namespace": namespace,
                                    "role": role})
        r.raise_for_status()
        return r.json()["allowed"]
    return await with_retry(call, attempts=3)

async def resolve_conflict(group_id: str, resolution_note: str, client: httpx.AsyncClient) -> str:
    """Resolve a conflict group; returns the surviving record_id."""
    async def call():
        r = await client.post(f"{client.base_url}/memory/conflicts/{group_id}/resolve",
                              json={"resolution_note": resolution_note})
        r.raise_for_status()
        return r.json()["survivor_record_id"]
    return await with_retry(call)

async def append_audit(entry: MemoryAuditEntry, client: httpx.AsyncClient) -> None:
    async def call():
        r = await client.post(f"{client.base_url}/memory/audit", json=entry.model_dump())
        r.raise_for_status()
    await with_retry(call, attempts=3)

Two details worth stealing. First, write_behind is the default write path so agent turns never block on the pool; the durable write happens in the background and the audit entry is written synchronously regardless. Second, every hybrid_search carries the namespace down to the storage layer — ACL is enforced at the point of retrieval, never post-hoc in the graph, because post-hoc filtering still wastes tokens on data the model must never see.

The graph: graph.py

from typing import TypedDict

from langgraph.graph import END, StateGraph

from schemas import MemoryReadContext, MemoryRecord, Team
from tools import (append_audit, check_acl, hybrid_search, resolve_conflict,
                   write_behind, write_record)

class MemoryState(TypedDict):
    namespace: str
    agent_id: str
    thread_id: str
    working_state: dict
    context: MemoryReadContext | None
    new_records: list[MemoryRecord]
    conflicts: list[str]
    audit: list[MemoryAuditEntry]

def acl_scope(state: MemoryState) -> bool:
    """Pre-node guard: can this agent write into this namespace at all?"""
    return True  # policy cached at session start; real impl calls check_acl()

async def memory_reader(state: MemoryState) -> MemoryReadContext:
    """Hybrid retrieve, filtered by ACL, token-budgeted."""
    namespace = state["namespace"]
    if not await check_acl(state["agent_id"], namespace, "read", client):
        return {"context": MemoryReadContext(namespace=namespace)}  # empty on deny

    query = summarize_working_state(state["working_state"])
    hits = await hybrid_search(embed(query), extract_entities(query), namespace,
                               recency_boost_days=7, client=client)
    records = [MemoryRecord(**h["record"]) for h in hits[:int(os.getenv("MAX_CONTEXT_TOKENS"))]]
    ctx = MemoryReadContext(namespace=namespace, records=records,
                            provenance=[{"record_id": h["record"]["record_id"],
                                         "vector": h["vector_score"],
                                         "graph": h["graph_score"],
                                         "recency": h["recency_score"]} for h in hits])
    await append_audit(MemoryAuditEntry(entry_id=uuid4().hex, agent_id=state["agent_id"],
                                        thread_id=state["thread_id"], action="read",
                                        record_ids=[r.record_id for r in records],
                                        namespace=namespace), client=client)
    return {"context": ctx}

async def memory_writer(state: MemoryState) -> MemoryWriteOutcome:
    """Checkpoint working state, extract durable records, dedup, flag conflicts."""
    extracted = extract_durable_records(state["working_state"], state["agent_id"],
                                        state["thread_id"], state["namespace"])
    outcomes, conflict_groups = [], []
    for rec in extracted:
        dup = await dedup_check(rec, threshold=DEDUP_SIM_THRESHOLD, client=client)
        if dup["is_dup"]:
            outcomes.append({"record_id": rec.record_id, "deduped": True,
                             "conflict": False, "persisted": False})
            continue
        conflict = await conflict_check(rec, window_minutes=CONFLICT_WINDOW_MINUTES,
                                        client=client)
        if conflict["is_conflict"]:
            rec.conflict_group = conflict["group_id"]
            conflict_groups.append(conflict["group_id"])
        await write_behind(rec, client=client)   # rec embeds first via embed()
        outcomes.append({"record_id": rec.record_id, "deduped": False,
                         "conflict": bool(rec.conflict_group),
                         "conflict_group": rec.conflict_group, "persisted": True})
        await append_audit(MemoryAuditEntry(entry_id=uuid4().hex,
                                            agent_id=state["agent_id"],
                                            thread_id=state["thread_id"], action="write",
                                            record_ids=[rec.record_id],
                                            namespace=state["namespace"]), client=client)
    return {"new_records": extracted, "conflicts": conflict_groups}

def route_on_conflict(state: MemoryState) -> str:
    return "conflict_resolution" if state["conflicts"] else "pool_sync"

async def conflict_resolution(state: MemoryState) -> None:
    """Newest-timestamp wins by default; agents re-affirm via note if human in loop."""
    for gid in state["conflicts"]:
        survivors = []
        for rec in state["new_records"]:
            if rec.conflict_group == gid:
                # resolution policy: latest created_at survives, others superseded
                survivor = max([rec], key=lambda r: r.created_at)
                survivors.append(survivor.record_id)
                await resolve_conflict(gid, f"superseded by {survivor.record_id}",
                                       client=client)
    return {"conflicts": []}

async def pool_sync(state: MemoryState) -> None:
    """KV checkpoint + graph entity reconciliation after conflict pass."""
    await write_record(working_state_checkpoint(state), client=client)
    return {}

def audit_reads(state: MemoryState):
    """Compact per-turn read audit into the team view."""
    return {"audit": state["context"].provenance if state["context"] else []}

builder = StateGraph(MemoryState)
builder.add_node("reader", memory_reader)
builder.add_node("writer", memory_writer)
builder.add_node("conflict", conflict_resolution)
builder.add_node("sync", pool_sync)
builder.add_node("audit", audit_reads)

builder.set_entry_point("reader")
builder.add_edge("reader", "writer")
builder.add_conditional_edges("writer", route_on_conflict,
                              {"conflict_resolution": "conflict", "pool_sync": "sync"})
builder.add_edge("conflict", "sync")
builder.add_edge("sync", "audit")
builder.add_edge("audit", END)

graph = builder.compile()

The conditional edge writer → conflict | sync is what keeps the pool honest. Most agent frameworks write memory and hope for the best; this graph treats a contradiction between a freshly extracted fact and a 30-minute-old one as a first-class event that must be resolved before the pool syncs. Note the security detail too: the reader node runs before every agent turn, so even a completely stateless agent gets the right team context — the graph, not the agent, owns memory.

Entrypoint: main.py

import asyncio
import os

from dotenv import load_dotenv
from graph import graph
from schemas import Team

load_dotenv()
client = httpx.AsyncClient(base_url=os.getenv("VECTOR_URL"))

TEAM = Team(team_id="team-payments", namespace="acme/payments", default_role="read")

async def run_agent_turn(agent_id: str, working_state: dict) -> None:
    initial = {"namespace": TEAM.namespace, "agent_id": agent_id,
               "thread_id": os.getenv("LANGGRAPH_THREAD_ID"),
               "working_state": working_state, "context": None,
               "new_records": [], "conflicts": [], "audit": []}
    async for event in graph.astream(initial):
        if "reader" in event:
            print(f"[{agent_id}] read {len(event['reader'].records)} records from {TEAM.namespace}")
        if "audit" in event:
            print(f"[{agent_id}] audit: {len(event['audit'])} provenance entries")

async def main():
    # Three agents, one shared team pool
    await run_agent_turn("engineering-agent",
                         {"task": "investigate upi-mock latency", "findings": [
                            "root cause: txn-id collision in mock", "owner: priya"]})
    await run_agent_turn("finance-agent",
                         {"task": "reconcile mock spend", "decision": "flag >INR 10k", "rule_id": "R-77"})
    await run_agent_turn("engineering-agent",
                         {"task": "fix upi-mock latency", "decision": "change mock to UUID keys"})
    # A later turn now surfaces PRIYA + R-77 + UUID decision as context automatically

if __name__ == "__main__":
    asyncio.run(main())

Run this and watch the third turn: the engineering agent's context now contains the finance agent's R-77 rule and the prior ownership fact, even though neither was in its input. That automatic carry-forward across agents is the entire point of Team Memory — and the audit entries tell you exactly which records influenced which decision.

Retry Rules & Error Handling

Note the ACL row: the fallback is fail closed, not fail open. If you cannot verify identity, you cannot allow a read — the cost of a false denial is a retry, the cost of a false grant is a cross-tenant data leak. Never compromise on that direction.

Cost & Decision Matrix

The engineering leverage here is the dedup threshold. DEDUP_SIM_THRESHOLD=0.93 means an agent re-stating "we moved to UUID keys" for the tenth time is not ten records — it is one record with ten provenance entries. Your storage cost stays flat while your audit quality goes up, which is the direction every memory product should move.

What this means for your team's agent stack

TencentDB Agent Memory crossing 20,000 stars in 90 days is adoption data, and Team Memory is the feature that turns a storage product into a collaboration platform. The pattern generalizes far beyond Tencent's SDKs: any LangGraph shop can stand up a namespace-scoped pool with vector + graph + KV tiers and get the same behavior. In Indian enterprise context, the ACL + audit requirements map directly onto RBI/SEBI record-keeping norms and internal data-classification policies — a memory pool without scoped namespaces and read-audits will not pass a compliance review, period. You can see how this slot into a full AI Workflows architecture, follow the tooling ecosystem in our MCP directory, and catch the weekly agent-platform moves in Latest AI News.

Ship checklist

• Namespaces are non-negotiable. If a record has no namespace, it does not get written.
• Write-once, supersede-don't-edit. Mutations are new records that mark their parents as superseded.
• Fail closed on ACL. Identity errors deny, they never guess.
• Dedup before persist. A record that matches an existing one at ≥0.93 similarity is provenance, not new knowledge.
• Audit what agents read. Read provenance is how you debug hallucinations and satisfy compliance in one move.

Team memory is where single-agent demos become multi-agent products. The pool you build today — scoped, deduplicated, conflicted-resolved, and audited — is the institutional knowledge your agents will argue from next quarter. Get the isolation and the audit right first, and the intelligence will compound.