Businesses rely on multiple tools to manage operations, from customer service to project management. What if you could bring all these services into a single conversational interface? This blog explores how we built a Slack bot that connects with Model Context Protocol(MCP), allowing it to interact with various services be it GitHub, Jira or Stripe all at the same time. By leveraging LangGraph, we enable seamless processing and decision-making, making our bot truly intelligent. This blog will cover the motivation behind this project, the architecture of our solution, the technical implementation, challenges faced, and future improvements.

MCP (Model Context Protocol)

Model Content Protocol (MCP) is a standardized communication framework that allows AI-driven applications to interact with multiple external services, tools, and APIs in a structured manner. MCP enables seamless integration by defining a common interface for different systems, making it easier to retrieve, process, and share data across platforms.

In our case, MCP serves as the intermediary layer that allows our Slack bot to interact with various third-party services, such as GitHub, Jira, and Stripe. By leveraging MCP, we eliminate the need for hard-coded integrations, making the system highly scalable and adaptable.

Understanding the Architecture

1. Slack App

The Slack App acts as the interface through which users communicate with our system. When a user sends a message, such as “@bot check GitHub issue #123”, Slack captures this interaction and forwards it to our Modal Server. The Slack App enables seamless interaction within Slack, ensuring users can issue commands in natural language without leaving their workspace.

2. Modal Server (Event Processor)

The Modal Server serves as the event processor for incoming Slack messages. It listens to and handles Slack events, extracting the key information from user messages. Once it processes a request, it packages the event into a structured chat message and forwards it to the LangGraph Server for further decision-making. This component ensures that every user interaction is translated into a format that the system can process efficiently.

3. LangGraph Server

The LangGraph Server functions as the central decision-making unit of the system. Upon receiving a request from the Modal Server, it determines the appropriate course of action by analyzing the user’s intent. It decides whether an MCP server is required, selects the appropriate tool or API to call, and manages the overall workflow. By using the MCP Adapter, the LangGraph Server facilitates communication with external MCP services, ensuring accurate responses are generated for user queries.

4. MCP Adapter

The MCP Adapter is a crucial intermediary that allows the LangGraph Server to interface with multiple MCP servers. It acts as a bridge between our bot and external services. This component dynamically connects to the required MCPs, retrieves the necessary data, and sends it back to the LangGraph Server for processing. This abstraction layer eliminates the need for direct integrations with individual services, making the system scalable and adaptable.

5. Response Handling

Once the LangGraph Server, with the help of the MCP Adapter, has processed the request, the final response is sent back to Slack. The Slack App receives this response and posts it in the appropriate conversation thread. For instance, if a user requested a GitHub issue update, they would receive a response like, “GitHub Issue #123 is assigned to John Doe.” This step ensures that users receive clear, actionable responses directly within their Slack workspace, improving efficiency and collaboration.

Implementing the Slack Bot

Step 1: Setting Up the Slack App

To interact with Slack, we created a Slack App and configured it to listen for messages.

• Enabled Event Subscriptions.
• Defined a webhook endpoint /events/slack.
• Configured bot permissions to read and post messages.

Step 2: Building the Modal Server

The Modal Server is a FastAPI application that listens to Slack events and processes messages before sending them to LangGraph.

@app.post("/events/slack")
async def slack_endpoint(req: Request):
    return await APP_HANDLER.handle(req)

This ensures that messages mentioning the bot (e.g., “@bot check Jira”) are sent for processing.

Step 3: Processing Requests in LangGraph

LangGraph serves as the decision-making engine. It:

• Determines if the bot needs to call an MCP server.
• Routes the request to the appropriate tool.
• Uses AI-based reasoning to generate a response.

Example logic for handling Slack messages:

async def route(state: State, *, config: RunnableConfig) -> dict[str, list[BaseMessage]]:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Routing user query to appropriate service."),
        ("placeholder", "{messages}"),
    ])
    model = load_chat_model()
    response = await model.ainvoke(prompt.ainvoke(state.messages, config))
    return {"messages": [response]}

Step 4: Using MCP Adapter to Connect with External Services

The MCP Adapter allows our bot to interact with different services dynamically. It:

• Lists available tools.
• Routes user queries to the correct API.
• Retrieves and processes responses.

For example, fetching tools from an MCP server:

class GetTools(MCPSessionFunction):
    async def __call__(self, server_name: str, session: ClientSession) -> list[dict[str, Any]]:
        tools = await session.list_tools()
        return [{"name": tool.name, "description": tool.description} for tool in tools.tools]

Step 5: Sending Responses Back to Slack

Once LangGraph processes the request, the response is posted back to Slack:

await APP_HANDLER.app.client.chat_postMessage(
    channel=channel_id,
    thread_ts=thread_ts,
    text=response_text,
)

This ensures that users receive an immediate reply within their Slack channel.

Demo Video

To see the bot in action, check out our demo video below:

Challenges and Solutions

1. Handling Large Workflows

• Challenge: Some requests require multiple steps (e.g., fetching GitHub PR details and checking Jira status).
• Solution: We designed LangGraph to handle multi-step workflows efficiently, ensuring each request is processed in a structured manner.

2. Managing Multiple MCP Servers

• Challenge: Different MCP servers have unique APIs and data structures.
• Solution: The MCP Adapter abstracts the API differences, allowing seamless integration without requiring separate logic for each service.

3. Rate Limits on External APIs

• Challenge: APIs like Slack and GitHub impose rate limits, which could disrupt bot functionality.
• Solution: We implemented caching and batch processing to minimize API calls and ensure a smooth experience for users.

Conclusion

This project showcases how AI-driven automation can boost productivity by integrating multiple services into Slack. With LangGraph and MCP, businesses can streamline workflows, reduce app-switching, and provide a smoother user experience.

If you’re looking to enhance your team’s efficiency, consider building a Slack bot that connects all your tools into a single conversational interface.

Building a SlackBot for workflow automation using Model Context Protocol(MCP) was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

In our previous blog, we built a universal assistant that could use Claude’s Model Context Protocol (MCP) to interface with different services (such as our filesystem, a sqlite database, etc.), and help make our daily tasks more efficient.

In this blog, we’ll extend our Universal Assistant to be able to connect to almost any API and automatically call its endpoints!

You can access the code for this blog here.

Steps

Note: I’ve included links for all the relevant changes for greater clarity.

Initial Setup

In a regular MCP server setup, we’re limited by the number of MCP servers out there, or have to build our own. So adding support for each service we use with our assistant would mean creating a robust MCP implementation of the same. This approach might currently not be scalable.

Fortunately, there is an easier way of doing the same. Most of the popular APIs can be specified in a standard OpenAPI format. We just need a generic MCP server that can convert an OpenAPI spec into MCP tools.

After some research, I found this mcp server, which takes an OpenAPI Spec and dynamically exposes its endpoints as tools. All we have to do is add a config to our mcp-servers-config.json for our desired spec!

{
  "mcpServers": {
    ...
    "jira": {
        "description": "Project management and issue tracking, e.g., get, modify, or search projects, issues, bugs, etc.",
        "command": "npx",
        "args": [
            "-y",
            "openapi-mcp-server@1.1.0",
            "/path/to/openapi-specs/jira-openapi.json"
        ],
        "env": {
            "OPENAPI_MCP_HEADERS": "{\"Authorization\": \"Basic [base64 encoded version of: <YOUR.JIRA.EMAIL@YOUR.JIRA.TOKEN>]\"}",
            "JIRA_HOST": "your_workspace.atlassian.net",
            "JIRA_EMAIL": "you_email@example.com",
            "JIRA_API_TOKEN": "YOUR.JIRA.TOKEN"
        }
    }
  }
}

We’ve connected our JIRA workspace to our assistant using a simple config. Now, let’s tell our assistant to create a JIRA ticket for us.

It’s able to select the right tool (from a list of 500 or so tools!) most of the time. However, it cannot correctly map the information from the conversation context to tool arguments.

Tracing this with the help of LangSmith reveals the problem. The tool definitions returned by the MCP server has refs to the information schema (instead of the actual schema). Therefore, the target schema of tool input arguments is effectively not available to our tool orchestrator.

To address the problem, we need to make sure that the tool definition has resolved schema references.

Improving the openapi-mcp-server integration

Openapi-mcp-server by itself doesn’t check for the presence of refs and resolve them. This is what I needed to correct. After going through LangChain docs, I found a utility called openapi_spec_to_openai_fn, which automatically resolves the refs during the transformation.

So let’s extend GetTools of the mcp_wrapper.py with GetOpenAPITools.

class GetOpenAPITools(GetTools):
    def __init__(self, openapi_spec: dict, **kwargs):
        self.openapi_spec_dict = openapi_spec
        self.openapi_spec = OpenAPISpec.from_spec_dict(self.openapi_spec_dict)

    async def __call__(
        self, server_name: str, session: ClientSession
    ) -> list[dict[str, Any]]:
        tools = await session.list_tools()
        openai_fns = openapi_spec_to_openai_fn(self.openapi_spec)[0]
        
        # Making the JSON schema given by openapi_spec_to_openai_fn
        # compatible to Runtool function (MCP) 
        for fn, tool in zip(openai_fns, tools.tools):
            op_id = "".join(tool.name.split("-")[1:])
            # This will be useful later
            fn["metadata"] = {}
            fn["metadata"]["tool_info"] = extract_inlined_operation_data(
                self.openapi_spec_dict, op_id
            )

            fn["parameters"]["properties"] = merge_json_structure(
                fn["parameters"]["properties"]
            )
            fn["name"] = tool.name
            # truncate all openai_fn descriptions to 1024 characters
            if len(fn["description"]) > 1024:
                fn["description"] = fn["description"][:1021] + "..."

        return openai_fns

The graph may now use GetOpenAPITools instead of the usual GetTools while asking for tools from an openapi-mcp-server.

async def mcp_orchestrator(
    state: State, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
    """Orchestrates MCP server processing."""

    ...
    
    # Fetch tools from the MCP server conditionally
    tools = []
    args = (
        server_config["args"][1:]
        if server_config["args"][0] == "-y"
        else server_config["args"]
    )

    # Separate integration for openapi-mcp-server@1.1.0
    # TODO: refactor this into an adapter pattern later
    # You can also contribute to "Github: esxr/langgraph-mcp" for this!
    if args[0] == "openapi-mcp-server@1.1.0":
        openapi_path = args[1]

        # Get the openapi file as a json
        with open(openapi_path, "r") as file:
            openapi_spec = json.load(file)  # Converts JSON to a Python dictionary

        # convert the spec to openai tools
        tools = await mcp.apply(
            server_name,
            server_config,
            mcp.GetOpenAPITools(openapi_spec),
        )
    else:
        tools = await mcp.apply(server_name, server_config, mcp.GetTools())


    ...

This yields much better results!

However, though openapi_spec_to_openai_fn resolves the refs, it ignores some of the operation details available in the spec. E.g., it ignores examples, and any additional annotations that API spec providers may have added.

My first intuition was to enhance openapi_spec_to_openai_fn to include these, but for APIs like JIRA (with 500+ operations) that significantly increases the number of tokens with the bind tools. After some experimentation, the approach I chose was to selectively refine the schema details after the tool selection by the orchestrator.

The graph structure for that looks somewhat like this:

The refine_tool_call binds that single selected tool with the additional schema details and re-attempts the tool orchestration prompt.

TOOL_REFINER_PROMPT = """You are an intelligent assistant with access to various specialized tools.

Objectives:
1. Analyze the conversation to understand the user's intent and context.
2. Select the most appropriate info from the conversation for the tool_call
3. Combine tool outputs logically to provide a clear and concise response.

Steps to follow:
1. Understand the conversation's context.
2. Select the most appropriate info from the conversation for the tool_call.
3. If there is a tool response, combine the tool's output to provide a clear and concise answer to the user's query, or attempt to select another tool if needed to provide a more comprehensive answer.

{tool_info}

System time: {system_time}
"""

Here’s how the refine_tool_call node looks like in assistant_graph.py

async def refine_tool_call(
    state: State, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
    """Call the MCP server tool."""

    if state.current_tool == None:
        return

    # Fetch the current state and configs
    ...

    # Get the tool info
    tool_info = state.current_tool.get("metadata", {}).get("tool_info", {})

    # Bind the tool call to the model
    # Prepare the LLM
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", configuration.tool_refiner_prompt),
            ("placeholder", "{messages}"),
        ]
    )
    model = load_chat_model(configuration.tool_refiner_model)
    message_value = await prompt.ainvoke(
        {
            "messages": state.messages[:-1],
            "tool_info": str(tool_info),
            "system_time": datetime.now(tz=timezone.utc).isoformat(),
        },
        config,
    )

    # Replace the last tool call with this one
    # we don't want 2 tool calls in a line
    last_msg_id = state.messages[-1].id
    response = await model.bind_tools([state.current_tool]).ainvoke(
        message_value, config
    )
    response.id = last_msg_id

    return {
      "messages": [response], 
      "current_tool": None # unset the current tool
    }

You can view all the above changes together here.

Conclusion

In this article we saw how to turn any server end-point with an Open API specification to an MCP server using a generic wrapper. This greatly enhances the current reach of our MCP based Universal assistant!

Building a Universal Assistant to connect with any API was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

Model Context Protocol (MCP) is an open protocol that enables seamless integration between LLM applications and external data sources and tools. Whether you’re building an AI-powered IDE, enhancing a chat interface, or creating custom AI workflows, MCP provides a standardized way to connect LLMs with the context they need.

Think of MCP like a USB-C port for AI applications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCP provides a standardized way to connect AI models to different data sources and tools.

LangGraph is a framework designed to enable seamless integration of language models into complex workflows and applications. It emphasizes modularity and flexibility. Workflows are represented as graphs. Nodes correspond to actions, tools, or model queries. Edges define the flow of information between them.

LangGraph provides a structured yet dynamic way to execute tasks, making it ideal for writing AI applications involving natural language understanding, automation, and decision-making.

In this earlier article we enhanced LangGraph’s retrieval agent template to develop and deploy an AI solution.

In this work, we combine LangGraph with MCP to build our own Universal Assistant. You can access the code here.

Before we get into it, let’s experience what we built in this demo:

Note: In the next blog, we explore how to enhance the capabilities of this assistant even further, so that we can plug any OpenAPI spec for an API and connect it!

Building a Universal Assistant to connect with any API

Implementation

For our universal assistant we implement a multi-agent pattern as follows:

Assistant receives the user message and decides the agent to use. The agent node decides the right tool to use, and calls the tool on the MCP server. Since all our agents are based on MCP, a single MCP-Agent node is sufficient for LLM based orchestration, and another single node is sufficient to work with MCP servers to invoke their tools.

Our Implementation Walk-thru

There are 3 main parts to our implementation:

1. Building the Router
2. The Assistant
3. A generic MCP wrapper

Building the Router

Our graph to build the router is implemented in build_router_graph.py. It collects routing information based on tools, prompts, and resources offered by each MCP server using our mcp_wrapper.py. It indexes this routing information for each server in a vector database.

The Assistant

The assistant graph is implemented in assistant_graph.py. The following animation describes the role of various nodes and the flow of control thru it, with the help of an example

A Generic MCP Wrapper

mcp_wrapper.py employs a Strategy Pattern using an abstract base class (MCPSessionFunction) to define a common interface for executing various operations on MCP servers. The pattern includes:

1. Abstract Interface

• MCPSessionFunction defines an async __call__ method as a contract for all session functions.

2. Concrete Implementations

• RoutingDescription class implements fetching routing information based on tools, prompts, and resources.
• GetTools class implements fetching tools for the MCP server and transforming them to the format consumable by LangGraph.
• RunTool class implements invoking a tool on MCP server and returning its output.

3. Processor Function

• apply serves as a unified executor. It:
• Initialises a session using stdio_client from mcp library.
• Delegates the actual operation to the provided MCPSessionFunction instance via await fn(server_name, session).

4. Extensibility

• New operations can be added by subclassing MCPSessionFunction without modifying the cor e processor logic. for e.g. we should be able to add support for getting tools and executing tools using this pattern.

Setting it up

Create and activate a virtual environment

git clone https://github.com/esxr/langgraph-mcp.git 
cd langgraph-mcp python3 -m venv .venv 
source .venv/bin/activate

Install Langgraph CLI

pip install -U "langgraph-cli[inmem]"

Note: “inmem” extra(s) are needed to run LangGraph API server in development mode (without requiring Docker installation)

Install the dependencies

pip install -e .

Configure environment variables

cp env.example .env

Add your OPENAI_API_KEY, GITHUB_PERSONAL_ACCESS_TOKEN etc. to the .env

Note: We have added support for Milvus Lite Retriever (support file based URI). Milvus Lite won’t work on Windows. For Windows you may need to use Milvus Server (Easy to start using Docker), and change the MILVUS_DB config to the server based URI. You may also enhance the retriever.py to add retrievers for your choice of vector databases!

Conclusion

We can now try to connect it to services we use daily (e.g. JIRA, Slack, other channels etc) to help make our daily tasks more efficient!

We tackle that in the following blog:

Building a Universal Assistant to connect with any API

Building our own Universal Assistant with LangGraph and MCP was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

Let’s declare independence. Independence from $20 a month for ChatGPT subscription. and independence from having to be connected all the time.

Empowered with an Apple M2 MacBook Air (and just 16 GB RAM), we start with local deployment and running of large language models!

You can check out esxr.io for more.

Ollama

Based on a quick research and exploration of vLLM, llamaCPP, and Ollama, let me recommend Ollama! It is a great way to deploy quantized versions of LLMs on personal computers.

Let’s start by installing Ollama.

Let’s also take a quick look on leaderboard of LLMs.

Choosing a suitable model

The 16 GB of RAM we have might pose a slight challenge. A 7 billion parameter model, at full 32-bit precision, would require approximately 28 GB of RAM. However, at half precision (16 bits), it would need around 14 GB, and at 8-bit precision, about 7 GB. Therefore, a 7 billion parameter model with 8-bit quantization appears to be a viable option for our hardware specifications. Let’s choose Zephyr-7B-beta!

Quantization

Now, we obviously can’t run a full model on our computer. We will use quantization.

Model quantization in Large Language Models (LLMs) is like compressing a large, detailed photo into a smaller file size. It reduces the size of model’s “weight matrices” (by losing precision) to make it faster and use less computer memory while multiplying matrices in the background, similar to making a photo easier to send or load, but with slightly less detail. (for e.g. multiplying 2 doubles is more costly than multiplying 2 ints)

You can know more about quantization in detail here.

Getting a Quantized Model

Upon conducting research, we discover TheBloke on Huggingface, who is well known for publishing quantized versions of various models. Our next step is to explore his workspace on Huggingface to identify suitable quantized models for our needs.

Based on these use case notes it seems Q5_K_M quantization may be one of the best bargains for us.

Now let’s check, if Ollama has this model, in the quantization we want.

Running and testing the Model

Amazing! Let’s run a quick test. Let’s get our model to program Levenshtein distance function.

We type ollama run zephyr:7b-beta-q5_K_M in the terminal, and we get something like this.

Now, we ask the question.

Amazing! We just ran an LLM model on our very own machine!

Ollama WebUI

While command line is good, let’s get ourselves a ChatGPT like user experience. Amidst several user experience options available, I really liked Ollama Web UI. Let’s install and run it locally.

git clone https://github.com/ollama-webui/ollama-webui.git
cd ollama-webui

Follow the steps given in the repo to setup and run the UI. You can also check out esxr.io.

Now let’s run the same query using our newly installed UI.

Like ChatGPT, this system also maintains a record of our queries, allowing us to revisit them later. With this capability, we now have a personal LLM to perform tasks for us within our own computing environment.

Conclusion

In this blog, we have successfully installed and run a Large Language Model on our personal machine, leveraging Ollama. For convenience, we integrated it with a ChatGPT-style user-interface. We also learnt about model quantization, a technique enabling us to scale down large models with a trade-off in precision. This allows models to be hosted on personal computing environments, which typically have lesser computational power.

This blog serves as a foundational guide for any future projects in the realm of generative AI and LLMs, providing a solid base for exploration and development in this exciting field.

You can check out esxr.io for more fun stuff!

Host GPT Locally with Ollama was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

This series explains:

• Locally running a base AI model on your desktop
• Building any AI solution on top of that model, using your own data
• Attaching a UI to your solution within minutes

Empowered with this knowledge, anyone can create their own AI solution, tailored to their specific needs. The future of AI is accessible to everyone, we just need to take the first step!

Originally published at http://navvy.co on January 28, 2024.

Build and Deploy GenAI Solutions Locally was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

Adding ChatGPT like search system to the MacOS filesystem with Spotlight, Ollama, and LangChain

Introduction

In one of my YouTube videos, I locally hosted LLMs using Ollama, and interacted with them in a ChatGPT like interface using Ollama Web UI also hosted locally. This continues that journey. Here, I build a solution using a locally hosted LLM and LangChain (the most popular framework for building LLM solutions).

The Youtube video for the same is available here

Motivation

I love the built-in SpotLight on Mac. It indexes my documents and allows me to search for keywords and find relevant documents. In a lot of cases, I am not looking for a document, I am looking for answers to my questions that I know are somewhere there in my documents.

LLMs are good at answering questions. But they don’t have my documents. So I want to build a solution that retrieves relevant documents using Spotlight, and then passes them to LLM to generate answers to my questions. Such technique where we first retrieve information from a database or corpus, and then pass it on to an LLM to generate a response is also referred to as a Retrieval Augmented Generation solution. Since we’ll do everything local, let’s call this Local RAG.

Setup

Let’s begin by creating a python environment. Let me use conda for this.

conda create -n local_rag python=3.11
conda activate local_rag

Let’s now install langchain

pip install langchain

For this exercise I want to search thru PDF documents. So let’s install pypdf. It will help extract text from PDF documents.

pip install pypdf

Programmatic use of Spotlight

Let’s first write a python program to use Spotlight programmatically. Let’s create a shell program which receives the query as an input.

import argparse  # To parse command line arguments
def main():
    # Setup argument parser
    parser = argparse.ArgumentParser(description='Search files using Spotlight.')
    parser.add_argument('query', type=str, help='Search query')
    # Parse commandline arguments
    args = parser.parse_args()

if __name__ == "__main__":      # Entry point of the program
    main()

Let’s now write a function to search using spotlight for files matching this query. We’ll use the mdfind command and run it as a subprocess. We’ll then call this from our main function.

import argparse  # To parse command line arguments
import subprocess  # To run spotlight as a subprocess
def search_with_spotlight(query):
    # Prepare the command
    command = ["mdfind"]
    command += [query]

    # Execute mdfind command and capture output
    result = subprocess.run(command, capture_output=True, text=True)

    # Check if the command was successful
    if result.returncode != 0:
        print("Error in executing search")
        return []

    # Split the output into lines to get individual file paths
    file_paths = result.stdout.strip().split('\n')
    return file_paths

def main():
    # Setup argument parser
    parser = argparse.ArgumentParser(description='Search files using Spotlight.')
    parser.add_argument('query', type=str, help='Search query')
    # Parse commandline arguments
    args = parser.parse_args()
    # Perform spotlight search
    search_results = search_with_spotlight(args.query)
    # Print search results
    print(search_results)

if __name__ == "__main__":  # Entry point of the program
    main()

Let’s run it now.

python spotlight.py "search_with_spotlight using mdfind"

We should see the program file in results.

Let me also extend this to support restricting the search to a specific folder. We’ll add a new argument to our program to optionally take in a folder. We’ll relay this folder to the search function. And use it to restrict the search to this folder.

import argparse  # To parse command line arguments
import subprocess  # To run spotlight as a subprocess

def search_with_spotlight(query, folder=None):
    # Prepare the command
    command = ["mdfind"]
    if folder is not None:
        command += ["-onlyin", folder]
    command += [query]
    # Execute mdfind command and capture output
    result = subprocess.run(command, capture_output=True, text=True)
    # Check if the command was successful
    if result.returncode != 0:
        print("Error in executing search")
        return []
    # Split the output into lines to get individual file paths
    file_paths = result.stdout.strip().split('\n')
    return file_paths

def main():
    # Setup argument parser
    parser = argparse.ArgumentParser(description='Search files using Spotlight.')
    parser.add_argument('--folder', type=str, help='Folder to search in', default=None)
    parser.add_argument('query', type=str, help='Search query')
    # Parse commandline arguments
    args = parser.parse_args()
    # Perform spotlight search
    search_results = search_with_spotlight(args.query, folder=args.folder)
    # Print search results
    print(search_results)

if __name__ == "__main__":  # Entry point of the program
    main()

We can now run our search in a specific folder like this:

python spotlight.py --folder . "search_with_spotlight using mdfind"

Since we restricted our search to the current folder, it got us just our program file (which has the query string in it).

Layering LLM on top of Spotlight

While Spotlight gets us the files of interest containing certain keywords. It doesn’t precisely answer our questions. Let’s say we have a bunch of resumes in a folder, and want to ask a complex question like “Give me a short (about 100 words) summary, including contact details, of candidates having coursera certification in Generative AI”. Based on keywords like resume, coursera, generative AI; Spotlight may get us the resume files of interest. But it will not give us a short summary including contact details for the candidates. Let’s now use LLMs to do this for us.

Let’s begin by writing our driver function.

def generate_answer(question: str, file_paths: list[str], **kwargs: dict) -> str:
    # Get Document objects with text content from file paths
    docs = get_docs(file_paths)

    # Construct QA chain
    qa_chain = build_qa_chain()
    # Prepare input
    input = {"docs":docs, "question":question}
    input.update(kwargs)
    # Invoke the chain
    return qa_chain.invoke(input)

if __name__ == "__main__":  # Entry point of the program
    print(generate_answer(
        question="Give me a short, 100 words summary of each candidate having a coursera certification in Generative AI. Please include their contact details as well.",
        file_paths=["~/Documents/Resumes/Pranav Dhoolia.pdf"],
        doc_type="Candidate Resume"
    ))

For this exercise let’s assume all our documents are PDFs. Let’s write the get_docs function to extract text documents from pdf files.

from langchain.document_loaders import PyPDFLoader
from langchain.schema import Document
def get_docs(file_paths: list[str]) -> list[Document]:
    docs = []
    for file_path in file_paths:
        loader = PyPDFLoader(file_path)
        pages = loader.load()
        docs.append(Document(
            page_content="\n".join(page.page_content for page in pages),
            metadata={"source": file_path}
        ))
    return docs

def generate_answer(question: str, file_paths: list[str], **kwargs: dict) -> str:
    ...

Let’s now build our QA chain. We’ll use the LangChain Expression Language syntax for this. Let’s start by defining the Prompt, the LLM model, and the OutputParser

from langchain.document_loaders import PyPDFLoader
from langchain.prompts import PromptTemplate
from langchain.schema import Document, StrOutputParser
from langchain_community.chat_models import ChatOllama
from langchain_core.runnables.base import RunnableSerializable
def get_docs(file_paths: list[str]) -> list[Document]:
    ...

def build_qa_chain() -> RunnableSerializable:
    return (
        {}  ## Build the data to be injected into the prompt
        | PromptTemplate.from_template(
            "Use only the following context to answer the question at the end."
            "\nDo not use anything other than the context below to answer the question."
            "\nI'll repeat it is extremely important that you only use the provided context below to answer the question."
            "\nIf the context below is not sufficient to answer, just say that you don't know, don't try to make up an answer."
            "\n\nContext:\n\n{context}\n\nQuestion: {question}"
        )
        | ChatOllama(model="zephyr:7b-beta-q5_K_M")
        | StrOutputParser()
    )

def generate_answer(question: str, file_paths: list[str], **kwargs: dict) -> str:
    ...

Let’s now build the data to be injected into the prompt. we’ll use the docs, question, and doc_type expected while invoking the chain.

...
from langchain_core.prompts import format_document
...
def build_qa_chain() -> RunnableSerializable:
    def build_context(docs: list[Document], doc_type: str="Document") -> str:
        doc_prompt = PromptTemplate.from_template("{page_content}")
        context = ""
        for i, doc in enumerate(docs):
            context += f"{doc_type} {i + 1}\n---\n{format_document(doc, doc_prompt)}\n\n"
        return context
    return (
        {
            "context": lambda input: build_context(input["docs"], input.get("doc_type", "Document")),
            "question": lambda input: input["question"]
        }
        | ...
    )

Debugging LLM solutions

One of the reasons I really love LangChain, is that with LangSmith, it offers a very powerful debugging experience. So before running our LLM layer, let’s set up LangSmith.

1. Go to LangSmith and login with your credentials

2. Create your API key for your Organization

3. Create a new project

4. Setup the environment variables. I have set these up in my .zshrc.

5. Let’s now run our LLM layer.

It returns here a quick summary of the candidate matching the query, along with their contact details.

Let’s now goto LangSmith and see the various aspects of LLM execution that happened here!

• Let go back to our project and check Traces

We see a trace for our chain execution. Let’s expand it

We can see the things corresponding to components we used in building the chains, i.e., the data, the prompt, the chat model, and the output parser

For the data, we can see the input, and what the context built from it looks like

For the prompt, we can see data input to the template, and the prompt generated from it

For the chat model, we can see the prompt input to the model, and the output generated by it

For the output parser, we can see the input to it, and the output produced by it.

This offers a powerful general debugging experience.

Further, if we want to fine tune our model, based on the usage of our application over a period of time, we can select the traces we want to use for fine tuning, and easily construct a dataset to use for fine tuning from it like this.

Select the trace(s) and click on the Add to Dataset button

Since we don’t already have a dataset, we’ll create a new one

And now we can see a preview of dataset being created as well. — In the preview here we can see as input columns: the context, the question, and the doc domain type, and as output the LLM’s response.

Over a period of time, once we have enough traces, we can select and add all good traces to this dataset, and use it for fine tuning our model!

Combining the two

So we have built a programmatic interface to Spotlight, and we have built an LLM based QA chain that given some content can precisely answer a question based on that content. We now need to chain the two together. Before we do that, let’s first understand a little complication that we anticipate here. As we saw earlier, for our complex query, Spotlight didn’t return anything. That’s because it searched for files contained all the words in our query. But our query really wants a much smaller subset of keywords to be searched for. Most of the query is really a description of what we want from the files found containing that subset of keywords.

So we need to first extract the keywords from our query, and then use Spotlight to search for files containing those keywords. Once again an LLM chain is perhaps the best for identifying those keywords so let’s write first write a function to do that.

import json
from langchain.prompts import PromptTemplate
from langchain.schema import StrOutputParser
from langchain_community.chat_models import ChatOllama

def extract_keywords(query: str) -> list[str]:
    # Construct the keyword extraction chain
    keyword_extraction_chain = (
        {
            "query": lambda query: query
        }
        | PromptTemplate.from_template(
            "From the query below please extract search keywords. "
            "The search keywords must be part of the query string. "
            "Avoid words general to this domain such as: candidates, resume, contact, and details. "
            "Output a json formatted list of keyword strings. "
            "Each item in the output json should be a string."
            "\n\nQuery:\n```{query}\n```"
        )
        | ChatOllama(model="zephyr:7b-beta-q5_K_M")
        | StrOutputParser()
    )
    # Invoke the chain
    response = keyword_extraction_chain.invoke(query)
    # Get the substring between last `[` and `]`
    return json.loads(f'[{response[response.rfind("[") + 1:response.rfind("]")].strip()}]')

if __name__ == "__main__":  # Entry point of the program
    print(extract_keywords("Give me a short (about 100 words) summary of candidates having coursera certification in Generative AI. Please include their contact details."))

Now that we have our search keywords extractor, let’s modify our main function to combine everything.

...
from keywords import extract_keywords
from qa import generate_answer
def search_with_spotlight(query, folder=None):
    ...
    return file_paths

def main():
    ...
    args = parser.parse_args()
    # Let's extact keywords from the query
    keywords = extract_keywords(args.query)
    # Construct an OR query from the keywords
    query = " OR ".join(keywords)
    # Perform spotlight search
    search_results = search_with_spotlight(query, folder=args.folder)
    # Generate answer from the query and top 3 search results
    return generate_answer(args.query, search_results[:3], doc_type="Candidate")

if __name__ == "__main__":  # Entry point of the program
    print(main())

Let’s now run everything together!

python spotlight.py --folder ../test/resume "Give me a short (about 100 words) summary of candidates having coursera certification in Generative AI. Please include their contact details."

Works like a charm!

Conclusion

We successfully used locally hosted LLMs along with Langchain framework to build an LLM solution. This solution provided amazing answers to quite complex questions on our local content. What’s more, we didn’t even have to setup any additional tools.

We used Spotlight to search for files containing certain keywords, and then used LLMs to precisely answer questions based on those files. We also saw the power of the LangChain framework, in how easy it was to use LangSmith to debug our LLM based solutions, and how easily we could use it to create datasets, based on the usage traces of our solution, to fine tune our models.

Adding ChatGPT-Like functionality to MacOS Spotlight Search was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

Let us create a Generic Chat UI that we can use with ANY LLM solution we create in future!

In a previous blog, we learnt how to extend the MacOS Spotlight search and build a smart solution on top of it that answers our questions by using the filesystem as a database.

We tested it on resumes stored in a folder to only retrieve resumes matching a specific criteria. However, the only way we could interact with such a system was through a commandline interface. Not exactly the dream!

We want our solution to look somewhat like that of ChatGPT! As we saw in the first video, Ollama WebUI offers a very similar user experience. However, Ollama WebUI has primarily been designed to allow interactions with raw, out-of-the-box LLMs. Our motivation here is to use Ollama WebUI as the UI for our custom local RAG solution. So let’s get started!

The code to the repo is provided here:

https://github.com/esxr/local_llms_rag_solution

Our intuition here is that Ollama web UI expects certain API end-points to be available to interact with Ollama’s API, i.e., the chat, generate, list, … etc. The main API being the chat which is invoked when we send a message. So if can provide the same API, and wrap it around our custom solution, we should be able to use Ollama WebUI for our user experience. And that’s exactly what we pursue in this video.

If you haven’t seen the previous blog or watched the videos, its recommended to watch them because this is an extension of that.

Setup

Let’s start by setting up the environment. To build our wrapper API we will use FastAPI. We will make pass-thru requests to Ollama’s API for all the other end-points that we don’t want to handle.

conda create -n gen_ui python=3.11
conda activate gen_ui
pip install fastapi uvicorn httpx

Let’s build & configure Ollama WebUI

git clone https://github.com/ollama-webui/ollama-webui.git
cd ollama-webui/

# Copying required .env file
cp -RPp example.env .env

Let’s now configure the .env file. We want our wrapper API to sit in middle of Ollama Web UI and Ollama. So we’ll modify the configuration here to point to our wrapper API instead of Ollama’s API.

Let’s now build the Ollama WebUI frontend.

# Building Frontend Using Node
npm i
npm run build

Let’s now install the Ollama WebUI backend dependencies.

cd ./backend
pip install -r requirements.txt -U

Let’s configure the backend to use our wrapper API instead.

Let’s now serve Ollama WebUI backend.

# Serving Frontend with the Backend
sh start.sh

Pass-thru Wrapper

Let’s now start with wrapper API. Let’s begin by writing a pass-thru wrapper first. FastAPI extends the startlette library, which provides us with base middleware classes. We’ll use the BaseHTTPMiddleware to intercept all client requests coming on to our server, send them to Ollama’s API, and return the response to the client. We’ll add this middleware to the FastAPI application. We’ll use uvicorn to run our FastAPI application on port 5000.

# main.py
import asyncio
import json
from datetime import datetime

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
app = FastAPI()
OLLAMA_SERVER_URL = "http://localhost:11434"
class RelayMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        other_server_url = f'{OLLAMA_SERVER_URL}{request.url.path}'
        body = b""
        async for chunk in request.stream():
            body += chunk
        async with httpx.AsyncClient() as client:
            req_data = {
                "method": request.method,
                "url": other_server_url,
                "headers": request.headers.raw,
                "params": request.query_params,
                "content": body
            }
            response = await client.request(**req_data)
            return Response(response.content, status_code=response.status_code, headers=dict(response.headers))
# Add the middleware to the FastAPI application
app.add_middleware(RelayMiddleware)
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000, log_level="info")

Let’s do a quick test of our wrapper. Make sure to have Ollama running on port 11434.

Go to the dashboard at: http://localhost:8080. You should see the dashboard without any errors.

Now select the model you want and chat. You should see the chat working as expected.

Great! So our pass-thru wrapper is working as expected.

Customizing /api/chat

Let’s now override the /api/chat in our wrapper. Let’s begin by excluding /api/chat from the pass-thru wrapper. We’ll do this by adding a condition in the middleware to exclude /api/chat from the pass-thru wrapper.

...
class RelayMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Check if the request is for the /api/chat route
        if request.url.path == "/api/chat":
            # If so, just call the next item in the middleware stack
            return await call_next(request)

# Otherwise, handle the request as before
        ...

Before we customize the /api/chat endpoint, Let’s do a quick network analysis of Ollama WebUI to see what it expects from the /api/chat endpoint.

We find that:

1. The /api/chat is a POST request with a text/event-stream content-type, and a application/x-ndjson response.

2. The request body is a JSON object with the following structure:

{   
  "model": "<model_name>",   
  "messages": [     
      {       
        "role": "<user|assistant>",       
        "content": "<text>"     
      }   
  ] 
}

3. The response is a application/x-ndjson stream. With each message having line having the following structure:

{     
  "model": "<model_name>",     
  "created_at": "<datetime.utc.isoformat()>",     
  "message": {"role": "assistant", "content": "<text>"},     
  "done": <False|True> // True for the last message 
}

4. We learnt that the chat endpoint is a streaming endpoint. We read the input as an event-stream, and we’ll format the dummy output to the Newline delimited JSON (application/x-ndjson) format. NDJSON is a convenient format for storing or streaming structured data that may be processed one record at a time. Each line in an NDJSON file is a valid JSON value, but the entire file is not a valid JSON array. This format is particularly useful for streaming LLM outputs. Ollama uses it as well.

Let’s now customize the /api/chat endpoint to first push some dummy response. To begin with the processing logic is just a placeholder, and pretty much echoes back the request content.

...
@app.post("/api/chat")
async def chat(request: Request):
    def process(input_data):
        # Process the input_data and return a string
        # This is a placeholder for actual processing logic
        processed_string = "Processed: " + str(input_data)
        return processed_string

async def generate_ndjson(model: str, msg: str):
        for word in msg.split():
            yield json.dumps({
                "model": model,
                "created_at": datetime.utcnow().isoformat() + "Z",
                "message": {
                    "role": "assistant",
                    "content": word + " "
                },
                "done": False
            }) + "\n"
            await asyncio.sleep(0.1)
        yield json.dumps({
            "model": model,
            "created_at": datetime.utcnow().isoformat() + "Z",
            "message": {"role": "assistant", "content": "."},
            "done": True
        }) + "\n"
    # initialize input
    input_string = ""
    # Extract input from Ollama WebUI request (MIME type: text/event-stream)
    async for bytes in request.stream():
        if bytes:
            input_string = bytes.decode()
        else:
            continue
    # Process input_data
    input_data = json.loads(input_string)
    output = process(input_data)
    # Stream output in Ollama WebUI desired format (MIME type: application/x-ndjson)
    return StreamingResponse(generate_ndjson(model=input_data["model"], msg=output), media_type="application/x-ndjson")
...

Let’s now test our customized /api/chat endpoint.

Voila, our dummy echo based response is working as expected.

Wrap /api/chat around our RAG solution

Let’s now modify the chat API to invoke our RAG solution. But before that, we’ll need to modify our RAG solution. If you can recall from the repo, the code to our main function looks somewhat like this:

...
def main():
    # Setup argument parser
    parser = argparse.ArgumentParser(description="Search files using Spotlight.")
    parser.add_argument("--folder", type=str, help="Folder to search in", default=None)
    parser.add_argument("query", type=str, help="Search query")

    # Parse commandline arguments
    args = parser.parse_args()
    # Create search query
    keywords = extract_keywords(args.query)
    query = " OR ".join(keywords)
    # Perform spotlight search
    search_results = search_with_spotlight(query, folder=args.folder)
    # Generate answer for the original query and top 3 search results
    print(
        generate_answer(
            args.query,
            search_results[:3],
            doc_type="Candidate"
        )
    )

if __name__ == "__main__":  # Entry point of the program
    main()
...

Let’s modify this code a bit so that it can be used with other programs as well. Our final code should look somewhat like this

def chat(query, folder=None):
    # Let's extact keywords from the query
    keywords = extract_keywords(query)

    # Construct an OR query from the keywords
    search_query = " OR ".join(keywords)
    # Perform spotlight search
    search_results = search_with_spotlight(search_query, folder=folder)
    # Generate answer for the original query and top 3 search results
    print(generate_answer(query, search_results[:3], doc_type="Candidate"))
    # Print search results
    print(search_results)

if __name__ == "__main__":  # Entry point of the program
    # Setup argument parser
    parser = argparse.ArgumentParser(description='Search files using Spotlight.')
    parser.add_argument('--folder', type=str, help='Folder to search in', default=None)
    parser.add_argument('query', type=str, help='Search query')
    # Parse commandline arguments
    args = parser.parse_args()
    chat(args.query, args.folder)

Note that we’ve changed the function signature, and simplified it, so that we’re left with this signature

...
def chat(query, folder=None):
...

We can now import and plug this function anywhere we want to invoke the LLM-powered search.

let’s go back to our main.py` file for this project, and change the process function like so, replacing

def process(input_data):
        ...
        processed_string = "Processed: " + str(input_data)
        return processed_string

with this

def process(input_data):
        ...
        processed_string = chat(
            input_data["messages"][-2]["content"],
            folder=["~/Documents/Resumes/"]
        )
        return processed_string

That’s it! That’s all that was required to link our RAG solution to the API endpoint. We can now test the solution to see if it works.

Works like a charm!

Conclusion

In this blog, we have successfully designed a rapid and efficient approach to provide a user interface for all future projects, utilizing an existing open-source project (Ollama WebUI) as a foundation. Additionally, we have gained substantial insights into streaming APIs, a commonly utilized element in the realm of generative AI and LLMs.

We have also explored the API structure required by the Ollama WebUI and developed a wrapper for Ollama. This wrapper intercepts requests intended for Ollama and redirects them to the appropriate endpoints in our solution, seamlessly integrating with the Ollama framework.

Making a ChatGPT-like UI for your application in 15 minutes was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.

LangGraph is the most popular platform for developing AI solutions. It provides ready templates for common AI solution patterns. We took its retrieval agent template and customised it for our client needs. In the subsequent sections I describe those customisations in detail.

Please refer to the code in this repo to follow along.

Start with langgraph template

Let’s install the LangGraph CLI

pip install langgraph-cli --upgrade

Let’s now create a new app from a template

langgraph new

Indexing Graph Enhancements

Add starter_urls, and hops to IndexConfiguration in configuration.py. Also add a method to get the list of urls from the comma-separated starter_urls string.

starter_urls: str = field(
	default="https://zohlar.com",
	metadata={
		"description": "Comma-separated string of starter URLs to crawl for indexing web pages."
	},
)

hops: int = field(
 default=2,
 metadata={
  "description": "Maximum number of hops to traverse pages linked to the starter URLs."
 },
)

def parse_starter_urls(self) -> list[str]:
 """Parse the starter URLs into a list.
 Returns:
  list[str]: A list of URLs parsed from the comma-separated string.
 """
 return [url.strip() for url in self.starter_urls.split(",") if url.strip()]

Support For Open-source Retriever (Milvus)

Milvus is open source, and its Lite version can work off a file URI. This makes things economical (for dev), and handy for demos.

Let’s begin by adding langchain-milvus as a dependency in pyproject.toml

Then add milvus to retriever_provider list

retriever_provider: Annotated[
	Literal["elastic", "elastic-local", "pinecone", "mongodb", "milvus"],
	{"__template_metadata__": {"kind": "retriever"}},
] = field(
	default="milvus",
	metadata={
		"description": "The vector store provider to use for retrieval. Options are 'elastic', 'pinecone', 'mongodb', or, 'milvus'."
	},
)

Now to add a new method to create a milvus retriever in retrieval.py

@contextmanager
def make_milvus_retriever(
    configuration: IndexConfiguration, embedding_model: Embeddings
) -> Generator[VectorStoreRetriever, None, None]:
    """Configure this agent to use milvus lite file based uri to store the vector index."""
    from langchain_milvus.vectorstores import Milvus

vstore = Milvus (
    embedding_function=embedding_model,
    collection_name=configuration.user_id,
    connection_args={"uri": os.environ["MILVUS_DB"]},
    auto_id=True
)
yield vstore.as_retriever()

and then use this in the factory method

@contextmanager
def make_retriever(
    config: RunnableConfig,
) -> Generator[VectorStoreRetriever, None, None]:
    # ... same code as before

match configuration.retriever_provider:
      # ... same code as before
      case "milvus":
          with make_milvus_retriever(configuration, embedding_model) as retriever:
              yield retriever
      case _:
          # ... as before

We’ll also add the following file uri in .env to store the vector index:

## Milvus
MILVUS_DB=milvus.db

Support for Crawling

Out of the box implementation of index_graph (in index_graph.py) expects as input all the documents to be indexed. Since we are enhancing the graph to include an ingestion pipeline that crawls starting at the specified URL, we’ll modify the index_docs node to kick start the crawl if docs list in the state is empty and starter_urls configuration has been provided.

Custom Crawler (using Playright)

We can create a Crawler Component using playwright, which uses headless browsers (Note: Please add playwright & requests as dependencies in pyproject.toml)

You can access the code to that here.

Note: Just adding playwright package to the python environment is not sufficient. The headless browser, and its dependencies also need to be installed. So we need to run playright install, and playwright install-deps as well.

You may use LangGraph Studio app on mac, using docker to locally deploy your graphs. In this case playwright install and playwright install-deps need to happen in the docker. For that we’ll add the following to langgraph.json

"dockerfile_lines": ["RUN pip install playwright", "RUN python -m playwright install", "RUN python -m playwright install-deps"]

APIfy Crawler

For simpler usecases, we can also use a readymade apify based crawler. We have to modify the indexx_graph.py like so:

# new imports
import json

from langchain_community.utilities import ApifyWrapper
from langchain_community.document_loaders import ApifyDatasetLoader
# ... existing code
def load_site_dataset_map() -> dict:
    with open("sites_dataset_map.json", 'r', encoding='utf-8') as file:
        return json.load(file)
def apify_crawl(tenant: str, starter_urls: list, hops: int):
    site_dataset_map = load_site_dataset_map()
    if dataset_id := site_dataset_map.get(tenant):
        loader = ApifyDatasetLoader(
            dataset_id=dataset_id,
            dataset_mapping_function=lambda item: Document(
                page_content=item["html"] or "", metadata={"url": item["url"]}
            ),
        )
    else:
        apify = ApifyWrapper()
        loader = apify.call_actor(
            actor_id="apify/website-content-crawler",
            run_input={
                "startUrls": starter_urls,
                "saveHtml": True,
                "htmlTransformer": "none"
            },
            dataset_mapping_function=lambda item: Document(
                page_content=item["html"] or "", metadata={"url": item["url"]}
            ),
        )
        print(f"Site: {tenant} crawled and loaded into Apify dataset: {loader.dataset_id}")
    return loader.load()
# ... existing code
async def index_docs(
    state: IndexState, *, config: Optional[RunnableConfig] = None
) -> dict[str, str]:
    # ... as before
    with retrieval.make_retriever(config) as retriever:
        # code to kick start crawl if required
        configuration = IndexConfiguration.from_runnable_config(config)
        if not state.docs and configuration.starter_urls:
            print(f"starting crawl ...")
            # state.docs = await crawl (
            #     configuration.user_id,
            #     configuration.parse_starter_urls(),
            #     configuration.hops
            # )
            state.docs = apify_crawl (
                configuration.user_id,
                [{"url": url} for url in configuration.parse_starter_urls()],
                configuration.hops
            )
        # rest remains the same as before
        stamped_docs = ensure_docs_have_user_id(state.docs, config)
        if configuration.retriever_provider == "milvus":
            retriever.add_documents(stamped_docs)
        else:
            await retriever.aadd_documents(stamped_docs)
    return {"docs": "delete"}
# ... existing code

I changed default values for parameters saveHtml, and htmlTransformer because, OpenAI embedding models understand HTML very well, and the clean up performed by html transformer loses some useful information.

Handling Edge Cases

Splitting documents for fine-grained retrieval

So far we directly embedded a crawled document. Crawled documents may be arbitrarily large.

1. This puts an uncertainity on how many top_k retrieval results we should consider for response generation.
2. For complex queries, where composing high quality answers may need information from multiple documents, limiting top_k may lead to poor results.

To address this, we split crawl documents into multiple documents as follows:

# src/retrieval_graph/index_graph.py

# new import
from langchain_text_splitters import RecursiveCharacterTextSplitter
# same as before
async def index_docs(
    state: IndexState, *, config: Optional[RunnableConfig] = None
) -> dict[str, str]:
    if not config:
        raise ValueError("Configuration required to run index_docs.")
    with retrieval.make_retriever(config) as retriever:
        configuration = IndexConfiguration.from_runnable_config(config)
        if not state.docs and (configuration.starter_urls or configuration.apify_dataset_id):
            print(f"starting crawl ...")
            crawled_docs = apify_crawl(configuration)
            # use a 1000 char size overlapping window based splitter to create smaller document chunks to index
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
            state.docs = text_splitter.split_documents(crawled_docs)
            
            # same as before

We can now retrieve a larger set of top_k results. Since the maximum chunk size here is 1000, we can deterministically configure the top_k based on the token limits supported by the language model used in the response generation node. E.g., if the token limit is 100,000 tokens, we can potentially use top_k upto 500 (assuming an average token size of 5 characters).

# src/retrieval_graph/retreival.py
# ...
def make_milvus_retriever(
    configuration: IndexConfiguration, embedding_model: Embeddings, **kwargs
) -> Generator[VectorStoreRetriever, None, None]:
    # ...
    yield vstore.as_retriever(search_kwargs=configuration.search_kwargs)
# ...

Here, search_kwargs = {"k": 10} conveys that retriever will return top 10 results.

Rate limits during embedding

Embedding Models put limits on the number of token per unit time. Initially, we embedded the whole set of crawled (or split) documents in a single call. For large sites, this ran into rate limit errors.

With the following enhancement we can add a batch size configuration, split documents into multiple batches, and add a time delay to avoid rate limits.

# src/retrieval_graph/configuration.py
# ...
@dataclass(kw_only=True)
class IndexConfiguration(CommonConfiguration):
    # ...
    batch_size: int = field(
        default=400,
        metadata={
            "description": "Number of documents to index in a single batch."
        },
    )
    # ...

# src/retrieval_graph/index_graph.py
# ...

# generator function to create batches
def create_batches(docs, batch_size):
    """Chunk documents into smaller batches."""
    for i in range(0, len(docs), batch_size):
        yield docs[i:i + batch_size]
async def index_docs(
    state: IndexState, *, config: Optional[RunnableConfig] = None
) -> dict[str, str]:
    if not config:
        raise ValueError("Configuration required to run index_docs.")
    with retrieval.make_retriever(config) as retriever:
        configuration = IndexConfiguration.from_runnable_config(config)
        # ...
        stamped_docs = ensure_docs_have_user_id(state.docs, config)
        # embed in batches to avoid rate limit errors
        batch_size = configuration.batch_size
        for i, batch in enumerate(create_batches(stamped_docs, batch_size)):
            if configuration.retriever_provider == "milvus":
                retriever.add_documents(batch)
            else:
                await retriever.aadd_documents(batch)
            # sleep if there are more batches to embed
            if i < (len(stamped_docs) // batch_size):
                time.sleep(60)
    return {"docs": "delete"}
# ...

Cloud Deployment

We can now deploy the retrieval graph to langgraph cloud. The process is quite straight-forward:

1. Create a Plus account on LangSmith. This is necessary. Free-tier developer account does not allow deploying to LangGraph cloud. This costs 39 USD / month.
2. Now from LangSmith left-nav bar, go to Deployments > LangGraph Platform. Here we can initiate a + New Deployment.
3. We can Import from GitHub my langGraph repo! And add environment configurations (the ones in .env file).
4. And just Submit to deploy! This step takes around 15+ minutes. Once this finishes, click on LangGraph Studio to access the studio to test the graphs!

References

This blog was written in collaboration with Praneet Dhoolia (https://praneetdhoolia.github.io) https://praneetdhoolia.github.io/2024/12/31/expanding-langgraph-retrieval-template.html

Building & Deploying an Agent with LangGraph was originally published in HeurisLabs on Medium, where people are continuing the conversation by highlighting and responding to this story.