Open Webui

Langgraph

Human In The Loop

Langchain

From Open WebUI to LangGraph: Building a Human-in-the-Loop Pipe for Real-Time AI Control

A hands-on guide to streaming conversations, handling interrupts, and keeping humans in the loop

Image from LangChain documentation

Introduction

Open WebUI provides a flexible interface for working with custom models and workflows, while LangGraph offers a runtime for building agent-based applications. Both tools are designed to be extensible, and connecting them allows for more interactive workflows. This is where Pipe functions come in: they route user inputs, handle metadata, and stream responses back to the chat interface.

In this article, we’ll go through a custom Pipe implementation that connects Open WebUI to a LangGraph agent with human-in-the-loop support. With this setup, the agent can pause mid-run to request feedback from the user and resume once input is provided. This enables a simple, interactive loop where humans and agents can work together during a conversation.

Requirements & Repository

The full implementation of this Pipe is available on GitHub:

GitHub - pessini/openwebui-langgraph-pipe: A custom Open WebUI Pipe for LangGraph with real-time…

Make sure you have the following libraries in your environment set up:

• Python ≥ 3.11
• langgraph-sdk = 0.2.6
• Open WebUI = 0.6.25
• LangGraph server = 0.6.7 (running locally or in Docker)

If you’re running Open WebUI in Docker, you’ll need to install langgraph-sdk inside the container for it to work.

Download the LangGraph Pipe directly from the Open WebUI community: LangGraph HITL Pipe

How the Pipe Works

The Pipe connects Open WebUI to LangGraph by handling each step of the interaction. At a high level, it does four things:

1. Extracts metadata from the request (user, chat, and message info).
2. Creates or reuses a thread in LangGraph to keep conversation state.
3. Streams input to the agent and listens for messages, tool calls, or interrupts.
4. Handles human-in-the-loop by pausing when the agent asks for feedback and resuming once the user replies.

This structure keeps the integration simple but flexible. Each part of the Pipe — initialization, metadata extraction, thread management, and streaming — has its own role. In the next sections, we’ll look at them one by one.

Extracting User Metadata

Before sending anything to LangGraph, the Pipe needs to know who the user is and what chat it belongs to. Open WebUI includes this information in the request metadata.

I created a custom BaseModel class with a helper function to help with that:

class Metadata(BaseModel):
  user: str
  user_id: str
  chat_id: str
  message_id: str
  group_id: str
  user_message: str
  model_name: str

def _extract_metadata(
        self, 
        body: dict, 
        __metadata__: Optional[Dict] = None, 
        __user__: Optional[Dict] = None
    ) -> Metadata:
  """
  Extract and validate metadata from OpenWebUI request.
  
  Returns:
      Metadata object containing validated metadata
      
  Raises:
      ValueError: If required metadata is missing
  """
  if __metadata__ is None:
      raise ValueError("Missing metadata")
  
  # Extract required fields
  user = __user__.get("email") if __user__ else "anonymous"
  user_id = __metadata__.get("user_id")
  chat_id = __metadata__.get("chat_id")
  message_id = __metadata__.get("message_id")
  
  if not user_id or not chat_id:
      raise ValueError("Missing user_id or chat_id in metadata")
  
  messages = body.get("messages", [])
  if not messages:
      raise ValueError("No messages provided")
  
  # Extract group_id safely
  try:
      group_id = (
          __metadata__.get("model")
          .get("info")
          .get("access_control")
          .get("read")
          .get("group_ids")[0]
      )
  except (AttributeError, KeyError, IndexError):
      group_id = "default"
  
  user_message = messages[-1]["content"]
  
  # Get model name for status messages
  model_name = "AI"
  if __metadata__ and "model" in __metadata__:
      model_name = __metadata__["model"].get("name", "AI")
  
  return self.Metadata(
      user=user,
      user_id=user_id,
      chat_id=chat_id,
      message_id=message_id,
      group_id=group_id,
      user_message=user_message,
      model_name=model_name
  )

Key points:

• Metadata helps tie each run back to the right chat and user.
• You can also extend this to track roles, session IDs, or other custom tags.

This metadata will later be attached when creating or resuming LangGraph threads.

Creating and Managing Threads

Each conversation in LangGraph happens inside a thread. A thread keeps track of state so the agent knows the context of previous messages.

Creating (or reusing) a thread looks like this:

# threads are created independently and then associated with assistants when you start a run.
async def create_thread(self, user_id: str, thread_id: str) -> str:
    """Create or get existing thread."""
    if self.client is None:
        if self.valves.DEBUG_LOGGING:
            logger.info("LangGraph client is not initialized. Initializing...")
        if not self._initialize_client():
            raise RuntimeError("Failed to initialize LangGraph client.")
            
    try:
        thread = await self.client.threads.create(
            thread_id=thread_id,
            metadata={"user_id": user_id},
            if_exists="do_nothing"
        )
        return thread["thread_id"]
    except Exception as e:
        logger.error(f"Failed to create thread {thread_id}: {e}")
        raise

Key points:

• Threads = conversation state. Without them, each run would be stateless.
• Use the chat_id from metadata to reuse existing threads and be consistent between the two services.
• New chats create new threads, existing chats continue in the same one.

With threads in place, we can now stream runs and handle updates in real time.

Streaming Runs with Interrupt Handling

Once the client and thread are ready, the Pipe streams a run from LangGraph, receiving events as they happen instead of waiting for a final result. This includes tokens, tool calls, messages, status updates, and interrupts, letting the Pipe react dynamically to the agent’s actions.

async def _stream_run(
        self,
        thread_id: str,
        assistant_id: str,
        input_data: Dict | Command,
        metadata: Dict = None,
        config: Dict = None
    ) -> AsyncGenerator[Union[str, Dict], None]:
  """
  Common streaming method for both new runs and interrupt handling.
  """
  
  try:
    # Get the current state of the thread
    thread_state = await self.client.threads.get_state(thread_id)
    
    # Check for interrupts in tasks
    interrupts = thread_state.get('interrupts', []) if isinstance(thread_state, dict) else getattr(thread_state, 'interrupts', [])
    has_interrupts = bool(interrupts)
    
    # Check if input_data is a user message
    is_user_message = isinstance(input_data, dict) and 'messages' in input_data
    is_command_input = False
    
    if has_interrupts and is_user_message:
        if self.valves.DEBUG_LOGGING:
            logger.info(f"Converting user message to resume command for thread {thread_id}")
  
        input_data = Command(resume=input_data['messages'])
        is_command_input = True
  
  except Exception as e:
    if self.valves.DEBUG_LOGGING:
        logger.info(f"Could not check thread state for interrupts: {e}, proceeding with normal run")
  
  # Build base arguments
  stream_args = {
      "thread_id": thread_id,
      "assistant_id": assistant_id,
      "metadata": metadata or {},
      "config": config,
      "stream_mode": ["messages-tuple", "updates"]
  }
  
  input_key = "command" if is_command_input else "input"
  stream_args[input_key] = input_data
  
  if self.valves.DEBUG_LOGGING:
    logger.info(f"Starting stream_run with args: {stream_args}")
  
  try:
  
    async for chunk in self.client.runs.stream(**stream_args):
        
      if not chunk or not chunk.data:
          continue
      
      if self.valves.DEBUG_LOGGING:
          self._pretty_print_chunk(chunk)
  
      event_type = chunk.event
     # Handle different event types
      if event_type == "updates":
          
        # Handle human-in-the-loop interrupts
        if "__interrupt__" in chunk.data:
            interrupt_data = chunk.data["__interrupt__"][0]
            interrupt_id = interrupt_data['id']
            interrupt_value = interrupt_data['value']
            
            if self.valves.DEBUG_LOGGING:
                logger.info(f"Interrupt encountered in thread {thread_id}: id={interrupt_id}, value={interrupt_value!r}")
            
            yield interrupt_value
            continue  # Stop streaming here, wait for next user message
  
          # Show tool calls if not hidden
          if self.valves.SHOW_TOOL_CALLS:
            try:
              # Get the first node data (whatever the node name is)
              node_data = next(iter(chunk.data.values()), None)
              if node_data and node_data.get('messages'):
                 tool_calls = node_data['messages'][0].get('tool_calls', [])
                 if tool_calls:
                   tool_name = tool_calls[0].get('name')
                    if tool_name:
                      yield self._status(f"🛠️ Tool {tool_name} requested...", done=False)
                            
            except (IndexError, KeyError, AttributeError, TypeError) as e:
              logger.error(f"Error extracting tool call info: {e}")
  
      elif event_type == "messages":
        message, metadata_chunk, *_ = chunk.data
  
        if isinstance(message, dict):
          content = message.get("content")
          if not content:
            continue
  
          # Skip if this is from a tools node
          if (isinstance(metadata_chunk, dict) and 
            metadata_chunk.get("langgraph_node") == "tools"):
            continue
          
          yield self._status()  # Clear status
          yield content  # Yield the actual message content
  
  except Exception as e:
    logger.error(f"Streaming error: {e}")
    yield self._status(f"❌ Streaming error: {str(e)}", done=True)

Key points:

• runs.stream provides real-time updates, including messages, status changes, tool calls, and interrupts.
• If the thread has pending interrupts and a user message arrives, it’s converted into a resume command automatically.
• Tool calls are detected in the stream and can trigger status messages if SHOW_TOOL_CALLS is enabled.
• Only relevant messages are yielded; tool-generated messages or empty content are skipped.
• Errors during streaming are caught and reported without stopping the run.

This setup ensures the Pipe doesn’t just send input and wait — it responds to the agent in real time, supporting interactive and human-in-the-loop workflows.

Final Pipe Function

This is the main pipe function from Open WebUI and since we already handled most of the heavy on the helper _stream_run() above, the function is actually very simple:

async def pipe(
        self,
        body: dict,
        __metadata__: Optional[Dict] = None,
        __user__: Optional[Dict] = None,
    ) -> AsyncGenerator[Union[str, Dict], None]:
  """
  Main pipe function for OpenWebUI integration.
  """
  
  try:
    # Extract and validate metadata
    metadata = self._extract_metadata(body, __metadata__, __user__)
    # Show initial status
    yield self._status(f"🧠 {metadata.model_name} is thinking... Please wait a moment.")
    # Create thread
    thread_id = await self.create_thread(metadata.user_id, metadata.chat_id)
  
    # Prepare metadata and input data
    run_metadata = {
        "langfuse_user_id": metadata.user,
        "langfuse_tags": [metadata.group_id],
        "langfuse_session_id": thread_id,
        "message_id": metadata.message_id
    }
    
     # Always use the same flow - _stream_run will handle interrupts internally
    input_data = {"messages": metadata.user_message}
    
    first_chunk = True
    async for chunk in self._stream_run(
        thread_id=thread_id,
        assistant_id=self.valves.MODEL_ID,
        input_data=input_data,
        metadata=run_metadata
    ):
        # Clear status on first chunk
        if first_chunk:
            yield self._status()  # Empty message clears status
            first_chunk = False
        
        yield chunk
  
  except ValueError as e:
    logger.error(f"Validation error: {e}")
    yield self._status(f"❌ Error: {str(e)}", done=True)
    return
  
  except Exception as e:
    logger.error(f"LangGraph pipe error: {e}", exc_info=True)
    yield self._status(f"❌ Error: {str(e)}", done=True)

Key points:

• Metadata → we use the helper function _extract_metadata() to use on further steps.
• Thread → use of another method create_thread() to get the thread ID.
• Stream→ this will call the method to handle all the messages.

Note: In my setup, I use Langfuse for tracing, which is why you’ll see extra metadata fields like langfuse_tags and langfuse_session_id. These are optional and can be removed if you don’t need Langfuse integration.

Debugging & Logs

The Pipe can emit structured logs and status events, making it easier to trace what happened during a run. This is especially useful when dealing with interrupts or tool calls, since you can see exactly where the agent paused or which tool it invoked.

def _pretty_print_chunk(self, chunk):
  """Basic pretty printing with pprint"""
  
  if chunk.event.upper() == "MESSAGES":
      return
  
  import pprint
  print(f"\n{'='*60}")
  print(f"EVENT TYPE: {chunk.event.upper()}")
  print(f"{'='*60}")
  
  # Convert chunk data to dict for better formatting
  if hasattr(chunk, 'data') and chunk.data:
      pprint.pprint(chunk.data, indent=2, width=120, depth=10)
  else:
      print("No data available")

You can also set from the Valves whether to show logs and present tool call status for the user.

class Valves(BaseModel):
  LANGGRAPH_SERVER_URL: str = Field(default="http://localhost:2024")
  MODEL_ID: str = Field(default="agent")
  VERSION: str = Field(default="v1.0")
  DEBUG_LOGGING: bool = Field(default=False)
  SHOW_TOOL_CALLS: bool = Field(default=False)

Wrapping Up

We walked through how the Pipe connects Open WebUI with LangGraph: from initializing the client and managing threads to streaming runs, handling interrupts, and surfacing tool calls. With this setup, you can create human-in-the-loop workflows where users stay in control of the agent.

Screenshot of the Agent with Human-in-the-loop

For simplicity, the repo includes a minimal version for testing the Pipe with a basic agent and a human-in-the-loop flow, so you can try it out locally before extending it to your own projects.

Contribute on GitHub! Start discussions to help improve the Pipe and make it more useful for everyone. Thanks for reading!