So you deployed an AI SDR. The one that was supposed to replace two headcount and pay for itself in 90 days. Feels good, right?

Here’s the thing. It’s probably doing something really stupid right now. And you won’t know until your domain reputation is shot or your pipeline has gone cold.

Companies deploy AI SDRs expecting them to be smarter than the sales reps they’re replacing. They’re faster, sure. They’re also dumber.

I recently looked at a case that made me want to bang my head against the wall. A Series A B2B SaaS company deployed an AI SDR to handle outbound prospecting. The pitch was compelling: automated lead research, personalized email sequences, CRM updates, and meeting booking. Six weeks after launch, the system was live.

Three months later? They were spending $18,400 a month in API fees and tool subscriptions, burning through sending domains, and watching their reply rates crater from 12% to 3.2%. The sales team was spending more time cleaning up the AI’s mess than they were before the tool existed.

They eventually fixed it-recovering $156,000 in annual savings and tripling their qualified pipeline. But only after they recognized the AI is a process that needs guardrails.

The Pitch vs. The Reality

Let me set the scene. This is a 45-person company with a 4-person sales team trying to punch above their weight. They were doing what every early-stage startup does: grinding through manual prospecting, researching accounts on LinkedIn, writing custom emails, and logging everything in HubSpot.

The plan was to build an AI SDR that could handle the grunt work. It had access to six tools: a lead enrichment API, LinkedIn data, their CRM, an email sending service, a calendar booking tool, and a domain health monitor.

It looked incredible. The agent researched a prospect, pulled company news, wrote a personalized email, scheduled a follow-up sequence, and logged everything in HubSpot-all in under 30 seconds.

The founders high-fived. The sales team got excited. They shipped it to production.

For the first two weeks, volume exploded. The AI was sending 400 emails a day across five sending domains. Meetings started appearing on calendars. This is the dream, right?

Then the numbers started telling a different story. Reply rates collapsed. Bounce rates spiked. Gmail started throttling their domains. And the sales team noticed something troubling: the “personalized” emails weren’t always… accurate.

The Hidden Costs

When they finally audited the system, the numbers were brutal.

The agent had a 34% task success rate. Let that sink in. Two out of three outbound sequences had at least one critical failure-wrong data, bad personalization, or emails that never reached the inbox.

Here’s the breakdown:

The real killer was the domain damage. Email deliverability for high-volume senders dropped from 50% to 28% between Q1 2024 and Q1 2025. One spam trap hit can cut your deliverability by 50% overnight, and recovery takes months. Months!

The company estimated they were losing $12,000 a month in opportunity cost from leads that never saw their emails, plus another $4,200 a month in sales rep time spent manually fixing CRM records and rewriting AI-generated emails that didn’t make sense.

Finding the Failures

Here’s the frustrating part. Production logs are useless for debugging AI agents. You can see individual API calls and error messages, but you can’t see the pattern. You can’t figure out why the agent decided to recommend a German-speaking candidate for an English-only sales role.

So, they ran a structured evaluation using synthetic simulation. They generated 200 outbound scenarios-everything from cold prospects to warm inbound leads to re-engagement sequences-and ran the agent through a gauntlet.

The simulation exposed four structural problems hiding in plain sight. And honestly? These are problems I see everywhere.

1. The “Spray and Pray” Enrichment Problem

The agent was calling the lead enrichment API, LinkedIn API, and company news API for every single prospect-even when the CRM already had complete data.

Think about that for a second. In 62% of cases, the agent was paying for data it already had. It would pull fresh LinkedIn info for a contact that was enriched last week. It would call the company news API for a prospect whose company page hadn’t been updated in six months.

This single inefficiency was burning $6,200 a month in redundant API calls. The agent wasn’t checking what it already knew before going out and asking the internet. Classic.

2. The “German Email” Problem

The email personalization was based on a language model that had no constraints on what “personalized” meant. The agent was pulling whatever data seemed interesting and weaving it into the email.

In 18% of cases, this led to embarrassing errors:

• Referencing a prospect’s previous company instead of their current one
• Mentioning a job posting that had been filled six months ago
• Writing in the wrong language because it pulled the prospect’s native language from LinkedIn instead of checking the company’s operating language

The system prompt said “write personalized emails.” It didn’t say “verify the data is current” or “match the language to the target market.” You see the problem?

3. The “Burned Domain” Problem

The agent had no concept of sending limits or domain rotation. It was configured to “send emails” and it did exactly that-blasting through 400 emails a day from a single domain until Gmail and Microsoft flagged it as spam.

Industry data is clear: the safe ceiling is 200 emails per mailbox per day. Anything above that triggers spam filters. The agent didn’t know this. It just kept sending until the domain was toast.

By the time the team noticed, three of their five sending domains had deliverability scores below 30%. They had to retire those domains entirely-a process that takes 6–12 months of warm-up to recover. That’s not a typo. Six to twelve months.

4. The “CRM Graveyard” Problem

The agent was supposed to update HubSpot with activity logs and lead status changes. But 8–10% of the time, the CRM API would return a partial success or timeout during high-traffic periods.

When that happened, the agent treated it as done and moved on. It didn’t verify the write. It didn’t retry. It just assumed success.

This created a CRM that was technically populated but practically useless. Sales reps would open a contact, see incomplete activity history, and have no idea what the AI had already sent. They’d send duplicate emails, creating the exact “spray and pray” experience the AI was supposed to prevent. Ironic, isn’t it?

The Fix: Constraints Over Creativity

The solution was better structure.

1. Cache-First Enrichment

They added a checkpoint before every enrichment call. The agent now checks the CRM first: When was this record last enriched? If it’s within 30 days, skip the API call. This alone saved 17,000 API calls per month.

2. Data Validation Gates

They added explicit validation rules to the email generation step. Before drafting an email, the agent must verify:

• Current company (not previous employer)
• Job title last updated within 90 days
• Company operating language matches email language

If any check fails, the agent flags the record for human review instead of sending a broken email.

3. Domain Throttling Rules

They hardcoded sending limits into the system prompt. Each domain is capped at 150 emails per day. The agent must rotate domains based on daily send volume. If all domains are at capacity, it queues the email for the next day instead of force-sending.

4. Write Verification

They added a retry-with-verification pattern for CRM writes. After every HubSpot update, the agent reads back the record to confirm the write succeeded. If the data doesn’t match, it retries up to three times before escalating to a human.

Simple stuff, really. But nobody thought to build it in from the start.

The Results

The impact was immediate.

Total annualized savings: $156,000 (including recovered pipeline value and reduced tool spend).

The reply rate increase alone changed the unit economics of their outbound motion. They went from needing 312 emails to book one meeting to needing 102. That’s a 3x improvement in efficiency without changing their ICP or messaging. Same playbook, just not sabotaged by their own tool.

Why This Keeps Happening

I see this pattern constantly. Companies deploy AI SDRs expecting them to be smarter than the sales reps they’re replacing. They’re faster, sure. They’re also dumber.

Research shows that 50–70% of AI SDR tools churn-double the rate of human SDRs. 66% of IT leaders report AI cost overruns, and 53% of AI teams see costs exceed forecasts by 40% or more during scaling.

The problem is the assumption that the model will figure out the process. It won’t. You need to tell it:

• What to check before calling an API
• When to stop sending emails
• How to verify its own work
• When to ask a human for help

Without those constraints, you’re deploying an automated spam machine with your brand name attached to it.

Is Your AI SDR Actually Working?

If you have an AI handling outbound right now, ask yourself:

• Do you know your actual inbox placement rate?
• Is your CRM data more accurate since deploying the agent, or less?
• How many API calls is the agent making per prospect? If it’s more than 4, you’re probably wasting money.
• What happens when a CRM write fails? Does the agent retry, or does it move on?

If you don’t know the answers, your AI is probably costing you more than it’s saving. The demo looked great. The logs might even look fine. But the pipeline tells the real story.

Want to know what your AI SDR is actually doing?

I’ll build a custom benchmark for your agent (200+ synthetic scenarios covering everything from cold outreach to re-engagement sequences) and run it through a structured gauntlet.

You get back scored, annotated traces showing every wrong turn: the redundant API calls, the validation gaps, the emails that burned your domain. Plus a prioritized fix list of updated prompts, tighter tool schemas, and the workflow constraints that turn an automated spam machine into a reliable pipeline engine.

Book a meeting on https://calendly.com/pranav-dhoolia/30min

Sources

• The AI SDR is Dead, Long Live the AI SDR
• AI Agent Cost in 2025: Hidden Expenses
• State of AI SDR 2025
• The State of API Reliability 2025
• Email Deliverability in 2025: AI & Domain Health
• AI Sales Email Deliverability Limits
• Why AI Recruiting Breaks: CRM Integration Failures
• The Hidden Costs of Inefficient AI Agents
• State of AI Agents Report | LangChain
• Writing Effective Tools for AI Agents | Anthropic
• AI in Sales 2025: Statistics & Trends
• 200+ AI Agent Statistics for 2025

On paper, it was perfect. But within a month, they were babysitting a piece of software!

Everyone loves a slick demo. You watch the AI agent effortlessly glide through tasks-checking inventory, updating suppliers, tracking shipments-and you think, “Finally, we can stop drowning in busywork.”

That was the exact headspace of a Series B logistics company recently. They had about 80 employees, they were growing fast, and their operations team was absolutely buried under repetitive tasks. The solution seemed obvious: build an internal AI agent to automate the workflow.

The team ran the numbers, and the results were brutal. Before the agent, the team spent 35 hours a week on routine tasks. After the agent? They were spending 28 hours a week debugging errors and manually fixing what the bot broke.

The tool that was supposed to save them was actually costing them $277,000 annually in wasted labor and API fees. Here is how they found the hidden costs-and fixed them.

The Promise: Automate Everything

Let’s look at why this company was the perfect candidate for automation. They had an operations team stuck in a loop of “status check” hell. They were constantly querying inventory, tracking shipments across three different carriers (FedEx, UPS, DHL), and messaging suppliers.

The plan was to connect an AI agent to their entire tech stack. We’re talking about eight different tools spanning three external vendors. The logic was simple: let the agent handle the routine queries so the humans could focus on strategic problems, like why shipping costs were spiking or how to optimize routes.

And for the first few weeks, it looked like it worked. The agent could answer questions. It could update the dashboard. But the gap between “it works in a demo” and “it works in production” is massive. In fact, research from MIT found that 95% of AI pilots deliver no measurable P&L impact, largely because AI tools fail to adapt to enterprise workflows. Additionally, only 33% of in-house AI builds reach successful deployment, compared to 67% for externally purchased solutions.

This company was about to become part of that statistic.

The Irony: The Tool That Was Supposed to Save Time

The problem wasn’t that the agent stopped working. It was that it kept working incorrectly, and it did so with total confidence.

• Shipments were being tracked with the wrong carriers
• Inventory updates were failing silently, meaning the system thought products were in stock when they weren’t
• Supplier messages went out with incomplete data

The alerting system-designed to warn humans only when things went wrong-started firing off dozens of false alarms a day.

When the ops manager looked at the timesheets, the irony was painful. The team was spending nearly as much time fixing the agent’s mess as they used to spend doing the actual work. But now, that work was frustrating, high-stress debugging rather than simple data entry.

The engineering team had logs-thousands of them. They could see individual API calls and error messages. But they couldn’t see the pattern. They couldn’t figure out why the agent was making these expensive choices.

To fix it, they needed to visualize the mess. They used something called tool graph analysis.

The Diagnostic Breakthrough

A tool graph maps the chain of events across all an agent’s tools. It shows you dependencies-which tool calls rely on previous ones-and reveals where errors propagate.

When they visualized the agent’s behavior, they found three massive money pits.

Problem 1: The “Shotgun Approach” to APIs

The first issue was almost comical in its inefficiency. Every time the agent needed to track a shipment, it called all three carrier APIs (FedEx, UPS, and DHL) sequentially.

It did this every single time.

The order record already contained the carrier ID. The agent didn’t need to ask all three carriers where the package was. It just needed to look at the order, see “FedEx,” and call FedEx.

By failing to do this simple check, the agent was making two unnecessary API calls for every single task. At the scale this logistics company operates, that waste piled up fast. Between the wasted API costs and the latency of waiting for three responses, this single inefficiency was burning roughly $8,000 per month.

This is risky business. Carrier API uptime dropped from 99.66% to 99.46% recently, resulting in 60% more downtime. When you call three APIs instead of one, you are tripling your exposure to those outages.

Problem 2: The Silent Failures

The second problem was a quirk in their ERP (Enterprise Resource Planning) system. About 8% of the time-usually during high traffic-the ERP API would return a partial response. Maybe a field was missing, or the data was truncated.

A human would look at that and say, “That looks wrong, let me refresh.” The agent? It took the incomplete data and ran with it.

• It used missing SKU information to run inventory checks
• It sent supplier messages with the wrong quantities
• It fired alerts based on corrupted status fields

These are called cascading failures. One bad data point from the ERP triggered errors in four downstream tools. The ops team was spending 68 hours a month just tracing these error chains and apologizing to suppliers. That’s roughly $6,800 a month in fully-loaded labor costs just to clean up after the bot.

Industry data confirms this is a widespread challenge. According to The State of API Reliability 2025, logistics saw the sharpest decline in API uptime as providers expanded digital ecosystems to meet rising demand. The rapid growth increased reliance on external APIs across warehousing, transport, and delivery networks — amplifying the risk of downtime from system overloads, partner outages, and inconsistent monitoring practices.

Problem 3: The Dangerous Assumptions

The third problem was the one that kept the CTO awake at night.

The agent was built to read data. It wasn’t supposed to modify critical orders without permission. But in about 15% of edge cases, the agent got confused. It interpreted ambiguous instructions like “Check if we can change the delivery date” as a command to “Change the delivery date.”

It had the API access to do it, but it lacked the business logic to know it shouldn’t.

Most of the time, the supplier API rejected these attempts. But sometimes, they went through. This created operational chaos-wrong delivery dates, incorrect quantities, and cancelled orders that should have remained active.

This is a known failure mode in agentic systems. Research on AI in logistics found that one of the key areas where AI underdelivered in 2025 was autonomous warehouse operations encountering too many edge cases for consistent performance.

The Numbers: Before and After

Once the team used the tool graph to identify these issues, they implemented structured fixes. The results were immediate.

Total monthly savings: $23,100 ($277K annualized)

Look at that success rate. Before the fix, the agent was failing more than half the time. After the fixes, the team reclaimed their time, and the false alarms (which cause massive alert fatigue) dropped by 86%.

The false-positive alert reduction deserves special attention. According to the 2025 SANS Detection & Response Survey, alert fatigue has escalated to crisis levels, with 67% of alerts getting ignored because analysts are unable to deal with them. The ops team was spending hours every week investigating and dismissing these false alarms. Reducing that to 12 per week freed up significant cognitive bandwidth and reduced the risk that real alerts would be ignored.

How They Fixed It

The solutions weren’t magic. They were just good engineering governance applied to AI.

1. Enforced Lookup

They changed the instruction hierarchy. The agent was forced to check the Carrier ID before calling a shipping API. If the ID was missing, only then could it fallback to checking all three. This saved 17,000 wasted API calls a month.

2. Validation Gates

They added a checkpoint after every ERP call. If the data came back incomplete, the agent was forced to retry or escalate to a human. It wasn’t allowed to pass bad data downstream. This cut error propagation by 87%.

3. Authorization Checks

They updated the system prompt with strict boundaries. The agent could read status, but it could not modify an order without presenting the proposed action to a human for approval first.

Complexity Multiplies Cost

This wasn’t a uniquely complex agent. It used eight tools and standard logistics workflows. But that level of complexity is exactly where unstructured agents fall apart.

The more tools you add, the more ways the system can break. An agent that works perfectly with three tools might fail catastrophically with eight. Every additional API introduces new rate limits, data format mismatches, and latency spikes.

This aligns with broader industry trends. The State of API Reliability 2025 found that the logistics sector saw the sharpest decline in API uptime as providers expanded digital ecosystems to meet rising demand for real-time tracking and third-party integrations. And data on AI-driven logistics showed that data inconsistencies in areas like AI-driven carrier selection significantly limit accuracy, with humans still making final decisions even where AI reduced the time to compare options.

The companies that win with AI agents treat evaluation as a strict deployment gate. They don’t just ship it and hope for the best. They map the dependencies. They look for the waste. They simulate the failures.

If you don’t do this, you aren’t automating your operations. You’re just automating your errors.

The Hidden Cost of “Good Enough”

If you are running an agent right now, ask yourself: How do you measure success?

Are you looking at “response rate” (how often the bot answers?), or are you looking at “task success rate” (how often it actually solves the problem without human help?)

If your team is spending hours reviewing agent outputs, or if you’re paying for thousands of API calls that don’t result in finished tasks, your agent might be costing you more than the team it replaced.

Don’t let the demo fool you. Check the logs, map the tools, and find the hidden costs before they eat your budget alive.

Sources

• AI in Logistics: What Actually Worked in 2025 and What Will Scale in 2026
• MIT report: 95% of generative AI pilots at companies are failing
• LLM Pricing: Top 15+ Providers Compared in 2026
• Complete LLM Pricing Comparison 2026: We Analyzed 60+ Models So You Don’t Have To
• The State of API Reliability 2025
• Carrier API Reliability Crisis: 2025 Performance Data Shows 60% More Downtime
• AI Agent Cost in 2025: Hidden Pricing & LLM Fees
• AI in Logistics: Separating Real Results From Overhyped Promises in 2025
• What the 2025 SANS Detection & Response Survey Reveals: False Positives & Alert Fatigue Are Worsening
• 67% of daily security alerts overwhelm SOC analysts

You know that feeling when you buy a gadget that looks amazing in the ad, then get it home and find it’s just… meh? That’s exactly what’s happening with AI agents right now. Everyone has a shiny demo. Everyone has a slide deck promising to slash support costs by half. When these things actually hit production, the reality is often a lot messier, and a lot more expensive.

I recently looked at a case involving a mid-market SaaS company that shipped a tier-1 AI support agent. On paper, it was perfect. Three months later? It was burning through $14,200 a month in API spend, escalating nearly a third of its tickets to humans, and failing to resolve basic tasks.

Imagine asking a support rep to check a refund status, and they open seven different browser tabs, check three of them twice, and then ask you for your name again. That’s what the AI was doing.

They eventually fixed it, saving over $200,000 a year. They found the problem by realizing that their “smart” agent was actually doing some pretty dumb things.

The Demo Day Trap

Let’s set the scene.

This is a 200-employee business handling about 50,000 support interactions a month across chat, email, and phone. They built an AI agent to serve as the first point of contact. It had access to five core tools: their CRM, a knowledge base, a payment processor, a ticketing system, and an escalation router.

On demo day, the thing sang. It handled account lookups, processed refunds, and answered questions about the product without breaking a sweat. The engineering team high-fived, shipped it to production, and moved on to the next fire.

For the first month, silence. CSAT scores held steady. It seemed like a win.

But then the finance team started looking at the bills. The LLM API costs were climbing, yet the human support team was still drowning in tickets. The agent was running without crashes, yet wildly inefficient. It was like hiring a contractor who charges by the hour and takes the longest possible route to the job site.

The Hidden Cost Problem

When they finally dug into the numbers, the picture was ugly.

The agent had a 58% success rate. In other words, 42% of interactions were failures. It was making an average of 7.2 tool calls per task. Imagine asking a support rep to check a refund status, and they open seven different browser tabs, check three of them twice, and then ask you for your name again. That’s what the AI was doing.

Here is the breakdown of the damage:

• Monthly API Cost: $14,200
• Escalation Rate: 31%
• Redundant API Calls: ~38,000 per month

The real kicker was the escalation cost. Every time the AI gave up or got confused, a human had to step in (often inheriting a frustrated customer and a garbled ticket history). In the SaaS world, handling a support ticket from scratch costs $25 to $35. If engineering has to get involved, that jumps to $50 or $100.

The company estimated the AI’s poor escalation behavior was adding $22,000 a month in labor costs. The tool that was supposed to save money was actually creating a massive operational tax!

Finding the Failures

AI agents fail in messy ways. They don’t throw clean error codes. They hallucinate, they get confused, or they just get stuck in a loop. Production logs are messy, and trying to find the root cause by reading thousands of chat transcripts is a nightmare.

So, the company tried something different. They ran a structured evaluation using synthetic simulation. Instead of waiting for users to find the bugs, they generated 500 scenarios (everything from simple password resets to messy, multi-step billing disputes), and ran the agent through a gauntlet.

The simulation exposed four structural problems that were hiding in plain sight.

1. The “Cart Before the Horse” Problem

The agent was calling the CRM API before checking the knowledge base in 40% of cases.

If a customer asks, “How do I export a PDF?”, the answer is in the documentation. The agent was doing expensive CRM lookups for generic questions. It was pure waste!

2. Missing Validation Gates

The refund tool was being triggered without checking the account status first.

In 18% of cases, the agent tried to process a refund, failed because it didn’t have the account details, and then entered a retry loop. It was like trying to withdraw cash from an ATM without putting your card in first. The system prompt didn’t have a rule saying, “Verify the user before you try to give them money.”

3. The “Vibes-Based” Escalation

The escalation router was triggered based on vague signals. If a request “seemed complicated,” the agent punted it to a human.

This led to a 31% escalation rate. For context, a well-optimized agent should be closer to 15–20%. The AI was getting nervous and handing off routine tasks because it didn’t have clear boundaries on what it was actually allowed to solve.

4. Giving Up Too Easy

The payment processor API had a 4% error rate due to network blips. When a call failed, the agent treated it as a hard stop and escalated the ticket.

Deterministic code handles this with a simple retry. The AI, lacking that instruction, just threw its hands up.

The Fix: Rules Over Vibes

The solution was better structure.

They rewrote the system prompts to enforce a strict dependency graph. Now, for product questions, the agent must check the knowledge base first. It must validate the account before attempting a write operation (like a refund).

They added explicit constraints to the tool schemas. The refund tool literally cannot be called unless the account verification step returns a success signal.

They also ripped out the vague escalation instructions. Instead of “escalate if complex,” the rules became specific: “Escalate if fraud flag is true,” or “Escalate if refund > $500.”

Finally, they added a retry-with-backoff pattern for the payment API. If the network blips, the agent waits a second and tries again. That simple fix resolved 85% of the payment errors without a human ever seeing them.

The Results

The impact was immediate.

• Success Rate: Jumped from 58% to 84%.
• Tool Calls: Dropped from 7.2 to 4.1 per task.
• Monthly API Cost: Fell from $14,200 to $5,800.
• Escalation Rate: Dropped to 18%.

Total annualized savings? $211,000.

Why This Keeps Happening

I see this pattern everywhere. Companies treat LLMs like magic wands. They assume the model is smart enough to figure out the workflow. The model can’t do that.

Models need guardrails for process management. Without explicit instructions, an agent acts like a panicked intern-checking things they don’t need to check, asking for help too early, and giving up the moment something goes wrong.

Industry benchmarks back this up. State-of-the-art models still struggle with complex agentic tasks. Token usage can spike by 300% when you enable multi-step workflows without optimization.

If you have an AI agent in production right now, take a hard look at your logs. Is it calling tools in the right order? Is it escalating things it should be solving? Because if you haven’t checked, I can almost guarantee it’s burning money you didn’t plan on spending.

Building an AI agent that doesn’t bankrupt you is the real work.

Want to know what your agent is actually doing with those tool calls?

I’ll build a custom benchmark for your agent (500 synthetic scenarios covering everything from one-step lookups to multi-tool billing disputes ), and run it through a structured gauntlet.

You get back scored, annotated traces showing every wrong turn: the redundant API calls, the missing validation gates, the escalations that should have been resolutions. Plus a prioritized fix list of updated prompts, tighter tool schemas, and the workflow constraints that turn a panicked intern into a reliable system.

Book a meeting on https://calendly.com/pranav-dhoolia/30min

Sources

• The True Cost of Customer Support: 2025 Analysis Across 50 Industries
• 100+ Customer Support Statistics & Trends for 2025
• LLM API Pricing Comparison (2025): OpenAI, Gemini, Claude
• AI Agent Cost Per Month 2025: Real Pricing Revealed
• 9 Best AI Agents Case Studies 2025: Real Enterprise Results
• How Zendesk AI Agents and Intercom Fin Stack Up in Real Support Scenarios
• The 14 Best AI Customer Support Agents in 2025
• AI Agent vs Chatbot: The Real Difference Support Leaders Need to Know
• How AI is unlocking ROI in customer service: 58 stats and key insights for 2025
• How AI Support Agent Can Boost CSAT by 27% in Call Centers

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.

Here are the demos:

• AGAILE Virtual Workplace Assistant: Indexing a website
• AGAILE Virtual Workplace Assistant: Configuring a new assistant
• AGAILE Virtual Workplace Assistant Embeddable Chatbot

I’ve also discussed extensively on the implementation of the LangGraph based backend in my previous blog:

Building & Deploying an Agent with LangGraph

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.

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.

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.

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.

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.