LangChain & LlamaIndex

Use Convoy as a cost-efficient LLM backend for LangChain and LlamaIndex applications. This guide shows how to build custom LLM wrappers that route requests through Convoy’s batch processing pipeline — ideal for offline workloads, bulk document processing, and background AI tasks.

Testing tip — Build and verify this integration end-to-end with convoy-mock first. It’s a free, synthetic model that returns a callback in ~60 seconds and is never billed. Once your wiring works, swap model="convoy-mock" for a production model like claude-3-haiku.

Convoy processes requests in batches. Results arrive within minutes to hours, depending on queue volume and batch size (up to 24-hour SLA). This makes Convoy ideal for offline/background LangChain and LlamaIndex workloads — not real-time chat or streaming use cases.

When to Use Convoy with LangChain/LlamaIndex

Use Case	Convoy Fit	Why
Bulk document summarization	✅ Excellent	Process hundreds of docs at batch pricing
Offline RAG indexing	✅ Excellent	Generate embeddings/summaries in background
Content generation pipelines	✅ Excellent	Blog posts, reports, translations
Evaluation/scoring datasets	✅ Excellent	Grade outputs without time pressure
Real-time chat	❌ Not suitable	Latency too high for interactive use
Streaming responses	❌ Not suitable	Convoy returns complete responses
Agent tool calls	⚠️ Limited	Only if agent can wait for async results

LangChain Integration

Custom LLM Class

Create a LangChain-compatible LLM that submits to Convoy and polls for results:


"""convoy_langchain.py — LangChain LLM wrapper for Convoy batch inference."""
 
from typing import Any, List, Optional
 
import httpx
import time
 
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from langchain_core.outputs import Generation, LLMResult
 
 
class ConvoyLLM(LLM):
    """LangChain LLM that routes requests through Convoy's batch API.
 
    Example:
        >>> from convoy_langchain import ConvoyLLM
        >>> llm = ConvoyLLM(api_key="convoy_sk_...", model="claude-3-haiku")
        >>> result = llm.invoke("Explain quantum computing.")
    """
 
    api_key: str
    base_url: str = "https://api.cnvy.ai"
    model: str = "claude-3-haiku"
    max_tokens: int = 1024
    temperature: Optional[float] = None
    system: Optional[str] = None
    poll_interval: float = 10.0
    timeout: float = 3600.0  # 1 hour max wait
 
    @property
    def _llm_type(self) -> str:
        return "convoy"
 
    @property
    def _identifying_params(self) -> dict:
        return {
            "model": self.model,
            "base_url": self.base_url,
            "max_tokens": self.max_tokens,
        }
 
    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        """Submit prompt to Convoy and poll for result."""
        client = httpx.Client(timeout=30.0)
 
        # Build request
        params: dict = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": [{"role": "user", "content": prompt}],
        }
        if self.system:
            params["system"] = self.system
        if self.temperature is not None:
            params["temperature"] = self.temperature
        if stop:
            params["stop_sequences"] = stop
 
        # Submit to Convoy
        response = client.post(
            f"{self.base_url}/cargo/load",
            headers={
                "X-API-Key": self.api_key,
                "Content-Type": "application/json",
            },
            json={"params": params},
        )
        response.raise_for_status()
        cargo_id = response.json()["cargo_id"]
 
        if run_manager:
            run_manager.on_text(f"Submitted to Convoy: {cargo_id}\n")
 
        # Poll for result
        start = time.time()
        while time.time() - start < self.timeout:
            tracking = client.get(
                f"{self.base_url}/cargo/{cargo_id}/tracking",
                headers={"X-API-Key": self.api_key},
            )
            tracking.raise_for_status()
            data = tracking.json()
 
            if data["status"] == "COMPLETED":
                client.close()
                return data["response"]["content"][0]["text"]
            elif data["status"] == "FAILED":
                client.close()
                raise RuntimeError(f"Convoy request failed: {data.get('error')}")
 
            time.sleep(self.poll_interval)
 
        client.close()
        raise TimeoutError(f"Convoy request {cargo_id} timed out after {self.timeout}s")

Basic Usage


from convoy_langchain import ConvoyLLM
 
llm = ConvoyLLM(
    api_key="convoy_sk_your_key_here",
    model="claude-3-sonnet",
    max_tokens=2048,
    system="You are a helpful assistant.",
)
 
# Simple invocation
result = llm.invoke("Summarize the key principles of clean code.")
print(result)

With LangChain Chains


from langchain_core.prompts import PromptTemplate
from convoy_langchain import ConvoyLLM
 
llm = ConvoyLLM(api_key="convoy_sk_your_key_here", model="claude-3-sonnet")
 
# Summarization chain
summarize_prompt = PromptTemplate.from_template(
    "Summarize the following text in 3 bullet points:\n\n{text}"
)
 
chain = summarize_prompt | llm
 
result = chain.invoke({"text": "Your long document text here..."})
print(result)

Batch Processing with LangChain

Process multiple documents efficiently:


from langchain_core.prompts import PromptTemplate
from convoy_langchain import ConvoyLLM
 
llm = ConvoyLLM(
    api_key="convoy_sk_your_key_here",
    model="claude-3-haiku",
    max_tokens=512,
)
 
prompt = PromptTemplate.from_template(
    "Generate a product description for: {product_name}\n"
    "Features: {features}\n"
    "Target audience: {audience}"
)
 
chain = prompt | llm
 
# Batch invoke — each request goes through Convoy's batch pipeline
products = [
    {"product_name": "CloudSync Pro", "features": "Real-time sync, encryption", "audience": "Enterprise IT"},
    {"product_name": "DataVault", "features": "Backup, versioning, compliance", "audience": "Healthcare"},
    {"product_name": "FlowBuilder", "features": "Drag-and-drop, integrations", "audience": "Marketing teams"},
]
 
results = chain.batch(products)
for product, result in zip(products, results):
    print(f"\n{'='*40}")
    print(f"Product: {product['product_name']}")
    print(result)

LangChain’s .batch() method processes items sequentially by default. Each item submits to Convoy and polls independently. For true parallel submission, use the async pattern below.

Async Batch (Parallel Submission)

Submit all prompts at once, then poll for results in parallel:


"""convoy_langchain_async.py — Async LangChain wrapper for parallel batch processing."""
 
import asyncio
from typing import Any, List, Optional
 
import httpx
 
from langchain_core.callbacks import AsyncCallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
 
 
class AsyncConvoyLLM(LLM):
    """Async LangChain LLM for Convoy — enables parallel batch processing."""
 
    api_key: str
    base_url: str = "https://api.cnvy.ai"
    model: str = "claude-3-haiku"
    max_tokens: int = 1024
    temperature: Optional[float] = None
    system: Optional[str] = None
    poll_interval: float = 10.0
    timeout: float = 3600.0
 
    @property
    def _llm_type(self) -> str:
        return "convoy-async"
 
    def _call(self, prompt: str, **kwargs) -> str:
        """Sync fallback — runs async in event loop."""
        return asyncio.run(self._acall(prompt, **kwargs))
 
    async def _acall(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        """Submit to Convoy and poll asynchronously."""
        async with httpx.AsyncClient(timeout=30.0) as client:
            params: dict = {
                "model": self.model,
                "max_tokens": self.max_tokens,
                "messages": [{"role": "user", "content": prompt}],
            }
            if self.system:
                params["system"] = self.system
            if self.temperature is not None:
                params["temperature"] = self.temperature
            if stop:
                params["stop_sequences"] = stop
 
            # Submit
            response = await client.post(
                f"{self.base_url}/cargo/load",
                headers={"X-API-Key": self.api_key, "Content-Type": "application/json"},
                json={"params": params},
            )
            response.raise_for_status()
            cargo_id = response.json()["cargo_id"]
 
            # Poll
            import time
            start = time.time()
            while time.time() - start < self.timeout:
                tracking = await client.get(
                    f"{self.base_url}/cargo/{cargo_id}/tracking",
                    headers={"X-API-Key": self.api_key},
                )
                tracking.raise_for_status()
                data = tracking.json()
 
                if data["status"] == "COMPLETED":
                    return data["response"]["content"][0]["text"]
                elif data["status"] == "FAILED":
                    raise RuntimeError(f"Convoy failed: {data.get('error')}")
 
                await asyncio.sleep(self.poll_interval)
 
            raise TimeoutError(f"Convoy {cargo_id} timed out")

Usage with parallel batch:


import asyncio
from convoy_langchain_async import AsyncConvoyLLM
 
llm = AsyncConvoyLLM(api_key="convoy_sk_your_key_here", model="claude-3-haiku")
 
prompts = [
    "Write a tweet about AI safety",
    "Write a tweet about cloud computing",
    "Write a tweet about developer tools",
    "Write a tweet about open source",
    "Write a tweet about machine learning",
]
 
# All prompts submitted in parallel, polled concurrently
results = asyncio.run(llm.abatch(prompts))
for prompt, result in zip(prompts, results):
    print(f"Prompt: {prompt}")
    print(f"Result: {result}\n")

LlamaIndex Integration

Custom LLM Class


"""convoy_llamaindex.py — LlamaIndex LLM wrapper for Convoy batch inference."""
 
import time
from typing import Any, Optional, Sequence
 
import httpx
 
from llama_index.core.base.llms.types import (
    ChatMessage,
    ChatResponse,
    CompletionResponse,
    LLMMetadata,
    MessageRole,
)
from llama_index.core.llms.custom import CustomLLM
from llama_index.core.llms.callbacks import llm_completion_callback
 
 
class ConvoyLlamaIndex(CustomLLM):
    """LlamaIndex LLM that routes requests through Convoy's batch API.
 
    Example:
        >>> from convoy_llamaindex import ConvoyLlamaIndex
        >>> llm = ConvoyLlamaIndex(api_key="convoy_sk_...", model="claude-3-haiku")
        >>> response = llm.complete("Explain quantum computing.")
    """
 
    api_key: str = ""
    base_url: str = "https://api.cnvy.ai"
    model: str = "claude-3-haiku"
    max_tokens: int = 1024
    temperature: Optional[float] = None
    system_prompt: Optional[str] = None
    poll_interval: float = 10.0
    timeout: float = 3600.0
 
    @property
    def metadata(self) -> LLMMetadata:
        return LLMMetadata(
            model_name=self.model,
            context_window=200000,  # Claude context window
            num_output=self.max_tokens,
            is_chat_model=True,
        )
 
    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
        """Submit a completion request to Convoy."""
        text = self._submit_and_poll(
            messages=[{"role": "user", "content": prompt}]
        )
        return CompletionResponse(text=text)
 
    @llm_completion_callback()
    def chat(self, messages: Sequence[ChatMessage], **kwargs: Any) -> ChatResponse:
        """Submit a chat request to Convoy."""
        formatted_messages = [
            {"role": msg.role.value, "content": msg.content}
            for msg in messages
            if msg.role != MessageRole.SYSTEM
        ]
 
        # Extract system message if present
        system = self.system_prompt
        for msg in messages:
            if msg.role == MessageRole.SYSTEM:
                system = msg.content
                break
 
        text = self._submit_and_poll(messages=formatted_messages, system=system)
        return ChatResponse(
            message=ChatMessage(role=MessageRole.ASSISTANT, content=text)
        )
 
    def stream_complete(self, prompt: str, **kwargs):
        """Convoy doesn't support streaming — falls back to complete."""
        raise NotImplementedError(
            "Convoy uses batch processing and does not support streaming. "
            "Use .complete() or .chat() instead."
        )
 
    def stream_chat(self, messages, **kwargs):
        """Convoy doesn't support streaming — falls back to chat."""
        raise NotImplementedError(
            "Convoy uses batch processing and does not support streaming. "
            "Use .complete() or .chat() instead."
        )
 
    def _submit_and_poll(
        self,
        messages: list[dict],
        system: Optional[str] = None,
    ) -> str:
        """Submit to Convoy and poll for result."""
        client = httpx.Client(timeout=30.0)
 
        params: dict = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": messages,
        }
        if system or self.system_prompt:
            params["system"] = system or self.system_prompt
        if self.temperature is not None:
            params["temperature"] = self.temperature
 
        # Submit
        response = client.post(
            f"{self.base_url}/cargo/load",
            headers={"X-API-Key": self.api_key, "Content-Type": "application/json"},
            json={"params": params},
        )
        response.raise_for_status()
        cargo_id = response.json()["cargo_id"]
 
        # Poll
        start = time.time()
        while time.time() - start < self.timeout:
            tracking = client.get(
                f"{self.base_url}/cargo/{cargo_id}/tracking",
                headers={"X-API-Key": self.api_key},
            )
            tracking.raise_for_status()
            data = tracking.json()
 
            if data["status"] == "COMPLETED":
                client.close()
                return data["response"]["content"][0]["text"]
            elif data["status"] == "FAILED":
                client.close()
                raise RuntimeError(f"Convoy request failed: {data.get('error')}")
 
            time.sleep(self.poll_interval)
 
        client.close()
        raise TimeoutError(f"Convoy request {cargo_id} timed out after {self.timeout}s")

Basic Usage


from convoy_llamaindex import ConvoyLlamaIndex
 
llm = ConvoyLlamaIndex(
    api_key="convoy_sk_your_key_here",
    model="claude-3-sonnet",
    max_tokens=2048,
)
 
# Completion
response = llm.complete("What are the benefits of microservices architecture?")
print(response.text)

Document Summarization Pipeline

Use Convoy with LlamaIndex’s document processing:


from llama_index.core import (
    SimpleDirectoryReader,
    SummaryIndex,
    Settings,
)
from convoy_llamaindex import ConvoyLlamaIndex
 
# Configure Convoy as the LLM
Settings.llm = ConvoyLlamaIndex(
    api_key="convoy_sk_your_key_here",
    model="claude-3-sonnet",
    max_tokens=4096,
    system_prompt="You are a document analyst. Provide clear, structured summaries.",
)
 
# Load documents
documents = SimpleDirectoryReader("./documents").load_data()
 
# Build summary index (uses Convoy for summarization)
index = SummaryIndex.from_documents(documents)
 
# Query the index
query_engine = index.as_query_engine()
response = query_engine.query("What are the main themes across these documents?")
print(response)

Document indexing with Convoy works best for offline batch processing. Each LLM call goes through Convoy’s batch pipeline, so building an index may take longer than with a real-time API — but at significantly lower cost.

RAG with Convoy (Offline Indexing)

Build a RAG pipeline where the indexing phase uses Convoy for cost-efficient processing:


from llama_index.core import (
    SimpleDirectoryReader,
    VectorStoreIndex,
    Settings,
)
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from convoy_llamaindex import ConvoyLlamaIndex
 
# Use local embeddings (fast) + Convoy for generation (cost-efficient)
Settings.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")
Settings.llm = ConvoyLlamaIndex(
    api_key="convoy_sk_your_key_here",
    model="claude-3-sonnet",
    max_tokens=2048,
)
 
# Load and index documents (embeddings are local, fast)
documents = SimpleDirectoryReader("./knowledge_base").load_data()
index = VectorStoreIndex.from_documents(documents)
 
# Query uses Convoy for generation (batch-processed)
query_engine = index.as_query_engine()
response = query_engine.query("How do I configure authentication?")
print(response)

Callback-Based Pattern (No Polling)

For large batch jobs, avoid polling overhead by using callbacks:


"""convoy_batch_pipeline.py — Submit many prompts, receive results via webhook."""
 
import httpx
import json
from pathlib import Path
 
 
class ConvoyBatchPipeline:
    """Submit LangChain/LlamaIndex workloads to Convoy with callback delivery."""
 
    def __init__(self, api_key: str, callback_url: str, base_url: str = "https://api.cnvy.ai"):
        self.api_key = api_key
        self.callback_url = callback_url
        self.base_url = base_url
        self.client = httpx.Client(
            base_url=base_url,
            headers={"X-API-Key": api_key, "Content-Type": "application/json"},
            timeout=30.0,
        )
 
    def submit_batch(
        self,
        prompts: list[str],
        model: str = "claude-3-haiku",
        max_tokens: int = 1024,
        system: str | None = None,
    ) -> list[str]:
        """Submit a batch of prompts. Returns list of cargo_ids."""
        cargo_ids = []
        for i, prompt in enumerate(prompts):
            params = {
                "model": model,
                "max_tokens": max_tokens,
                "messages": [{"role": "user", "content": prompt}],
            }
            if system:
                params["system"] = system
 
            response = self.client.post(
                "/cargo/load",
                json={
                    "params": params,
                    "callback_url": self.callback_url,
                    "metadata": {"batch_index": str(i), "batch_size": str(len(prompts))},
                },
            )
            response.raise_for_status()
            cargo_ids.append(response.json()["cargo_id"])
 
        return cargo_ids
 
    def submit_langchain_batch(
        self,
        chain,
        inputs: list[dict],
        model: str = "claude-3-haiku",
        max_tokens: int = 1024,
    ) -> list[str]:
        """Format LangChain chain inputs and submit as a batch."""
        prompts = [chain.first.format(**inp) for inp in inputs]
        return self.submit_batch(prompts, model=model, max_tokens=max_tokens)

Usage:


from convoy_batch_pipeline import ConvoyBatchPipeline
from langchain_core.prompts import PromptTemplate
 
pipeline = ConvoyBatchPipeline(
    api_key="convoy_sk_your_key_here",
    callback_url="https://your-server.com/webhook/convoy",
)
 
# Define your chain's prompt
prompt = PromptTemplate.from_template(
    "Write a {tone} product review for: {product}"
)
 
# Submit batch — results arrive at your callback URL
inputs = [
    {"tone": "enthusiastic", "product": "Wireless headphones"},
    {"tone": "professional", "product": "Project management software"},
    {"tone": "casual", "product": "Coffee subscription service"},
]
 
cargo_ids = pipeline.submit_langchain_batch(
    chain=prompt | None,  # We only need the prompt formatting
    inputs=inputs,
    model="claude-3-haiku",
)
 
print(f"Submitted {len(cargo_ids)} prompts:")
for cid in cargo_ids:
    print(f"  {cid}")
print("\nResults will arrive at your callback URL.")

LangChain Chat Model (Messages API)

For chains that use chat messages instead of raw prompts:


"""convoy_chat_model.py — LangChain ChatModel wrapper for Convoy."""
 
import time
from typing import Any, List, Optional
 
import httpx
 
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langchain_core.outputs import ChatGeneration, ChatResult
 
 
class ConvoyChatModel(BaseChatModel):
    """LangChain ChatModel that routes through Convoy's batch API."""
 
    api_key: str
    base_url: str = "https://api.cnvy.ai"
    model: str = "claude-3-haiku"
    max_tokens: int = 1024
    temperature: Optional[float] = None
    poll_interval: float = 10.0
    timeout: float = 3600.0
 
    @property
    def _llm_type(self) -> str:
        return "convoy-chat"
 
    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Submit chat messages to Convoy."""
        # Separate system message
        system = None
        chat_messages = []
        for msg in messages:
            if isinstance(msg, SystemMessage):
                system = msg.content
            elif isinstance(msg, HumanMessage):
                chat_messages.append({"role": "user", "content": msg.content})
            elif isinstance(msg, AIMessage):
                chat_messages.append({"role": "assistant", "content": msg.content})
 
        # Submit and poll
        client = httpx.Client(timeout=30.0)
        params: dict = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": chat_messages,
        }
        if system:
            params["system"] = system
        if self.temperature is not None:
            params["temperature"] = self.temperature
        if stop:
            params["stop_sequences"] = stop
 
        response = client.post(
            f"{self.base_url}/cargo/load",
            headers={"X-API-Key": self.api_key, "Content-Type": "application/json"},
            json={"params": params},
        )
        response.raise_for_status()
        cargo_id = response.json()["cargo_id"]
 
        # Poll for result
        start = time.time()
        while time.time() - start < self.timeout:
            tracking = client.get(
                f"{self.base_url}/cargo/{cargo_id}/tracking",
                headers={"X-API-Key": self.api_key},
            )
            tracking.raise_for_status()
            data = tracking.json()
 
            if data["status"] == "COMPLETED":
                text = data["response"]["content"][0]["text"]
                usage = data["response"].get("usage", {})
                client.close()
                return ChatResult(
                    generations=[
                        ChatGeneration(
                            message=AIMessage(content=text),
                            generation_info={
                                "cargo_id": cargo_id,
                                "model": data["response"].get("model"),
                                "input_tokens": usage.get("input_tokens"),
                                "output_tokens": usage.get("output_tokens"),
                            },
                        )
                    ]
                )
            elif data["status"] == "FAILED":
                client.close()
                raise RuntimeError(f"Convoy failed: {data.get('error')}")
 
            time.sleep(self.poll_interval)
 
        client.close()
        raise TimeoutError(f"Convoy {cargo_id} timed out")

Usage with LCEL (LangChain Expression Language):


from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from convoy_chat_model import ConvoyChatModel
 
chat = ConvoyChatModel(
    api_key="convoy_sk_your_key_here",
    model="claude-3-sonnet",
    temperature=0.7,
)
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a creative writing assistant."),
    ("human", "Write a {length} story about {topic}."),
])
 
chain = prompt | chat | StrOutputParser()
 
result = chain.invoke({"length": "short", "topic": "a robot learning to paint"})
print(result)

Installation


# For LangChain
pip install langchain-core httpx
 
# For LlamaIndex
pip install llama-index-core httpx
 
# For async patterns
pip install httpx[http2]

Configuration

Environment Variables


export CONVOY_API_KEY=convoy_sk_your_key_here
export CONVOY_BASE_URL=https://api.cnvy.ai  # or http://localhost:8000 for local dev

Using with `.env` files


import os
from dotenv import load_dotenv
from convoy_langchain import ConvoyLLM
 
load_dotenv()
 
llm = ConvoyLLM(
    api_key=os.environ["CONVOY_API_KEY"],
    base_url=os.environ.get("CONVOY_BASE_URL", "https://api.cnvy.ai"),
    model="claude-3-sonnet",
)

Best Practices

1. Use Callbacks for Large Batches

For more than 10 prompts, use the callback pattern instead of polling:


# ❌ Polling — blocks your process for each request
results = chain.batch(large_list_of_inputs)  # Could take hours
 
# ✅ Callback — submit all, receive results asynchronously
pipeline = ConvoyBatchPipeline(api_key=key, callback_url=webhook_url)
cargo_ids = pipeline.submit_batch(prompts)
# Results arrive at your webhook — process them there

2. Respect Rate Limits


import time
 
# Add delays between submissions
for prompt in prompts:
    cargo_id = submit(prompt)
    time.sleep(1.0)  # 60 req/min = 1 per second max

3. Handle Timeouts Gracefully


from convoy_langchain import ConvoyLLM
 
llm = ConvoyLLM(
    api_key="convoy_sk_your_key_here",
    timeout=7200.0,  # 2 hours for large batches
    poll_interval=30.0,  # Poll less frequently to reduce API calls
)

4. Use Appropriate Models

Workload	Recommended Model	Why
Bulk content generation	`claude-3-haiku`	Fast, cheap, good quality
Document analysis	`claude-3-sonnet`	Better reasoning
Complex summarization	`claude-3-sonnet`	Nuanced understanding
Simple classification	`claude-3-haiku`	Cost-efficient for simple tasks

Troubleshooting

`TimeoutError: Convoy request timed out`

Increase the timeout parameter (default is 1 hour)
Check the cargo status manually: GET /cargo/{cargo_id}/tracking
Convoy processes in batches — large queues may take longer

`RuntimeError: Convoy request failed`

Check the error message in the tracking response
Verify the model ID is valid (use GET /models)
Ensure max_tokens doesn’t exceed the model’s limit

`httpx.HTTPStatusError: 429`

You’ve hit the rate limit (60 requests/minute)
Add delays between batch submissions
Use the callback pattern to avoid polling overhead

LlamaIndex `NotImplementedError: streaming`

Convoy doesn’t support streaming — this is expected
Use query_engine.query() instead of query_engine.stream_query()
Set streaming=False in your LlamaIndex settings

Slow indexing with LlamaIndex

Each chunk requires a Convoy API call — this is by design for batch pricing
For faster indexing, use a real-time LLM for indexing and Convoy for queries
Or submit all chunks via the callback pattern and build the index after all results arrive

Next Steps

Python guide — Full Convoy client class with more patterns
Webhooks guide — Set up callback receivers for batch results
AWS Lambda guide — Deploy your LangChain pipeline serverlessly

LangChain & LlamaIndex

When to Use Convoy with LangChain/LlamaIndex

LangChain Integration

Custom LLM Class

Basic Usage

With LangChain Chains

Batch Processing with LangChain

Async Batch (Parallel Submission)

LlamaIndex Integration

Custom LLM Class

Basic Usage

Document Summarization Pipeline

RAG with Convoy (Offline Indexing)

Callback-Based Pattern (No Polling)

LangChain Chat Model (Messages API)

Installation

Configuration

Environment Variables

Using with .env files

Best Practices

1. Use Callbacks for Large Batches

2. Respect Rate Limits

3. Handle Timeouts Gracefully

4. Use Appropriate Models

Troubleshooting

TimeoutError: Convoy request timed out

RuntimeError: Convoy request failed

httpx.HTTPStatusError: 429

LlamaIndex NotImplementedError: streaming

Slow indexing with LlamaIndex

Next Steps

Using with `.env` files

`TimeoutError: Convoy request timed out`

`RuntimeError: Convoy request failed`

`httpx.HTTPStatusError: 429`

LlamaIndex `NotImplementedError: streaming`