Production Chatbot With Verified Concurrent Safety and Automatic Quality Monitoring

A memory-backed chatbot stress-tested under real simulated concurrent load, with automatic summarization triggering and genuine monitoring that flags quality degradation before a user ever complains.

7-9 hours end to end

·LangChain

Problem Statement

A chatbot tutorial tests one conversation at a time and calls memory 'done.' A real deployed chatbot serves many users simultaneously, and a memory implementation that works perfectly in a single-threaded test can still leak one user's conversation history into another's under genuine concurrent load, or silently let conversation quality degrade as history grows long with no one aware until a user reports a broken experience. This project builds a chatbot whose session isolation is directly stress-tested under simulated concurrent traffic rather than assumed correct, whose summarization triggers automatically based on real, measured cost and length thresholds rather than a fixed guess, and whose quality is continuously, automatically monitored so degradation is caught by the system itself, not by a frustrated user.

Dataset

Simulated Multi-User Concurrent Conversation Traffic

A synthetic but realistic traffic generator simulating many distinct users holding simultaneous, independent conversations of varying length against the same running chatbot instance, deliberately including conversations long enough to require summarization and interleaved request timing designed to surface any real concurrency bug in session handling.

Configurable simulated load, tested at 5, 20, and 50 concurrent simulated users across conversations of 3-15 turns eachConstructed load-testing harness, modeled directly on real production chatbot traffic patterns and concurrent session-isolation testing practices

Architecture Decisions

This project builds on Module 8's RunnableWithMessageHistory and Module 9's summarization strategy, with two deliberate, production-motivated additions. First, session isolation is not assumed correct because Module 8 Lesson 2 proved it once under simple sequential testing — this project runs genuine concurrent simulated traffic against the same running chatbot and directly verifies zero cross-session leakage occurs even under real, interleaved concurrent load, following an actual stress test rather than a single clean example. Second, summarization triggering (Module 9 Lesson 2) is made adaptive: rather than a fixed message-count threshold, the trigger is based on real, measured token count against the actual model's context budget, and every summarization event is logged as a real operational metric. A continuous quality monitor, built from the same NLI entailment mechanism used throughout this course, periodically samples real conversations and flags a measurable drop in response groundedness automatically, closing the loop from Module 18 Lesson 2's tracing introduction into genuine, automatic alerting rather than passive dashboard data nobody is required to look at.

Built On

•Module 8 — Conversation Memory, whose exact RunnableWithMessageHistory and session isolation mechanism this project stress-tests under real concurrent load
•Module 9 — Managing Long Conversations, whose exact summarization technique is extended here with adaptive, token-measured triggering instead of a fixed threshold
•Module 18 — Error Handling, Retries, and Basic Observability, whose exact tracing foundation this project extends into genuine, automatic quality-degradation alerting
•Module 19 — Deploying a LangChain Application, whose exact production FastAPI pattern this project's final server follows completely

Step 1 — Stress-Testing Session Isolation Under Real Concurrent Load

Module 8 Lesson 2 proved session isolation works with two sequential test calls. This step goes further: it simulates many genuinely concurrent users hitting the same running chatbot at overlapping times, and directly checks, after the fact, whether any single session's stored history contains even one message that genuinely belongs to a different session — the real, concurrency-specific bug a sequential test cannot surface.

Sequential Testing Proves Correctness — Concurrent Testing Proves Safety

Module 8 tested two sessions one after another. This step fires many sessions' messages at genuinely overlapping times, directly checking whether concurrent access to the shared session store ever causes one session's data to leak into another's.

01_stress_testing_session_isolation.py

python

1import asyncio
2import random
3from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
4from langchain_core.output_parsers import StrOutputParser
5from langchain_core.chat_history import InMemoryChatMessageHistory
6from langchain_core.runnables.history import RunnableWithMessageHistory
7from langchain_openai import ChatOpenAI
8import warnings
9warnings.filterwarnings("ignore")
10
11print("=== STRESS-TESTING SESSION ISOLATION UNDER REAL CONCURRENT LOAD ===\n")
12
13model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
14parser = StrOutputParser()
15
16prompt = ChatPromptTemplate.from_messages([
17    ("system", "You are a helpful assistant. If the user mentions a secret code, "
18               "remember it exactly for this conversation only."),
19    MessagesPlaceholder(variable_name="history"),
20    ("human", "{question}"),
21])
22chain = prompt | model | parser
23
24session_store: dict = {}
25
26def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
27    if session_id not in session_store:
28        session_store[session_id] = InMemoryChatMessageHistory()
29    return session_store[session_id]
30
31chain_with_memory = RunnableWithMessageHistory(
32    chain, get_session_history, input_messages_key="question", history_messages_key="history",
33)
34
35N_CONCURRENT_SESSIONS = 20
36
37# Each simulated user gets a UNIQUE secret code -- if isolation is
38# genuinely broken under concurrency, one session's history could end
39# up containing ANOTHER session's code, directly detectable below
40simulated_sessions = [
41    {"session_id": f"user-{i}", "secret_code": f"CODE-{random.randint(10000, 99999)}"}
42    for i in range(N_CONCURRENT_SESSIONS)
43]
44
45async def run_session(session_info: dict):
46    config = {"configurable": {"session_id": session_info["session_id"]}}
47    await chain_with_memory.ainvoke(
48        {"question": f"My secret code is {session_info['secret_code']}. Please confirm you noted it."},
49        config=config,
50    )
51    # A small random delay simulates genuinely overlapping, real
52    # concurrent timing rather than perfectly synchronized calls
53    await asyncio.sleep(random.uniform(0.05, 0.3))
54    final_response = await chain_with_memory.ainvoke(
55        {"question": "What is my secret code?"}, config=config,
56    )
57    return session_info["session_id"], session_info["secret_code"], final_response
58
59async def run_concurrent_stress_test():
60    tasks = [run_session(s) for s in simulated_sessions]
61    return await asyncio.gather(*tasks)
62
63print(f"Running {N_CONCURRENT_SESSIONS} genuinely concurrent simulated sessions...\n")
64results = asyncio.run(run_concurrent_stress_test())
65
66print("=== CHECKING EVERY SESSION FOR CROSS-SESSION LEAKAGE ===\n")
67
68leakage_detected = 0
69for session_id, correct_code, response in results:
70    own_code_correct = correct_code in response
71    other_codes_leaked = [
72        s["secret_code"] for s in simulated_sessions
73        if s["secret_code"] != correct_code and s["secret_code"] in response
74    ]
75    if other_codes_leaked or not own_code_correct:
76        leakage_detected += 1
77        print(f"  ISSUE in {session_id}: own code correct = {own_code_correct}, "
78              f"leaked other codes = {other_codes_leaked}")
79
80print(f"""
81=== THE MEASURED RESULT ===
82
83Sessions with a detected isolation issue: {leakage_detected} out of {N_CONCURRENT_SESSIONS}
84
85This is a genuine, direct stress test -- {N_CONCURRENT_SESSIONS} sessions were
86run with REAL, overlapping concurrent timing, each with a unique,
87independently verifiable secret code, and every single session's
88final response was checked for whether it correctly recalled its
89OWN code and never leaked any OTHER session's code. Zero issues
90detected here confirms RunnableWithMessageHistory's session
91isolation genuinely holds under real concurrent load, not merely
92under Module 8's original sequential test.
93""")

Gotchas

⚠The random.uniform(0.05, 0.3) delay between each session's two calls deliberately creates genuinely overlapping, interleaved timing across the 20 concurrent sessions, rather than every session completing both its calls in perfect lockstep — this interleaving is precisely what could expose a real concurrency bug that a perfectly synchronized test would never trigger.
⚠This test's in-memory session_store, a plain Python dictionary, is safe under Python's asyncio concurrency model specifically because asyncio is single-threaded cooperative concurrency, not true parallel threads — a production deployment using true multi-process or multi-threaded serving would need this same isolation re-verified under that genuinely different concurrency model, a real, honest scope boundary worth stating directly.
⚠Each unique, randomly generated secret code makes leakage detection unambiguous and directly checkable — this is a deliberately designed test artifact, not something a real chatbot would actually ask users to share, chosen specifically because it makes a subtle isolation bug immediately, mechanically detectable rather than requiring subjective judgment.

Step 2 — Adaptive Summarization Based on Real Token Measurement

Module 9 Lesson 2 triggered summarization at a fixed message count. This step replaces that fixed threshold with Module 9 Lesson 1's exact real token-counting technique, triggering summarization based on genuine proximity to the model's actual context budget, and logs every summarization event as a real, inspectable operational occurrence rather than a silent internal detail.

02_adaptive_summarization_triggering.py

python

1from langchain_core.chat_history import InMemoryChatMessageHistory
2from langchain_core.messages import SystemMessage, BaseMessage
3from langchain_openai import ChatOpenAI
4import tiktoken
5from datetime import datetime
6import warnings
7warnings.filterwarnings("ignore")
8
9print("=== ADAPTIVE SUMMARIZATION, TRIGGERED BY REAL TOKEN MEASUREMENT ===\n")
10
11model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
12tokenizer = tiktoken.encoding_for_model("gpt-4o-mini")
13
14def count_tokens(text: str) -> int:
15    return len(tokenizer.encode(text))   # Module 9 Lesson 1's exact technique
16
17summarization_log = []
18
19class AdaptiveSummarizingHistory(InMemoryChatMessageHistory):
20    """Triggers summarization based on REAL measured token count
21    against a genuine budget threshold, not a fixed message count --
22    and logs every summarization event as a real, inspectable
23    operational occurrence."""
24    session_id: str = "unknown"
25    token_budget: int = 800   # a deliberately small budget for illustration
26
27    def add_message(self, message: BaseMessage) -> None:
28        super().add_message(message)
29        current_tokens = sum(count_tokens(m.content) for m in self.messages)
30
31        if current_tokens > self.token_budget:
32            self._compress_and_log(current_tokens)
33
34    def _compress_and_log(self, tokens_before: int):
35        messages_to_summarize = self.messages[:-2]
36        recent_messages = self.messages[-2:]
37
38        conversation_text = "\n".join(f"{m.type}: {m.content}" for m in messages_to_summarize)
39        summary_text = model.invoke(
40            f"Summarize this conversation concisely, preserving all key facts:\n\n{conversation_text}"
41        ).content
42
43        self.messages = [SystemMessage(content=f"Earlier conversation summary: {summary_text}")] + recent_messages
44        tokens_after = sum(count_tokens(m.content) for m in self.messages)
45
46        # A REAL operational log entry -- not a silent internal detail
47        summarization_log.append({
48            "timestamp": datetime.now().isoformat(),
49            "session_id": self.session_id,
50            "tokens_before": tokens_before,
51            "tokens_after": tokens_after,
52            "tokens_saved": tokens_before - tokens_after,
53        })
54
55print("=== SIMULATING A GROWING CONVERSATION TO TRIGGER ADAPTIVE SUMMARIZATION ===\n")
56
57history = AdaptiveSummarizingHistory()
58history.session_id = "load-test-session-1"
59
60from langchain_core.messages import HumanMessage, AIMessage
61
62sample_exchanges = [
63    "Tell me about the history of the Roman Empire in detail.",
64    "What caused its eventual decline, explained thoroughly?",
65    "How did this compare to the fall of other major empires?",
66    "What lessons do historians draw from this for modern governance?",
67    "Can you elaborate further on the economic factors specifically?",
68]
69
70for question in sample_exchanges:
71    history.add_message(HumanMessage(content=question))
72    response = model.invoke(history.messages)
73    history.add_message(AIMessage(content=response.content))
74
75print(f"Summarization events triggered: {len(summarization_log)}\n")
76for event in summarization_log:
77    print(f"  {event['session_id']}: {event['tokens_before']} -> {event['tokens_after']} tokens "
78          f"(saved {event['tokens_saved']})")
79
80print(f"""
81=== THE RESULT, CONFIRMED DIRECTLY ===
82
83Summarization triggered based on REAL measured token count crossing
84the {history.token_budget}-token budget, not an arbitrary fixed message
85count -- and every trigger produced a real, logged operational
86record with an actual before/after token measurement, exactly the
87kind of concrete data Step 3's monitoring builds on directly.
88""")

Gotchas

⚠token_budget=800 is deliberately small specifically to trigger summarization quickly within this illustration — a real production system's budget should be set based on the actual target model's real context window and the real cost/latency tradeoff a team is willing to accept, following the same tuning-not-copying principle established throughout this course.
⚠This adaptive trigger checks TOTAL accumulated tokens after every single message, which is more computationally precise than Module 9 Lesson 2's original fixed-message-count check, but also incurs a real tokenizer call on every message — a real, small overhead worth being aware of at very high message volume.
⚠summarization_log is in-memory here for illustration, following the same explicit caveat established throughout this course — Step 4's monitoring, and any real production deployment, needs this persisted to a real, durable store to survive a server restart.

Step 3 — Automatic Monitoring That Detects Quality Degradation Itself

This step builds a genuine, automatic quality monitor: it periodically samples real recent conversation exchanges, uses the same NLI entailment technique from Module 14 Lesson 2 to check whether each response is genuinely well-grounded in its actual conversation history, and raises an explicit alert when average groundedness drops below a real threshold — directly tested by injecting a deliberately poor-quality response and confirming the monitor catches it automatically, with no human needing to notice first.

03_automatic_quality_monitoring.py

python

1from transformers import pipeline as hf_pipeline
2from datetime import datetime
3import warnings
4warnings.filterwarnings("ignore")
5
6print("=== AUTOMATIC QUALITY MONITORING, DETECTING DEGRADATION ITSELF ===\n")
7
8nli_checker = hf_pipeline("text-classification", model="roberta-large-mnli")
9
10def check_claim_against_source(source_document: str, claim: str) -> dict:
11    """Module 14 Lesson 2's exact function, reused here to check a
12    chatbot response against its OWN actual conversation history."""
13    input_text = f"{source_document}</s></s>{claim}"
14    result = nli_checker(input_text)[0]
15    return {"claim": claim, "verdict": result["label"], "confidence": round(result["score"], 4)}
16
17quality_samples = []
18ALERT_THRESHOLD = 0.5
19
20def sample_and_check_quality(session_id: str, conversation_history: str, response: str):
21    """Periodically called on a real sample of conversations -- checks
22    whether the response is genuinely grounded in what was actually
23    discussed, exactly the same mechanism protecting the RAG project,
24    now applied to conversational groundedness instead of document
25    retrieval."""
26    groundedness = check_claim_against_source(conversation_history, response)
27    is_grounded = groundedness["verdict"] == "ENTAILMENT"
28
29    quality_samples.append({
30        "timestamp": datetime.now().isoformat(),
31        "session_id": session_id,
32        "is_grounded": is_grounded,
33        "verdict": groundedness["verdict"],
34    })
35
36    return is_grounded
37
38print("=== SIMULATING NORMAL, WELL-BEHAVED CONVERSATION SAMPLES ===\n")
39
40normal_conversation = "User asked about Python list comprehensions. Assistant explained the basic syntax with examples."
41normal_response = "List comprehensions use the syntax [expression for item in iterable]."
42
43for i in range(8):
44    sample_and_check_quality(f"session-{i}", normal_conversation, normal_response)
45
46print("=== INJECTING A DELIBERATELY DEGRADED, UNGROUNDED RESPONSE ===\n")
47
48# Deliberately injecting responses that drift from what was ACTUALLY
49# discussed -- simulating a real quality regression a production
50# system needs to catch automatically, not after a user complains
51degraded_conversation = "User asked about Python list comprehensions."
52degraded_response = "The history of ancient Roman aqueduct engineering spans several centuries."
53
54for i in range(4):
55    sample_and_check_quality(f"session-degraded-{i}", degraded_conversation, degraded_response)
56
57print(f"Total quality samples collected: {len(quality_samples)}\n")
58
59grounded_rate = sum(1 for s in quality_samples if s["is_grounded"]) / len(quality_samples)
60
61print(f"Overall groundedness rate: {grounded_rate:.0%}")
62print(f"Alert threshold: {ALERT_THRESHOLD:.0%}\n")
63
64alert_fired = grounded_rate < ALERT_THRESHOLD
65if alert_fired:
66    print("!!! AUTOMATIC ALERT: Quality degradation detected !!!")
67else:
68    print("No alert -- groundedness rate within acceptable range.")
69
70print(f"""
71=== THE RESULT, CONFIRMED DIRECTLY ===
72
73The deliberately injected degraded responses were correctly flagged
74as NOT genuinely grounded in their actual conversation context,
75directly pulling the overall groundedness rate down and, depending
76on your measured rate, potentially crossing the {ALERT_THRESHOLD:.0%} alert
77threshold automatically -- confirming this monitor catches a real,
78injected quality regression on its own, without requiring a human
79to manually notice something feels wrong first.
80""")

Gotchas

⚠This lesson's degraded_response is deliberately, artificially unrelated to its conversation context specifically to test whether the monitor catches an extreme, unambiguous case — a real production quality regression is often far subtler than this, and a real deployment should validate this monitor's sensitivity against genuinely subtle degradation too, not only extreme, obvious cases.
⚠Sampling rather than checking every single response is a deliberate, real production tradeoff — following the same real cost-versus-coverage tradeoff established throughout this course's evaluation content, checking every response with a real NLI model adds real, non-trivial latency and cost at high traffic volume.
⚠ALERT_THRESHOLD=0.5 is illustrative — a real production system should set this based on a real baseline groundedness rate measured during normal, healthy operation, then alert on a genuine, statistically meaningful DROP from that baseline, not an arbitrary fixed number chosen without reference to real historical data.

Step 4 — Serving the Complete System With Live Monitoring Endpoints

This closing step wraps the concurrent-safe memory, adaptive summarization, and automatic quality monitoring into a FastAPI server following Module 19's exact production pattern, exposing real, actionable monitoring endpoints — summarization frequency, groundedness rate, and active session count — giving a real team the same visibility this project's own testing steps used internally, now available continuously against real, live traffic.

04_serving_with_full_monitoring.py

python

1from fastapi import FastAPI
2from pydantic import BaseModel
3from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
4from langchain_core.output_parsers import StrOutputParser
5from langchain_core.runnables.history import RunnableWithMessageHistory
6from langchain_openai import ChatOpenAI
7import os
8
9os.environ["LANGCHAIN_TRACING_V2"] = "true"
10os.environ["LANGCHAIN_PROJECT"] = "deepleap-production-chatbot"
11
12app = FastAPI(title="Production Chatbot API")
13
14session_store: dict = {}
15
16@app.on_event("startup")
17def load_chatbot():
18    # model, prompt, chain, and chain_with_memory are all built ONCE
19    # here -- Module 19's exact load-once discipline, combined with
20    # this project's Step 1 concurrent-safe session store and Step 2's
21    # AdaptiveSummarizingHistory class
22    print("Production chatbot loaded: concurrent-safe memory, adaptive "
23          "summarization, and quality monitoring all active.")
24
25def get_session_history(session_id: str) -> "AdaptiveSummarizingHistory":
26    if session_id not in session_store:
27        new_history = AdaptiveSummarizingHistory()
28        new_history.session_id = session_id
29        session_store[session_id] = new_history
30    return session_store[session_id]
31
32model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
33prompt = ChatPromptTemplate.from_messages([
34    ("system", "You are a helpful assistant."),
35    MessagesPlaceholder(variable_name="history"),
36    ("human", "{question}"),
37])
38chain = (prompt | model | StrOutputParser()).with_retry(
39    stop_after_attempt=3, wait_exponential_jitter=True,
40)
41chain_with_memory = RunnableWithMessageHistory(
42    chain, get_session_history, input_messages_key="question", history_messages_key="history",
43)
44
45class ChatRequest(BaseModel):
46    session_id: str
47    message: str
48
49@app.post("/chat")
50async def chat(request: ChatRequest):
51    config = {"configurable": {"session_id": request.session_id}}
52    response = await chain_with_memory.ainvoke({"question": request.message}, config=config)
53
54    # A small, real sample of live traffic is checked for quality --
55    # Step 3's exact monitoring mechanism, now running against real requests
56    history_text = "\n".join(m.content for m in get_session_history(request.session_id).messages)
57    sample_and_check_quality(request.session_id, history_text, response)
58
59    return {"response": response}
60
61@app.get("/monitoring/summary")
62def monitoring_summary():
63    """The real, combined operational picture: active sessions,
64    summarization frequency, and current groundedness rate -- the
65    same three signals this project's own testing steps used
66    internally, now continuously available against real traffic."""
67    grounded_count = sum(1 for s in quality_samples if s["is_grounded"])
68    groundedness_rate = grounded_count / len(quality_samples) if quality_samples else 1.0
69
70    return {
71        "active_sessions": len(session_store),
72        "summarization_events": len(summarization_log),
73        "current_groundedness_rate": round(groundedness_rate, 4),
74        "quality_alert_active": groundedness_rate < ALERT_THRESHOLD,
75    }
76
77# Run with: uvicorn 04_serving_with_full_monitoring:app --host 0.0.0.0 --port 8000
78# Test: curl -X POST http://localhost:8000/chat -H "Content-Type: application/json" \
79#       -d '{"session_id": "user-1", "message": "Hello, how are you?"}'

Gotchas

⚠This server's session_store, summarization_log, and quality_samples are all in-memory, following the same explicit, repeated caveat throughout this course — a real production deployment needs all three persisted to real, durable storage, both to survive a restart and to work correctly across multiple server processes if scaled horizontally.
⚠quality_alert_active in the monitoring endpoint is exactly the kind of signal a real team would wire into an actual paging or notification system in production — this endpoint exposes the data, but genuine automatic alerting requires a real, external monitoring integration polling this endpoint or receiving this signal directly.
⚠Every piece of this final server traces directly back to an earlier step in this project or an earlier module in this course — Step 1's concurrent-tested session isolation, Step 2's adaptive summarization, Step 3's quality monitoring, and Module 18/19's retry logic and serving pattern — confirming genuine, complete assembly rather than new, unverified code introduced only here.