v-agent-platform

State-Driven Multi-Agent Customer Service over Enterprise IM

A LangGraph customer-service agent for WeCom AiBot (WeChat Work bot), with hybrid-search + rerank RAG, hierarchical memory, and an offline evaluation harness.

Quick Start | Features | Architecture | Evaluation

Note

Single-tenant MVP. Multi-tenancy is deferred but not blocked.

What is v-agent-platform?

v-agent-platform is a reactive customer-service agent that answers customer inquiries over WeCom Aibot (WeChat Work intelligent bot) via a persistent WebSocket. Inbound messages are debounced, ordered through a sharded Redis Streams bus, then handled by a LangGraph agent that retrieves grounded answers from a Milvus knowledge base and remembers each customer across sessions.

Key highlights:

• LangGraph state machine: enter → compress → intent → agent → [tools]* → reflect → exit
• Hybrid-search + rerank RAG: one Milvus pass (dense cosine + BM25, WeightedRanker) into a candidate pool → cross-encoder rerank to top-k. Optional scalar metadata filter (category + price band) narrows the catalog before ranking. No cascade / confidence gate / LLM rewrite — the agent already wrote the query.
• Hierarchical memory: Redis working memory (hot) → Postgres event_memory (30-day vector recall) → user_profile (permanent)
• Token-efficient memory consolidation: extract mid-session at compression, promote at session end — no redundant LLM passes
• Strict per-shard serial ordering by (channel, channel_user_id)
• Built-in tools auto-described from docstrings: calculator, search, recall_memory, subagent
• Three-tier offline eval harness: retrieval (recall/MRR/nDCG), generation (RAGAS), system (latency + token cost)

Who is this for?

Anyone building or studying production-shaped Agentic systems over IM: the strict app/engine layering, the bus + worker concurrency model, the hierarchical-memory lifecycle, and the evaluation harness are all decoupled and readable. A good reference for LangGraph orchestration, hybrid-search + rerank RAG design, and how to evaluate a RAG pipeline end to end.

---

Contents

• 🗞️ Features
• 📽️ Architecture
• 📁 Project Structure
• 📖 Quick Start
  • Prerequisites
  • Install
  • Infrastructure
  • Schema
  • Configuration
  • Run
• 📊 Evaluation
• 🗝️ Tech Stack
• ⚠️ Security Notice
• 📝 License

---

🗞️ Features

Category	Details
Channel	WeCom AiBot persistent WebSocket; exponential-backoff reconnect + 30s heartbeat
Inbound bus	Redis Streams, sharded by (channel, channel_user_id); strict per-shard serial; DLQ on handler failure
Debounce	500ms idle window merges message bursts into one turn before the bus
Agent graph	LangGraph: enter → compress → intent classify → agent → tool loop → reflect → exit
Hybrid RAG	Single Milvus pass (dense + BM25 WeightedRanker) → candidate pool → cross-encoder rerank to top-k; optional category / price-band scalar filter; graceful degrade to hybrid order if the reranker is unreachable
Tools	calculator (safe AST), search (Milvus knowledge recall, with optional metadata filter), recall_memory (per-customer episodic), subagent (isolated single-shot delegate) + external MCP tools (OAuth-authed, cached)
Memory	Redis working memory (TTL) + Postgres event_memory (pgvector, 30-day) + user_profile (permanent JSONB)
Consolidation	Mid-session extraction at token threshold; session-end promotion (working memory → profile + events), or extract-from-history for short sessions
Reflection	Post-answer fact-check against tool results; bounded retry on hallucination
Observability	structlog structured logging + optional LangSmith tracing (env-gated)
Evaluation	retrieval (recall@k / MRR / nDCG by difficulty), conversational (multi-turn hit rate + task completion + RAGAS generation quality), system (latency p50/p95 + token cost)

---

📽️ Architecture

┌──────────────────────────────────────────────────────────────┐
│  /backend/app/  ── Shell & Gateway                            │
│  WS frame normalization, debounce, Redis Streams bus,          │
│  bus worker (glue), infra lifecycle (Postgres/Redis/Milvus)    │
│  ────────────────────────────────────────────────────────     │
│  /backend/v/    ── Agent Engine                               │
│  LangGraph orchestration, LLM reasoning, RAG retrieval,        │
│  hierarchical memory, tools, skills, model factory             │
└──────────────────────────────────────────────────────────────┘
            Unidirectional dependency:  app → v   (v never imports app)

Reactive flow:

Customer → WeCom AiBot WS → wecom_aibot_worker (debounce + normalize → SystemMessage)
  → Redis Streams (sharded, per-shard serial) → bus worker → LangGraph agent
  → tools (search / recall_memory / calculator / subagent)
  → reply via Redis pub/sub → WecomAibotClient → Customer

On 30-min silence a new session is minted, and the previous session's working memory is promoted to long-term storage inline (no external queue).

Detailed docs: doc/architecture.md · doc/data_flow.md · doc/call_chain.md · doc/gaps.md

---

📁 Project Structure

v-main/
├── backend/
│   ├── app/                          # Shell & Gateway (pure I/O, routing, infra)
│   │   ├── main.py                   # FastAPI entry: lifespan, bus consumer task, health routes
│   │   ├── wecom_aibot_worker.py     # Standalone WS worker process (inbound frames + pub/sub outbound)
│   │   ├── wecom_aibot/              # WeCom AiBot adapter
│   │   │   ├── client.py             # Persistent WS client (reconnect + heartbeat)
│   │   │   ├── debounce.py           # 500ms merge window (Redis hash + asyncio timer)
│   │   │   └── outbound.py           # Redis pub/sub → WS send bridge
│   │   ├── bus/                      # Redis Streams reactive bus
│   │   │   ├── shard.py              # mmh3 routing by (channel, channel_user_id)
│   │   │   ├── producer.py / consumer.py
│   │   │   └── worker.py             # make_bus_handler: glue graph + outbound + session-end consolidation
│   │   ├── gateway/                  # RequestIdMiddleware + /livez /readyz /health
│   │   └── store/                    # asyncpg pool + redis.asyncio factories
│   │
│   └── v/                            # Agent Engine (pure logic)
│       ├── agents/                   # state, nodes, edges, graph, checkpoints (Redis hot / PG cold)
│       │   ├── compression_node.py   # Mid-session memory extraction + history trim
│       │   └── intent_reflect.py     # Intent classifier + reflection (hallucination check)
│       ├── rag/retriever.py          # KnowledgeRetriever: Milvus hybrid search + cross-encoder rerank
│       ├── memory/                   # working (Redis) / event_memory (pgvector) / user_profile / extractor
│       ├── tools/                    # calculator, search, recall_memory, subagent
│       ├── models/                   # LLMCaller (timeout + cross-provider fallback) + factory
│       ├── skills/                   # markdown SOP loader + progressive-disclosure catalog
│       ├── hooks/                    # on_session_start, tool_guard
│       └── configs/                  # base.py (Pydantic settings) + agent_config.py (.agent/config.json loader)
│
│   ├── eval/                         # Offline evaluation harness
│   │   ├── retrieval/                # run_eval.py (hybrid+rerank recall/MRR/nDCG) + diagnose_misses.py + metrics.py
│   │   ├── conversational/           # run_conv_eval.py (multi-turn hit + task + RAGAS) + ragas_score.py
│   │   ├── generate/                 # run_generate_eval.py (single-turn RAGAS 4 metrics)
│   │   ├── system/                   # run_system_eval.py (latency + token cost)
│   │   ├── gen/                      # LLM data generators (catalog / QA / difficulty scorer)
│   │   ├── seed_milvus.py            # Embed corpus → Milvus collection
│   │   └── data/                     # products.jsonl / faq.jsonl / qa.jsonl / conv_cases.jsonl / reports
│   │
│   └── test/                         # pytest unit + e2e suites (586 unit tests, ≥80% coverage gate)
│
├── .agent/config.json                # MCP servers + skill sources (enable-gated, ${VAR} secrets)
├── docker/docker-compose.yml         # postgres(pgvector) + redis + etcd + minio + milvus 2.5 + app
├── scripts/                          # sql migrations, seed_knowledge_chunks, start_reranker (sidecar), mock_mcp_server
├── doc/                              # architecture / data_flow / call_chain / gaps
├── CLAUDE.md                         # Repo-wide engineering rules (+ per-layer CLAUDE.md)
└── README.md

---

📖 Quick Start

Prerequisites

• Python ≥ 3.11
• uv — package manager / runner
• Docker + Docker Compose (local Postgres + Redis + Milvus)
• psql CLI (for schema migration)
• An OpenAI-compatible LLM + embedding endpoint (default: DashScope compatible-mode — DeepSeek chat + Qwen text-embedding-v4)

Install

uv sync

Creates .venv/ and installs from uv.lock. Run everything with uv run ....

Infrastructure

docker compose -f docker/docker-compose.yml up -d postgres redis etcd minio milvus
docker compose -f docker/docker-compose.yml ps   # wait until healthy

The app service is profiled (profiles: ["app"]) and is not started by default.

Schema

bash scripts/db_apply.sh

Runs scripts/sql/ in order: pgvector extension → agent schema (working memory / checkpointer / user_profile / event_memory / knowledge_chunk) → dw business warehouse → meta NL2SQL metadata. Reads POSTGRES_DSN, defaulting to postgresql://postgres:postgres@localhost:5432/agent.

Configuration

Three layers, by design:

File	Holds	Precedence
config.yaml	non-secret tunables — temperature, timeouts, thresholds, TTLs, shard counts, RAG weights	lowest (above code defaults)
.env	secrets + endpoints — API keys, base URLs, model names, DSNs, bot credentials	overrides YAML
.agent/config.json	extensions — MCP servers + skill sources, each enable-gated, secrets via ${VAR}	committed; resolves ${VAR} from env

Final precedence (highest wins): env var > .env > config.yaml > code default. Edit config.yaml to change behaviour without touching code; keep secrets out of it.

cp config.example.yaml config.yaml    # tunables — edit freely
cp .env.example .env                  # secrets + endpoints — fill in
# .agent/config.json ships with all extensions disabled — flip enable:true to use

The YAML is sectioned by reflection: each top-level key maps to an AppSettings field (llm, rag, memory, bus, …) — no hand-written wiring, so a new settings class needs no extra config plumbing. Point at a different file with APP_CONFIG_FILE=/path/to/config.yaml.

.env (secrets + endpoints) minimum required:

Group	Keys
Chat LLM	LLM_BASE_URL / LLM_API_KEY / LLM_MODEL (+ _FALLBACK for the cross-provider fallback endpoint)
Embedding	EMBEDDING_BASE_URL / EMBEDDING_API_KEY / EMBEDDING_MODEL / EMBEDDING_DIM (independent endpoint)
Reranker	RAG_RERANK_URL / RAG_RERANK_MODEL (+ RAG_RERANK_API_KEY for a hosted reranker)
Postgres	POSTGRES_DSN
Redis	REDIS_URL
Milvus	MILVUS_URI / MILVUS_TOKEN (token cloud-only)
WeCom AiBot	WECOM_AIBOT_WS_URL / WECOM_AIBOT_BOT_ID / WECOM_AIBOT_SECRET
Eval judge (optional)	EVAL_JUDGE_BASE_URL / EVAL_JUDGE_API_KEY / EVAL_JUDGE_MODEL (RAGAS judge, decoupled from the agent)
LangSmith (optional)	LANGSMITH_TRACING / LANGSMITH_API_KEY

config.yaml tunables (see config.example.yaml): llm.temperature/timeout_seconds/thinking, memory.* TTLs + compression thresholds, bus.shard_count/debounce_ms, rag.min_k/dense_weight/bm25_weight/rerank_enabled/rerank_candidates, milvus.collection_name, intent.* thresholds, mcp.* cache/timeout.

.agent/config.json declares MCP servers + skill sources. Each entry has an enable flag (only true loads) and may reference secrets as ${VAR} (resolved from the environment at load time, so the file is committable). Example: a shop MCP server with "oauth_client_secret": "${SHOP_OAUTH_SECRET}", or a skill source {"enable": true, "path": "/srv/skills/internal"}.

Leaving WECOM_AIBOT_WS_URL empty disables the WS worker (it exits immediately on start).

Run

Two processes:

1. FastAPI main process (HTTP gateway + in-process bus worker):

uv run fastapi dev backend/app/main.py      # listens on :8000

2. WeCom AiBot WS worker (holds the persistent WebSocket):

uv run python -m backend.app.wecom_aibot_worker

Inbound aibot_msg_callback frames → normalized to SystemMessage → bus → same downstream as any channel. Outbound replies are published to Redis pub/sub wecom_aibot:outbound and forwarded over the WS by this worker.

Before first use, seed the knowledge base:

uv run python -m backend.eval.seed_milvus

Verify

uv run pytest --cov=backend --cov-fail-under=80   # 586 unit tests, ≥80% gate
uv run ruff check . --fix && uv run ruff format .

Health probes on the FastAPI process: GET /livez (always 200), GET /readyz (503 if Postgres/Redis down), GET /health (legacy).

---

📊 Evaluation

A multi-tier offline harness over a 150-doc corpus (120 products + 30 FAQ), 337 difficulty-balanced QA pairs, and 30 multi-turn conversational cases. See backend/eval/README.md for full results and parameter tuning.

# Retrieval — recall@k / MRR / nDCG, by difficulty (hybrid + rerank)
uv run python -m backend.eval.retrieval.run_eval

# Conversational — multi-turn hit rate + task completion + RAGAS generation quality
#   (start the reranker sidecar first; RAGAS judge via EVAL_JUDGE_* in .env)
uv run python -m scripts.start_reranker            # bge-reranker-v2-m3 on :8767
uv run python -m backend.eval.conversational.run_conv_eval --out report.json

# Generation (single-turn) — RAGAS: Faithfulness / ContextRecall / AnswerRelevancy / AnswerCorrectness
uv run python -m backend.eval.generate.run_generate_eval --gen-model qwen3-8b --no-think

# System — E2E latency (p50/p95/p99) + token consumption + cost estimate
uv run python -m backend.eval.system.run_system_eval

Headline retrieval result (337 queries, top_k=5): recall@5 ≈ 0.98 on the hybrid pass. The scalar metadata filter (category + price band) recovers the conversational misses where pure semantic rerank down-ranks off-type catalog items.

---

🗝️ Tech Stack

1. Agent Orchestration (click me)

LangGraph state machine in backend/v/agents/graph.py:

enter → compress → intent → agent → route_after_agent ─┬→ tools → agent → ...
                                                       └→ reflect ─┬→ agent (retry)
                                                                   └→ exit → END

• enter: stacks the per-turn system prompt — base persona, <customer_profile>, <recent_events>, session working memory, matched SOPs.
• compress: when messages exceed the token threshold, an LLM extracts working_memories (→ Redis) + event_memories (→ Postgres), then trims history behind a <compressed_history> marker.
• intent: flash-tier LLM classifies the turn (refund / logistics / complaint / general).
• agent: main-tier LLM with bound tools; loops through ToolNode until no more tool calls.
• reflect: fact-checks the reply against tool results; bounded retry on detected hallucination.

LLMCaller (backend/v/models/llm_caller.py) is the single entry point for all LLM calls: per-call timeout, cross-provider fallback (independent primary / fallback endpoints), structured output via Pydantic, and RunnableConfig propagation for nested LangSmith traces.

2. Hybrid RAG + Rerank (click me)

KnowledgeRetriever (backend/v/rag/retriever.py) runs one hybrid pass against a single Milvus collection (knowledge_chunks), then a cross-encoder rerank:

query (already written by the agent LLM)
  ▼ hybrid search: dense cosine + BM25, fused by WeightedRanker → candidate pool (default 20)
  ▼ optional scalar filter: metadata["category"] / price band (set when the customer named them)
  ▼ cross-encoder rerank: re-scores (query, passage) pairs → top_k
        LocalReranker (self-hosted sidecar, bge-reranker-v2-m3 @ :8767) / RemoteReranker (hosted API)
        reranker unreachable → degrade to hybrid order (never crashes)

• No cascade / confidence gate / LLM rewrite. This is agentic RAG — the query is already written by the agent LLM from conversation context, so a second retriever-side rewrite was redundant (and the old cascade's 25–49s latency source). Rerank does the quality lifting instead.
• Scalar metadata filter: search(query, category=…, price_min=…, price_max=…) narrows the catalog to the matching sub-set before ranking via a Milvus boolean expr over the product metadata JSON. Unset → unfiltered (prior behaviour). Aligns retrieval with how concrete "category + budget" asks are scored.
• Collection schema: source_type, source_id, text (BM25-analyzed), embedding (FLOAT_VECTOR 1024), sparse_embedding (auto from BM25 Function), metadata (JSON: category, brand, price).
• Indexes: HNSW (M=16, efConstruction=64, COSINE) + SPARSE_INVERTED_INDEX (BM25).

The search tool is a thin @tool wrapper over this retriever; the seeder reuses the exact same collection setup for schema parity.

3. Hierarchical Memory (click me)

Three temperature tiers, all sentence-shaped via MemoryEntry:

• Working memory (Redis, TTL = session window): per-session list, injected into the prompt each turn.
• Event memory (Postgres agent.event_memory, pgvector, 30-day TTL): cross-session episodic recall via the recall_memory tool (recency-weighted cosine).
• User profile (Postgres agent.user_profile, JSONB, permanent): one row per (channel, channel_user_id), fully injected at on_session_start.

Consolidation is token-efficient by design:

• Mid-session (token threshold) — compression_node extracts working + event memories from the dropped history. No profile write (session still active).
• Session-end (next message after silence) — _promote_prev_session fires fire-and-forget: if working memory is non-empty it promotes directly (no extra LLM call); if empty (short session) it runs extract_from_messages() over the checkpoint history, then promotes.

4. Bus & Concurrency (click me)

• RedisStreamShard: mmh3.hash(f"{channel}:{channel_user_id}") % shard_count (default 64 shards) → strict per-shard serial; different shards run concurrently.
• Debouncer: merges sub-500ms bursts from the same identity into one turn before the bus (Redis hash + asyncio timer).
• BusConsumer: one asyncio task per shard, XREADGROUP with DLQ on handler failure.
• Outbound replies do not re-enter the bus — they go straight back through the channel adapter (the bus is reactive-inbound only).
• LangGraph checkpointer: Redis on the hot path (RedisCheckpointer, TTL), with a langgraph-official AsyncPostgresSaver available for cold storage.

5. Models & Infra (click me)

• LLM / Embedding: OpenAI-compatible via langchain-openai ChatOpenAI(base_url=...). Chat has independent primary / fallback endpoints (cross-provider failover); embedding is a fully decoupled endpoint (commonly a different provider). check_embedding_ctx_length=False so DashScope-style endpoints receive raw strings, not token ids. All URLs / keys / model names live in .env; temperature / thinking are config.yaml tunables.
• Postgres: single instance, multiple schemas — agent (memory/sessions/checkpoints/pgvector), dw (business warehouse), meta (NL2SQL metadata). asyncpg only; psycopg2 forbidden.
• Milvus 2.5: vector + BM25 hybrid; deployed via docker-compose with etcd + minio.
• Redis: bus, working memory, checkpointer hot path, pub/sub outbound.
• Config: pydantic-settings, one class per env prefix, composed by a cached get_settings().

6. Evaluation Harness (click me)

backend/eval/ is split into sub-modules plus data generators:

• retrieval/ — run_eval.py (end-to-end hybrid+rerank: recall/MRR/nDCG/hit by difficulty) and diagnose_misses.py (isolates a miss into query vs recall vs rerank vs gold-labelling, and A/Bs the metadata filter).
• conversational/ — run_conv_eval.py drives multi-turn dialogues (a user-simulator persona vs the full agent graph) and scores retrieval hit rate, task completion (LLM judge), tool success, plus RAGAS generation quality (ragas_score.py). The RAGAS judge is decoupled from the agent-under-test via EVAL_JUDGE_*; references are synthesised mechanically from the code-enumerated gold.
• generate/ — run_generate_eval.py scores single-turn answers with RAGAS 0.4 (Faithfulness / ContextRecall / AnswerRelevancy / AnswerCorrectness); --gen-model overrides only the generation model (e.g. qwen3-8b --no-think).
• system/ — run_system_eval.py runs the full graph per query, capturing latency p50/p95/p99 and token counts (via a LangChain callback), with a configurable price table for cost estimation.
• gen/ — LLM generators that build the catalog, QA pairs, and difficulty scores; seed_milvus.py embeds the corpus into Milvus with the production schema.

7. MCP Integration (click me)

backend/v/mcp/ connects the agent to external Model Context Protocol servers (stdio / streamable-HTTP / SSE) and exposes their tools alongside the built-ins.

• Auth: OAuth 2.0 client_credentials (machine-to-machine) via oauth.py — the backend exchanges client_id/client_secret for a bearer token, cached in-process and auto-refreshed ~30s before expiry. Also supports api_key (static Bearer) and none. client.py resolves the bearer at connect time.
• Tool naming: the adapter prefixes each tool with its server id ({server}_{tool}) so two servers can expose the same tool name without colliding.
• Caching (cache.py): read-class tool results go through L1 (in-memory, 5-min) + L2 (Redis, 24-h), keyed by sha256(canonical args). Write-class tools (declared per-server in write_tools) skip the cache.
• Lifecycle (registry.py): MCPRegistry.connect_all() runs at FastAPI startup, discovers each server's tools, and build_graph(..., extra_tools=mcp_tools) binds them to the LLM + the guarded ToolNode. Closed on shutdown. The server list comes from .agent/config.json (loaded by configs/agent_config.py, each entry enable-gated with ${VAR} secrets) — no enabled server → disabled, zero overhead.
• Mock server (scripts/mock_mcp_server.py): a standalone FastMCP HTTP server with a /token endpoint + three bearer-gated tools — query_order, query_logistics, create_handoff_ticket (hardcoded fake data). Exercises the full OAuth → JSON-RPC path locally; swap for a real server by changing only the url/credentials.

# 1. start the mock server (listens on :9100)
uv run python -m scripts.mock_mcp_server

# 2. enable a server in .agent/config.json (set enable:true, fill ${VAR} secrets in .env), then start the app

---

⚠️ Security Notice

This platform is an MVP / learning project intended for trusted local or internal-network use. It does not ship production-grade hardening:

• No gateway authentication. Internal admin/health routes are open to anyone who can reach the process.
• Secrets in plaintext .env. LLM_API_KEY, EMBEDDING_API_KEY, POSTGRES_DSN, WeCom bot secrets are read from .env via pydantic-settings (and ${VAR} references in .agent/config.json resolve from the same environment). Never commit .env; rotate keys manually.
• Default credentials in .env.example / docker-compose.yml (e.g. postgres:postgres) must be changed before any non-local deployment.
• Plain transport / open CORS by default. Terminate TLS and restrict origins behind a reverse proxy if exposed beyond localhost.
• Milvus / Postgres / Redis are exposed without network-level access control by default — use firewall rules or Docker network isolation.

Caution

Do not expose this directly to the public internet without adding authentication, TLS, and access controls.

---

📝 License

Open source under the MIT License.