A reflection agent does not just generate a result: it evaluates its own work through a structured critique process. Instead of asking an LLM to directly generate its final result, an agentic workflow sends multiple prompts to the LLM giving it the opportunity to progress step by step towards high-quality output.

The reflection pattern can be implemented with LangGraph. It is convenient to create two separate agents (or nodes):

A writer node: responsible for generating content
A reviewer node: responsible for giving constructive feedback on the writer’s work

For demonstrations, we will use LangGraph’s Graph API to implement this pattern.

Graph API vs Functional API

LangGraph offers two different APIs for building agentic workflows:

Criterion	Graph API	Functional API
Definition	Defines the agent as a graph of nodes and edges	Defines the agent as a single function
Visualization	Yes, possible	No, no native visualization
Shared State	Yes, explicit management of shared state between nodes	No, no explicit state
Connections	Supports numerous branches and parallel paths	Suitable for linear workflows
Collaboration	Ideal for teams	Ideal for individual prototyping
Code style	Preferred if you like explicit graphs	Preferred if you like procedural code (normal IF/FOR loops)

When to use the Graph API?

You prefer to define your agent as a graph of nodes and edges
You need to visualize your workflow
You are managing shared state between multiple nodes
You have many branches or parallel paths
You work in a team

When to use the Functional API?

You prefer to define your agent as a single function
You already have existing procedural code
You prefer classic loops and conditions without explicit state management
You prototype linear workflows

2. Demo: Implementing the Reflection pattern with LangGraph

Project structure

The demonstration project is organized as follows:

reflection/
├── .env.example          # Variables d'environnement (exemple)
├── .langgraph_api/       # Checkpoints persistés par LangGraph
├── config.py             # Constantes et configuration
├── graph_simple.py       # Définition et compilation du graphe
├── langgraph.json        # Configuration de déploiement LangGraph
├── nodes.py              # Implémentation des nœuds writer, reviewer, publisher
├── prompts/
│   ├── reviewer_instructions.md   # Instructions système pour le reviewer
│   └── writer_instructions.md    # Instructions système pour le writer
├── requirements.txt      # Dépendances Python
├── sample_input.json     # Exemple d'entrée pour tester
└── state.py              # Définition de l'état et structures de données

The simple graph has three nodes:

writer_node: generates an initial or revised response
reviewer_node: evaluates the response and gives feedback
publisher_node: finalizes and publishes the approved response

A conditional edge decides, depending on the reviewer’s decision, which node to call next: writer_node (for review) or publisher_node (for publication).

State shared between all nodes includes:

revision_count: number of revisions made
original_customer_message: the original message from the customer
latest_message_response_by_writer: the last response from the writer
latest_reviewer_decision: the reviewer’s decision (APPROVE or REVISE)
latest_feedback_for_writer: the latest feedback from the reviewer for the writer
continue_revision: boolean indicator for continuation of revisions

State file: state.py

"""State and data structures for the reflection workflow."""

import os
from dataclasses import dataclass
from enum import Enum
from typing import Literal

from langgraph.graph import MessagesState

class Decision(str, Enum):
    """Enumeration for reviewer decision types."""

    APPROVE = "APPROVE"
    REVISE = "REVISE"

class NodeName(str, Enum):
    """Enumeration for node names in the graph."""

    WRITER = "writer_node"
    REVIEWER = "reviewer_node"
    PUBLISHER = "publisher_node"

@dataclass
class AIReviewerResponse:
    """Represents the AI reviewer's decision.
    
    Attributes:
        decision: The reviewer's decision (APPROVE or REVISE).
    """
    decision: Literal["APPROVE", "REVISE"]

class MessageResponseState(MessagesState):
    """Extended state for tracking reflection workflow.
    
    Attributes:
        revision_count: Number of revisions performed.
        original_customer_comment: The original customer feedback text.
        latest_feedback_for_writer: Most recent feedback from reviewer.
        latest_message_response_by_writer: Most recent response written.
        latest_reviewer_decision: Current decision from reviewer (APPROVE/REVISE).
        continue_revision: Flag indicating whether to continue revisions.
    """
    revision_count: int = 0
    original_customer_message: str
    latest_feedback_for_writer: str
    latest_message_response_by_writer: str
    latest_reviewer_decision: str
    continue_revision: bool = True

Status File Key Points:

MessageResponseState inherits from MessagesState (from LangGraph), which automatically manages message history
The Decision enumeration guarantees the consistency of APPROVE/REVISE values
The NodeName enumeration centralizes node names to avoid typo errors
AIReviewerResponse is a dataclass used for structured extraction of the reviewer decision via with_structured_output

Configuration file: config.py

"""Configuration constants for the reflection workflow."""

import os

# Constants
MAX_REVISIONS = int(os.environ.get("MAX_REVISIONS", 3))
DEFAULT_MODEL = os.environ.get("DEFAULT_MODEL", "gpt-5")
DEFAULT_TEMPERATURE = float(os.environ.get("DEFAULT_TEMPERATURE", 0.9))
MAX_REVISIONS_MESSAGE = os.environ.get(
    "MAX_REVISIONS_MESSAGE", 
    "MAX REVISIONS REACHED - APPROVING CURRENT VERSION"
)

The configuration file:

Load environment variables from .env file
Centralizes constants like MAX_REVISIONS, DEFAULT_MODEL, and DEFAULT_TEMPERATURE
Allows you to easily change application behavior without touching the code

Prompts for agents

Instructions for the writer (`prompts/writer_instructions.md`)

# Writer Instructions

## Role
You are a professional customer service representative writing responses to customer messages.

## Guidelines for Responses
- **Thank them** for taking time to share message
- **Address** any specific points they mentioned
- **Offer** additional assistance if needed
- **Encourage** future engagement
- **Keep tone** friendly and professional
- **Show appreciation** for their business

## Task
Write a professional, engaging response that encourages continued relationship
for the provided customer comment.

Instructions for the reviewer (`prompts/reviewer_instructions.md`)

# Reviewer Instructions

## Role
You are a senior customer service manager reviewing responses to customer messages.

## Evaluation Criteria
1. **TONE**: Appropriate for the customer's sentiment and concern level
2. **PERSONALIZATION**: Addresses the customer by name (if available) and references specific details
3. **COMPLETENESS**: Fully addresses all concerns or points raised
4. **PROFESSIONALISM**: Maintains company standards and brand voice
5. **SOLUTION-ORIENTED**: Offers concrete next steps where appropriate
6. **EMPATHY**: Shows understanding and care for customer experience

## Decision Guidelines
- **If recommending REVISE**: Provide specific suggestions for improvement
- **If APPROVE**: Explain why the response is ready to send

## Output Format
Format your feedback clearly with specific, actionable suggestions.
**DO NOT** rewrite the response to customer, just provide feedback and your decision.

Node implementation: nodes.py

Writer Node

The writer_node is the agent that writes the response to the client. Here’s what happens step by step:

The node examines the current state and determines whether it is a first draft or a revision
If revision_count > 0 and the reviewer’s decision is REVISE, we get the reviewer’s feedback
If it is a revision, we append the reviewer’s feedback to the context
We construct the complete LLM prompt by including the system instructions from the writer_instructions.md file, the original message from the client, and the feedback from the reviewer
We assemble the conversation history in the status messages
We call the LLM via ChatOpenAI with the model and the temperature of the configuration

def writer_node(
    state: MessageResponseState
) -> MessageResponseState:
    """Writer node that creates or revises responses to customer comments."""
    revision_count = state.get("revision_count", 0)
    latest_decision = state.get("latest_reviewer_decision")
    
    # Determine if this is a revision or initial write
    feedback = None
    if revision_count > 0 and latest_decision == Decision.REVISE.value:
        feedback = state.get("latest_feedback_for_writer", "")
    
    # Generate response
    messages = _create_writer_messages(state, feedback)
    
    # Initialize the LLM
    llm = ChatOpenAI(
        model=DEFAULT_MODEL,
        temperature=DEFAULT_TEMPERATURE
    )

    response = llm.invoke(messages)

    # Update and return state
    return _update_writer_state(state, response.content)

The helper function _create_writer_messages assembles the list of messages:

def _create_writer_messages(
    state: MessageResponseState, feedback: Optional[str] = None
) -> list:
    """Create message list for writer node invocation."""
    writer_prompt = open("prompts/writer_instructions.md").read()
    messages = [{"role": "system", "content": writer_prompt}] + state["messages"]
    original_message = state.get("original_customer_message", "")
    messages.append(HumanMessage(content=f"\n\nOriginal Customer Message: \n{original_message}\n\n"))
    
    if feedback:
        feedback_message = (
            f"\n\nReviewer Feedback:\n{feedback}\n\n"
            "Please revise your response based on this feedback."
        )
        messages.append(HumanMessage(content=feedback_message))
    
    return messages

Reviewer Node

The reviewer_node evaluates the written response and decides whether it should be approved or reviewed:

It first checks if the maximum number of revisions is reached → if yes, it automatically approves
It retrieves the last response from the writer and the original message from the client
It builds the review messages with the instructions from the reviewer_instructions.md file
It calls the LLM to get textual feedback
It extracts the structured decision (APPROVE or REVISE) via with_structured_output
It determines if a new revision is necessary

def reviewer_node(state: MessageResponseState) -> MessageResponseState:
    """Reviewer node that evaluates written responses and provides feedback."""
    revision_count = state.get("revision_count", 0)
    
    # Check if max revisions reached
    if revision_count >= MAX_REVISIONS:
        return {
            **state,
            "continue_revision": False,
            "latest_reviewer_decision": Decision.APPROVE.value,
            "messages": [
                HumanMessage(
                    content=MAX_REVISIONS_MESSAGE,
                    name=NodeName.REVIEWER.value
                ),
                # ...
            ] 
        }

    latest_response = state.get("latest_message_response_by_writer", "")
    original_message = state.get("original_customer_message", "")
    
    reviewer_prompt = open("prompts/reviewer_instructions.md").read()
    messages = [{"role": "system", "content": reviewer_prompt}] + state["messages"]
    review_content = (
        f"\n\nOriginal Customer Message: \n{original_message}\n\n"
        f"\n\nProposed Response: \n{latest_response}\n\n"
    )
    messages.append(HumanMessage(content=review_content))
    
    llm = ChatOpenAI(model=DEFAULT_MODEL, temperature=DEFAULT_TEMPERATURE)
    response = llm.invoke(messages)
    feedback_text = response.content

    # Extract decision using structured output
    decision_prompt = (
        "Based on the given feedback provided by the agent, identify if "
        "the agent suggests revision or approves the content as is. "
        f"Here is the feedback: \n{feedback_text}\n\n"
    )
    reviewer_response = llm.with_structured_output(
        AIReviewerResponse
    ).invoke(decision_prompt)
    decision = reviewer_response.get("decision", Decision.REVISE.value)
    
    continue_revision = (
        decision == Decision.REVISE.value and 
        revision_count < MAX_REVISIONS
    )

    return {
        **state,
        "latest_feedback_for_writer": feedback_text,
        "latest_reviewer_decision": decision,
        "continue_revision": continue_revision,
        "messages": [
            HumanMessage(
                content=f"{revision_count} - feedback for writer: {feedback_text}",
                name=NodeName.REVIEWER.value
            ),
            HumanMessage(
                content=f"{revision_count} - reviewer decision: {decision}",
                name=NodeName.REVIEWER.value
            ),
            HumanMessage(
                content=f"{revision_count} - continue revision?: {continue_revision}",
                name=NodeName.REVIEWER.value
            )
        ]
    }

Publisher node

The publisher_node is the final node: it receives the approved response and completes the workflow.

def publisher_node(state: MessageResponseState) -> MessageResponseState:
    """Final node that publishes the approved response."""
    return {**state}

Graph construction: graph_simple.py

"""Customer Comment Response System using LangGraph Reflection Pattern."""

from dotenv import load_dotenv
load_dotenv()  # Charger les variables d'environnement EN PREMIER

from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END

from state import MessageResponseState, Decision, NodeName
from nodes import writer_node, reviewer_node, publisher_node

def should_continue(state: MessageResponseState) -> str:
    """Conditional edge: route to writer for revision or publisher for approval."""
    if state.get("continue_revision", False):
        return NodeName.WRITER.value
    return NodeName.PUBLISHER.value

def create_reflection_graph() -> StateGraph:
    """Create and configure the LangGraph reflection workflow."""
    workflow = StateGraph(MessageResponseState)
    
    # Ajout des nœuds
    workflow.add_node(NodeName.WRITER.value, writer_node)
    workflow.add_node(NodeName.REVIEWER.value, reviewer_node)
    workflow.add_node(NodeName.PUBLISHER.value, publisher_node)
    
    # Ajout des arêtes
    workflow.add_edge(START, NodeName.WRITER.value)
    workflow.add_edge(NodeName.WRITER.value, NodeName.REVIEWER.value)
    workflow.add_conditional_edges(
        NodeName.REVIEWER.value,
        should_continue,
        {
            NodeName.WRITER.value: NodeName.WRITER.value,
            NodeName.PUBLISHER.value: NodeName.PUBLISHER.value
        }
    )
    workflow.add_edge(NodeName.PUBLISHER.value, END)
    return workflow

async def process_customer_message(
    customer_message: str,
    thread_id: str = "message_112233",
    checkpointer=None
):
    """Process a customer comment through the reflection workflow."""
    workflow = create_reflection_graph()
    
    app = (
        workflow.compile(checkpointer=checkpointer)
        if checkpointer
        else workflow.compile()
    )

    initial_state = {
        "messages": [HumanMessage(content=customer_message)],
        "original_customer_message": customer_message,
        "revision_count": 0,
        "latest_feedback_for_writer": "",
        "latest_message_response_by_writer": "",
        "latest_reviewer_decision": Decision.REVISE.value,
        "continue_revision": True
    }
    
    # IMPORTANT : quand on utilise un checkpointer, il faut fournir un thread_id
    config = {"configurable": {"thread_id": thread_id}}

    final_state = await app.ainvoke(initial_state, config=config)
    return final_state

async def main():
    """Demonstrate the reflection pattern with a sample customer comment."""
    sample_comment = (
        "I absolutely love my new laptop! The battery life is amazing and "
        "it's so fast. Best purchase I've made this year!"
    )
    
    checkpointer = InMemorySaver()
    result = await process_customer_message(
        customer_message=sample_comment,
        checkpointer=checkpointer,
        thread_id="message_112233"
    )
    
    print(f"\nComment: {sample_comment}")
    print(f"\nResponse: {result.get('latest_message_response_by_writer', 'N/A')}")

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

Important points when constructing the graph:

InMemorySaver is used as a checkpointer to store checkpoints in memory
When invoking a graph with a checkpointer, it is mandatory to provide a thread_id in the configurable section of the configuration
The should_continue function is the conditional edge: it reads the continue_revision field of the state to decide the next node

3. Demo: Running the graph and enabling LangSmith tracing

Enable LangSmith tracing

Observability is extremely important when building applications with LLMs. LangSmith gives us end-to-end visibility into how our application processes a request.

To activate tracing, simply define a few environment variables:

# Dans le fichier .env
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<votre_clé_API>

It’s that simple! You must make sure you have a LANGSMITH_API_KEY, which you obtain by creating an account on LangSmith.

Execution trace analysis

After enabling tracing and running the code, you can open the browser and view the full trace on LangSmith. Here is what we observe:

writer_node receives the status with the message from the client and generates the first response → this response becomes the input for the reviewer_node
reviewer_node evaluates the response and provides detailed feedback with an overall recommendation. In the example, the response is approved on the first try (APPROVE)
As no revision is needed, control passes to publisher_node
publisher_node completes the workflow

Sample test input (sample_input.json):

{
    "customer_message": "I absolutely love my new laptop! The battery life is amazing and it's so fast. Best purchase I've made this year!!",
    "initial_response": "",
    "reviewer_feedback": "",
    "final_response": "",
    "revision_count": 0,
    "messages": [
        {
            "role": "user",
            "content": "I absolutely love my new laptop! The battery life is amazing and it's so fast. Best purchase I've made this year!!"
        }
    ]
}

4. Demo: Extending the application with Human in the Loop

Concept of LangGraph interrupts

In LangGraph, we use interrupts to implement the Human-in-the-Loop (HITL) pattern. Interrupts allow you to pause graph execution at specific points and wait for external input before continuing.

What happens when you call interrupt?

At this exact line of code, graph execution is suspended
LangGraph immediately saves the current state of the graph via its persistence layer
Then it waits indefinitely until execution is explicitly resumed

How to resume execution?

We invoke the graph again, but this time we pass a Command command containing a resume value. This value becomes the return value of the original call to interrupt inside the node, so that execution continues exactly where it left off, with external input available.

# Exemple de code avec interrupt
from langgraph.types import interrupt, Command

def human_review_node(state):
    # Suspendre l'exécution et attendre une décision humaine
    human_response = interrupt({
        "Question": "Voulez-vous publier cette réponse ?",
        "AI response": state.get("latest_message_response_by_writer", "")
    })
    # La valeur de retour de interrupt() = valeur fournie lors du resume
    action = human_response.get("action")
    # ... traitement basé sur action

# Pour reprendre l'exécution après une interruption
final_state = await app.ainvoke(
    Command(resume={"action": "approve"}),
    config=config
)

Implementation of the human_review node

The human_review_node dynamically routes execution based on human input. It can return either the publisher_node or the rejection_node, depending on the selected action.

The node supports three actions:

approve: the response is approved and published
reject: the response is rejected
edit: the human reviewer modifies the content, then the response is published with the modified content

from langgraph.types import interrupt, Command
from typing import Literal

def human_review_node(
    state: MessageResponseState
) -> Command[Literal[NodeName.PUBLISHER.value, NodeName.REJECTION.value]]:
    """Human review node that pauses for human approval."""
    
    # Suspendre l'exécution ; le payload s'affiche sous result["__interrupt__"]
    human_review_response = interrupt({
        "Question": (
            "Do you want to publish the AI response? "
            "(response examples {'action': 'approve'} or {'action': 'reject'} or "
            "{'action': 'edit', 'edited_content': 'Thank you very much for your message.'})"
        ),
        "AI response": state.get("latest_message_response_by_writer", "")
    })

    action = human_review_response.get("action")
    state['human_review'] = action

    # Routage selon la réponse
    if action == "approve":
        return Command(
            goto=NodeName.PUBLISHER.value, 
            update={
                **state, 
                "messages": [HumanMessage(
                    content="Human reviewer approved the response.",
                    name=NodeName.HUMAN_REVIEW.value
                )]
            }
        ) 
    elif action == "reject":
        return Command(
            goto=NodeName.REJECTION.value, 
            update={
                **state, 
                "messages": [HumanMessage(
                    content="Human reviewer rejected the response.",
                    name=NodeName.HUMAN_REVIEW.value
                )]
            }
        )
    elif action == "edit":
        edited_content = human_review_response.get("edited_content")
        state["latest_message_response_by_writer"] = edited_content
        return Command(
            goto=NodeName.PUBLISHER.value, 
            update={
                **state, 
                "messages": [HumanMessage(
                    content="Human reviewer edited the response.",
                    name=NodeName.HUMAN_REVIEW.value
                )]
            }
        ) 
    else:
        return Command(goto=NodeName.REJECTION.value, update={**state})

def publisher_node(state: MessageResponseState) -> MessageResponseState:
    """Final node that publishes the approved response."""
    print("The response has been approved by human reviewer.")
    return {**state}

def rejection_node(state: MessageResponseState):
    """Final node that handles rejected responses."""
    print("The response has been rejected by human reviewer.")

Extended graph with HITL

The report is expanded to include the human_review field:

class MessageResponseState(MessagesState):
    # ... champs existants ...
    human_review: str = 'approve'  # 'approve' ou 'reject' ou 'edit'

The NodeName enumeration is also extended:

class NodeName(str, Enum):
    WRITER = "writer_node"
    REVIEWER = "reviewer_node"
    PUBLISHER = "publisher_node"
    HUMAN_REVIEW = "human_review_node"
    REJECTION = "rejection_node"

Construction of the extended graph with HITL:

def create_reflection_graph() -> StateGraph:
    """Create the HITL-extended LangGraph reflection workflow."""
    workflow = StateGraph(MessageResponseState)
    
    # Ajout des nœuds (5 nœuds maintenant)
    workflow.add_node(NodeName.WRITER.value, writer_node)
    workflow.add_node(NodeName.REVIEWER.value, reviewer_node)
    workflow.add_node(NodeName.HUMAN_REVIEW.value, human_review_node)
    workflow.add_node(NodeName.PUBLISHER.value, publisher_node)
    workflow.add_node(NodeName.REJECTION.value, rejection_node)
    
    # Ajout des arêtes
    workflow.add_edge(START, NodeName.WRITER.value)
    workflow.add_edge(NodeName.WRITER.value, NodeName.REVIEWER.value)
    workflow.add_conditional_edges(
        NodeName.REVIEWER.value,
        should_continue,
        {
            NodeName.WRITER.value: NodeName.WRITER.value,
            NodeName.HUMAN_REVIEW.value: NodeName.HUMAN_REVIEW.value  # Plus publisher directement
        }
    )
    workflow.add_edge(NodeName.PUBLISHER.value, END)
    workflow.add_edge(NodeName.REJECTION.value, END)

    return workflow

Example of HITL flow in process_customer_message:

async def process_customer_message(...):
    # ...
    # Premier invoke : le graphe s'arrête sur l'interruption HITL
    pending_state = await app.ainvoke(initial_state, config=config)
    print(pending_state)

    # Reprendre avec la décision humaine
    # True route vers publisher, False vers rejection
    final_state = await app.ainvoke(
        Command(resume={"action": "approve"}),
        config=config
    )
    return final_state

5. Demo: Extending the application with Memory and Context

Context vs. Memory

These two concepts are often confused but are distinct in LangGraph:

Context

Context is the information available to the agent at a given time. It generally includes:

Current user input
Recent conversation history
Any relevant document retrieved via techniques such as RAG (Retrieval-Augmented Generation)
System instructions
Environmental signals like tool outputs or API responses

Memory

Memory is what the agent retains over time to improve future responses and personalization. It is the agent’s long-term knowledge, built from previous interactions or stored information.

Short-term and long-term memory in LangGraph

We can think of memory in two categories:

Type	Description	Implementation in LangGraph
Short-term memory	Maintains message history within a single session	Agent state, persisted via `checkpointer`
Long-term memory	Stores information between different sessions	`memory store` (ex: `InMemoryStore`)

LangGraph manages short-term memory as part of agent state. This state is persisted via the checkpointer.

The memory store allows you to persist and retrieve information between different sessions.

In most real-world applications, we combine the two types of memory to construct a rich context for the LLM and generate better answers.

Implementation with InMemoryStore and semantic search

Using a memory store is quite simple:

Define a namespace: it’s simply a tuple that can have any length and represent anything (e.g. (user_id, "memories"))
Use store.put to save memories in this namespace
Use store.search on the same store to retrieve relevant stores

This can also be improved by providing an embedding template. This allows the store to index the memories and perform a semantic search.

New context schema (`state.py`)

@dataclass
class ContextSchema:
    user_name: str

Store configuration with semantic search (`graph_simple.py`)

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

async def main():
    # Création du store avec recherche sémantique activée
    embeddings = init_embeddings("openai:text-embedding-3-small")
    store = InMemoryStore(
        index={
            "embed": embeddings,
            "dims": 1536,
        }
    )

    # Stocker des mémoires utilisateur
    store.put(("user_112233", "memories"), "1", {"text": "My name is John"})
    store.put(("user_112233", "memories"), "2", {"text": "I am a software developer"})
    store.put(("user_112233", "memories"), "3", {"text": "I love technology and programming"})

    # Compiler le graphe avec le store
    app = workflow.compile(checkpointer=checkpointer, store=store)

    # Passer le contexte lors de l'invocation
    final_state = await app.ainvoke(
        initial_state,
        config=config,
        context={"user_name": "user_112233"}
    )

Extended writer node with memory and context (`nodes.py`)

from langgraph.runtime import Runtime
from langgraph.store.base import BaseStore

def writer_node(
    state: MessageResponseState,
    runtime: Runtime[ContextSchema],  # Accès au contexte
    store: BaseStore                   # Accès au memory store
) -> MessageResponseState:
    """Writer node with memory and context support."""
    
    # Extraire le nom d'utilisateur depuis le contexte
    user_name = runtime.context["user_name"]
    
    # Recherche sémantique dans le store
    items = store.search(
        (user_name, "memories"),
        query="What's the name of this user?",
        limit=1
    )
    memories = "\n".join(item.value["text"] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    
    print(f"Memories found for {user_name}: {memories}")
    
    revision_count = state.get("revision_count", 0)
    latest_decision = state.get("latest_reviewer_decision")
    
    feedback = None
    if revision_count > 0 and latest_decision == Decision.REVISE.value:
        feedback = state.get("latest_feedback_for_writer", "")
    
    # Générer la réponse avec les mémoires injectées dans le contexte
    messages = _create_writer_messages(state, feedback, memories)
    
    llm = ChatOpenAI(model=DEFAULT_MODEL, temperature=DEFAULT_TEMPERATURE)
    response = llm.invoke(messages)

    return _update_writer_state(state, response.content)

The _create_writer_messages function is extended to accept messages:

def _create_writer_messages(
    state: MessageResponseState,
    feedback: Optional[str] = None,
    memories: Optional[str] = None
) -> list:
    """Create message list for writer node invocation (with memory support)."""
    writer_prompt = open("prompts/writer_instructions.md").read()
    messages = [{"role": "system", "content": writer_prompt}] + state["messages"]

    # Injection des mémoires si disponibles
    if memories:
        messages.append(HumanMessage(content=memories))

    original_message = state.get("original_customer_message", "")
    messages.append(HumanMessage(content=f"\n\nOriginal Customer Message: \n{original_message}\n\n"))
    
    if feedback:
        feedback_message = (
            f"\n\nReviewer Feedback:\n{feedback}\n\n"
            "Please revise your response based on this feedback."
        )
        messages.append(HumanMessage(content=feedback_message))
    
    return messages

Data flow summary with memory:

Session précédente → store.put() → InMemoryStore (indexé avec embeddings)
                                           ↓
Nouvelle session → store.search() → Mémoires pertinentes extraites sémantiquement
                                           ↓
                                   Injectées dans le prompt du writer_node
                                           ↓
                               Réponse personnalisée générée

6. Demo: Deploying the application on LangSmith Cloud

Deployment prerequisites

LangSmith Cloud is a fully managed hosting platform designed specifically for agent-based workloads.

To deploy an application, the code must be hosted in a GitHub repository. Both public and private repositories are supported.

langgraph.json file

The langgraph.json file defines everything needed to deploy the application, including dependencies, environment variables, and other settings.

{
    "dependencies": ["."],
    "graphs": {
        "customer_response_reflection": {
            "path": "./graph_simple.py:create_reflection_graph",
            "description": "Customer comment response system using reflection pattern"
        }
    },
    "env": "./.env"
}

Explanation of fields:

dependencies: path to the directory containing the dependencies
graphs: dictionary of graphs to deploy, with their name, the path to the creation function, and a description
env: path to the .env file containing the environment variables

Deployment process

In the left sidebar of LangSmith, go to “Deployments”
Create a new deployment (if this is the first time or you are connecting a private repository, you need to link your GitHub account)
Select repository and branch
Provide correct path to langgraph.json file
Keep most other settings as default
Add required environment variables: OpenAI API key and any other necessary configuration values

Important: Do not include LANGSMITH_API_KEY in deployment environment variables. Since we are deploying directly to LangSmith, this variable is reserved and will be automatically provided by the platform.

Submit deployment and wait a few minutes
Monitor progress in the deployment details page
Once deployment is ready, a public endpoint will be available

We can use this endpoint:

Via the LangGraph SDK in code
Directly from the Studio UI

Via Studio: Click on “Studio” in the upper right corner of the deployment page → This opens an interactive interface to test and inspect the deployed application in real time.

7. Understand cost and latency implications

There can be several reasons why latency appears in agentic systems, and this is not specific to LangGraph applications.

Common sources of latency

Here are the most common causes:

Orchestration overhead: the cost of coordination between nodes
Model inference time: the time the LLM takes to generate a response
Tool execution times: the time taken by the tools (web search, external APIs, etc.)
Data recovery: retrieving information from a database or file system
Agent2Agent (A2A) Communication: In multi-agent systems, communication between agents can add delay
Architectural choices: patterns like reflection or retries naturally introduce additional steps (in our case, we used reflection intentionally because the improvement in quality was worth the compromise)

Performance Optimization Checklist

Here is a simple checklist for identifying performance bottlenecks in a LangGraph application:

1. Analyze the number of LLM calls

How many LLM calls does each node make per request?
Do we use an LLM where logic based on simple rules would suffice?
Are we using heavy reasoning models where a light model would do the job?

Tip: For tasks like routing, monitoring, or validation, you don’t need expensive models. Use smaller, faster models for flow control.

2. Look for parallelization opportunities

LangGraph supports parallel execution very well. For example:

Data recovery and metadata recovery can run together
Multiple tools can be executed at the same time
You can implement fan-out and fan-in patterns

3. Avoid redundant tool calls

Are you calling the same tool multiple times unnecessarily?
If the data does not change frequently (like weather data), call it only once and cache the result

4. Manage prompts and memory

Is your system prompt or chat history growing with each step?
If so, you may need better memory management

5. Revisiting Design Choices

Do you use heavy patterns (reflection, retries) where it is not necessary?
Can you simplify the workflow?

8. Demo: Extension of the application with Streaming

Streaming modes in LangGraph

With LangGraph, you can stream both the states of the graph and subgraphs during their execution. There are different stream modes:

Fashion	Description
`updates`	Only shows changes after each step
`values`	Gives full status every time
`custom`	Allows you to emit custom events from nodes or tools

One can also stream LLM tokens and even send user-defined custom events from inside nodes or tools during graph execution. LangGraph also supports combining multiple streaming modes.

Implementing streaming with custom events

In graph_simple.py, we configure streaming with updates and custom modes:

async def process_customer_message(...):
    # ...
    for chunk in app.stream(
        initial_state, 
        config=config, 
        context=context, 
        stream_mode=["updates", "custom"]  # Modes de streaming combinés
    ):
        print(chunk)

In nodes.py, we use get_stream_writer to access the StreamWriter and emit custom data:

from langgraph.config import get_stream_writer

def writer_node(
    state: MessageResponseState,
    runtime: Runtime[ContextSchema],
    store: BaseStore
) -> MessageResponseState:
    """Writer node with custom streaming events."""
    
    # Obtenir le StreamWriter
    writer = get_stream_writer()
    
    # Émettre un événement personnalisé (paire clé-valeur)
    writer({"writer_node": "extracting user name from the context"})

    user_name = runtime.context["user_name"]
    
    writer({"writer_node": "searching user memories from the store"})
    items = store.search(
        (user_name, "memories"), query="What's the name of this user?", limit=1
    )
    memories = "\n".join(item.value["text"] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    
    writer({"writer_node": f"Memories found for {user_name}: {memories}"})
    
    revision_count = state.get("revision_count", 0)
    latest_decision = state.get("latest_reviewer_decision")
    
    feedback = None
    if revision_count > 0 and latest_decision == Decision.REVISE.value:
        writer({"writer_node": "adding feedback for revision"})
        feedback = state.get("latest_feedback_for_writer", "")
    
    messages = _create_writer_messages(state, feedback, memories)
    writer({"writer_node": "prompt created, invoking LLM for response generation"})
    
    llm = ChatOpenAI(model=DEFAULT_MODEL, temperature=DEFAULT_TEMPERATURE)
    response = llm.invoke(messages)

    writer({"writer_node": "LLM response generated."})
   
    return _update_writer_state(state, response.content)

Streaming Key Points:

get_stream_writer() returns a StreamWriter which can be called from any node
We call writer({"key": "value"}) to emit a custom event in real time
This allows users to see what is happening in nodes during execution, not just after

9. Demo: Using the Responses API and running in the background

Background mode

Not all graph executions need to happen in real time. If we look at agents like Codex or Deep Research, they often take several minutes to solve complex problems, especially with reasoning models. This is where background mode becomes very useful.

Important point: You don’t always need to rely on LangChain integrations. LangGraph is a low-level orchestration framework, so you are free to use external libraries like the OpenAI client directly in your nodes or tools to unlock additional capabilities like background execution.

Implementation with the Responses API

There are two simple changes to enable background mode:

When initializing the LLM via ChatOpenAI, set use_responses_api=True
When invoking the graph, set background=True

To retrieve the result, we can use the OpenAI client directly and probe the response until it is complete.

from openai import OpenAI
from time import sleep
from langchain_openai import ChatOpenAI
from langgraph.config import get_stream_writer

def writer_node(
    state: MessageResponseState,
    runtime: Runtime[ContextSchema],
    store: BaseStore
) -> MessageResponseState:
    """Writer node with Responses API and background execution."""
    
    writer = get_stream_writer()
    writer({"writer_node": "extracting user name from the context"})

    user_name = runtime.context["user_name"]
    
    writer({"writer_node": "searching user memories from the store"})
    items = store.search(
        (user_name, "memories"), query="What's the name of this user?", limit=1
    )
    memories = "\n".join(item.value["text"] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    
    writer({"writer_node": f"Memories found for {user_name}: {memories}"})
    
    revision_count = state.get("revision_count", 0)
    latest_decision = state.get("latest_reviewer_decision")
    
    feedback = None
    if revision_count > 0 and latest_decision == Decision.REVISE.value:
        writer({"writer_node": "adding feedback for revision"})
        feedback = state.get("latest_feedback_for_writer", "")
    
    messages = _create_writer_messages(state, feedback, memories)
    writer({"writer_node": "prompt created, invoking LLM for response generation"})
    
    # Initialiser le LLM avec la Responses API activée
    llm = ChatOpenAI(
        model=DEFAULT_MODEL,
        temperature=DEFAULT_TEMPERATURE,
        use_responses_api=True,         # Activer la Responses API
        output_version="responses/v1"
    )

    writer({"writer_node": "Initializing OpenAI Responses API call with background=True"})

    # Invoquer avec background=True → la réponse est asynchrone
    response = llm.invoke(
        messages,
        background=True  # Mode arrière-plan
    )

    writer({"writer_node": f"Response id={response.id}"})
   
    # Utiliser directement le client OpenAI pour sonder la réponse
    client = OpenAI()
    resp = client.responses.retrieve(response.id)
    
    # Boucle de sondage jusqu'à ce que la réponse soit prête
    while resp.status in {"queued", "in_progress"}:
        writer({"writer_node": f"Current status: {resp.status}"})
        sleep(1)
        resp = client.responses.retrieve(resp.id)

    writer({"writer_node": f"Final status: {resp.status}\nOutput:\n{resp.output_text}"})
   
    return _update_writer_state(state, resp.output_text)

Explanation of background execution flow:

llm.invoke(messages, background=True) → The call returns immediately with a response.id (the response is waiting to be processed)
We use client.responses.retrieve(response.id) to probe the response status
As long as the status is "queued" or "in_progress", we wait 1 second and we probe again
When the status changes (completed), we get resp.output_text and continue

During the polling loop, the user is continually updated on the current status using the StreamWriter. Once the response is ready, we display the final output.

10. Appendices

Project file structure

adv-lg-tech-master/
├── reflection/                          # Version de base : pattern reflection simple
│   ├── .env.example
│   ├── .langgraph_api/
│   ├── config.py
│   ├── graph_simple.py
│   ├── langgraph.json
│   ├── nodes.py
│   ├── prompts/
│   │   ├── reviewer_instructions.md
│   │   └── writer_instructions.md
│   ├── requirements.txt
│   ├── sample_input.json
│   ├── sample_resume.json
│   └── state.py
├── reflection_hitl/                     # Version avec Human-in-the-Loop
│   └── ... (mêmes fichiers)
├── reflection_memory_context/           # Version avec mémoire et contexte
│   └── ... (mêmes fichiers)
├── reflection_responses_api/            # Version avec Responses API + background
│   └── ... (mêmes fichiers)
└── reflection_streaming/                # Version avec streaming personnalisé
    └── ... (mêmes fichiers)

Each folder represents an incremental stage of application development, adding new features to the previous version.

Environment variables

File .env.example:

OPENAI_API_KEY=<votre_clé_openai>
LANGSMITH_API_KEY=<votre_clé_langsmith>
LANGSMITH_TRACING=true
MAX_REVISIONS=3
DEFAULT_MODEL=gpt-5
DEFAULT_TEMPERATURE=0.9
MAX_REVISIONS_MESSAGE="MAX REVISIONS REACHED - APPROVING CURRENT VERSION"

Varies	Description	Default
`OPENAI_API_KEY`	OpenAI API key for LLM calls	Required
`LANGSMITH_API_KEY`	LangSmith API key for tracing	Required
`LANGSMITH_TRACING`	Enables/disables LangSmith tracing	`true`
`MAX_REVISIONS`	Maximum number of revisions before automatic approval	`3`
`DEFAULT_MODEL`	OpenAI model to use	`gpt-5`
`DEFAULT_TEMPERATURE`	Temperature for LLM generation	`0.9`
`MAX_REVISIONS_MESSAGE`	Message displayed when maximum revisions are reached	See above

Python Dependencies

requirements.txt file:

langchain
langsmith
langgraph
langchain-openai
python-dotenv
dotenv
langgraph-cli[inmem]
numpy
langgraph-api

Training created as part of learning advanced LangGraph techniques. Prerequisites: Introduction to Developing AI Agents and Developing Multi-agent Systems.

Search Terms

langgraph · techniques · ai · agents · orchestration · artificial · intelligence · generative · memory · node · context · application · graph · prompts · api · langsmith · reflection · streaming · writer · background · calls · configuration · deployment · extended

Table of Contents

1. Introduction to the Reflection pattern

What is a reflection agent?