Introduction to Multi-Agent Research Pipelines with vLLM
Modern research workflows increasingly rely on large language models to synthesize information, generate hypotheses, and produce reports. A single LLM call, however, is rarely sufficient for complex research tasks. Multi-agent pipelines break a large problem into specialized roles — planner, searcher, reader, critic, writer — each powered by a model that can be tuned, prompted, and scaled independently. vLLM, with its PagedAttention-based inference engine, is uniquely suited as the backend for such pipelines because it delivers high throughput, low latency, and native support for concurrent requests.
This guide walks through the architecture, implementation, and operational best practices for building a production-grade multi-agent research pipeline on top of vLLM. By the end, you will have a working system where several agents collaborate asynchronously to research a topic, critique findings, and produce a polished report.
What Is a Multi-Agent Research Pipeline?
A multi-agent research pipeline is a coordinated system in which multiple LLM-backed agents, each with a distinct role, exchange structured messages to accomplish a research objective. Rather than asking one model to "research X and write a report," you decompose the task:
- Planner Agent — decomposes the research question into sub-questions.
- Searcher Agent — formulates queries and retrieves documents from a corpus or web API.
- Reader Agent — extracts key facts and citations from retrieved passages.
- Critic Agent — evaluates evidence quality, flags contradictions, and requests follow-ups.
- Writer Agent — synthesizes verified findings into a final report.
Each agent is essentially a prompt template plus a call to a vLLM-served model. The orchestration layer routes messages between agents, enforces schemas, and handles retries.
Why vLLM Is the Right Backend
Multi-agent pipelines are bursty and concurrent: a planner may emit five sub-questions, each spawning a searcher, each producing multiple reader calls. vLLM addresses the resulting load profile through several features:
- Continuous batching — new requests are admitted into running batches without waiting for prior requests to finish, keeping GPU utilization high.
- PagedAttention — KV cache is managed in fixed-size pages, eliminating memory fragmentation and allowing many concurrent agents to share GPU memory efficiently.
- Tensor parallelism — large models can be sharded across GPUs, enabling agents to use 70B-class models with acceptable latency.
- OpenAI-compatible API — agents can use the standard
openaiPython client, simplifying orchestration code. - Guided decoding — vLLM supports JSON-schema-constrained generation via outlines/lm-format-enforcer, critical for enforcing structured inter-agent messages.
Architecture Overview
The pipeline consists of three layers:
- Inference layer — one or more vLLM server instances exposing an OpenAI-compatible endpoint.
- Agent layer — Python classes that wrap prompts, parse outputs, and expose a simple
run()interface. - Orchestration layer — an async event loop that routes messages, manages state, and persists intermediate artifacts.
Agents communicate via a shared message bus backed by asyncio.Queue for single-process deployments or Redis Streams for distributed setups. Each message is a JSON object with from, to, type, and payload fields.
Setting Up the vLLM Server
Install vLLM in an isolated environment. vLLM pins specific CUDA and PyTorch versions, so a dedicated virtualenv or container is recommended.
pip install vllm>=0.6.0
pip install openai>=1.40.0 pydantic>=2.0 aiohttpLaunch the server with a model suitable for agentic reasoning. For research pipelines, a 7B–14B instruct model is a good starting point for development; scale to 70B for production quality.
vllm serve meta-llama/Meta-Llama-3.1-8B-Instruct \
--port 8000 \
--max-model-len 16384 \
--gpu-memory-utilization 0.9 \
--enable-auto-tool-choice \
--guided-decoding-backend outlinesKey flags explained:
--max-model-len 16384— accommodates long retrieved passages and multi-agent context.--gpu-memory-utilization 0.9— reserves headroom for the OS and orchestration process.--enable-auto-tool-choice— enables tool/function calling for agents that need to invoke search APIs.--guided-decoding-backend outlines— enables JSON-schema-constrained generation for structured agent outputs.
Verify the server is healthy:
curl http://localhost:8000/v1/modelsDefining the Agent Base Class
All agents share common behavior: formatting prompts, calling vLLM, parsing structured output, and emitting messages. We define an abstract base class.
from __future__ import annotations
import asyncio
import json
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional
from openai import AsyncOpenAI
@dataclass
class Message:
sender: str
recipient: str
msg_type: str
payload: dict[str, Any]
@dataclass
class AgentConfig:
name: str
model: str
temperature: float = 0.3
max_tokens: int = 2048
system_prompt: str = ""
class BaseAgent(ABC):
def __init__(self, config: AgentConfig, client: AsyncOpenAI):
self.config = config
self.client = client
async def generate(
self,
user_prompt: str,
response_format: Optional[dict] = None,
) -> str:
messages = [
{"role": "system", "content": self.config.system_prompt},
{"role": "user", "content": user_prompt},
]
kwargs: dict[str, Any] = {
"model": self.config.model,
"messages": messages,
"temperature": self.config.temperature,
"max_tokens": self.config.max_tokens,
}
if response_format:
kwargs["extra_body"] = {"guided_json": response_format}
resp = await self.client.chat.completions.create(**kwargs)
return resp.choices[0].message.content or ""
@abstractmethod
async def run(self, inbox: asyncio.Queue[Message], outbox: asyncio.Queue[Message]) -> None:
...
The generate method accepts an optional Pydantic-style JSON schema passed through vLLM's guided_json extra body parameter, guaranteeing the model returns valid JSON for downstream parsing.
Implementing the Planner Agent
The planner takes a research question and produces a list of sub-questions. Constrained decoding ensures the output is always a valid list.
PLANNER_SCHEMA = {
"type": "object",
"properties": {
"sub_questions": {
"type": "array",
"items": {"type": "string"},
"minItems": 2,
"maxItems": 6,
}
},
"required": ["sub_questions"],
}
PLANNER_SYSTEM = (
"You are a research planner. Given a research question, "
"decompose it into 2-6 focused sub-questions that can be "
"answered independently. Return JSON only."
)
class PlannerAgent(BaseAgent):
def __init__(self, client: AsyncOpenAI):
super().__init__(
AgentConfig(
name="planner",
model="meta-llama/Meta-Llama-3.1-8B-Instruct",
temperature=0.2,
max_tokens=1024,
system_prompt=PLANNER_SYSTEM,
),
client,
)
async def run(self, inbox: asyncio.Queue[Message], outbox: asyncio.Queue[Message]) -> None:
while True:
msg = await inbox.get()
if msg.msg_type == "research_question":
raw = await self.generate(
f"Research question: {msg.payload['question']}",
response_format=PLANNER_SCHEMA,
)
data = json.loads(raw)
for sq in data["sub_questions"]:
await outbox.put(Message(
sender="planner",
recipient="searcher",
msg_type="sub_question",
payload={"question": sq, "parent": msg.payload["question"]},
))
inbox.task_done()Implementing the Searcher and Reader Agents
The searcher converts a sub-question into search queries and retrieves documents. For this example we use a stub retriever; in production you would plug in a vector store or web search API.
SEARCHER_SYSTEM = (
"You generate concise web search queries. "
"Return JSON with a 'queries' array of 1-3 strings."
)
SEARCH_SCHEMA = {
"type": "object",
"properties": {
"queries": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 3}
},
"required": ["queries"],
}
async def mock_retrieve(query: str, k: int = 3) -> list[dict]:
# Replace with: vector DB lookup, SerpAPI, Tavily, etc.
return [{"title": f"Result for {query}", "text": f"Sample passage about {query}.", "url": "https://example.com"}] * k
class SearcherAgent(BaseAgent):
def __init__(self, client: AsyncOpenAI):
super().__init__(
AgentConfig(name="searcher", model="meta-llama/Meta-Llama-3.1-8B-Instruct",
temperature=0.3, max_tokens=512, system_prompt=SEARCHER_SYSTEM),
client,
)
async def run(self, inbox, outbox):
while True:
msg = await inbox.get()
if msg.msg_type == "sub_question":
raw = await self.generate(
f"Sub-question: {msg.payload['question']}",
response_format=SEARCH_SCHEMA,
)
queries = json.loads(raw)["queries"]
docs = []
for q in queries:
docs.extend(await mock_retrieve(q))
await outbox.put(Message(
sender="searcher",
recipient="reader",
msg_type="documents",
payload={"sub_question": msg.payload["question"], "docs": docs},
))
inbox.task_done()The reader extracts structured facts from each document batch.
READER_SYSTEM = (
"You extract verifiable facts from documents. "
"Return JSON with a 'facts' array; each fact has "
"'claim' (string) and 'citation' (string)."
)
FACT_SCHEMA = {
"type": "object",
"properties": {
"facts": {
"type": "array",
"items": {
"type": "object",
"properties": {
"claim": {"type": "string"},
"citation": {"type": "string"},
},
"required": ["claim", "citation"],
},
}
},
"required": ["facts"],
}
class ReaderAgent(BaseAgent):
def __init__(self, client: AsyncOpenAI):
super().__init__(
AgentConfig(name="reader", model="meta-llama/Meta-Llama-3.1-8B-Instruct",
temperature=0.1, max_tokens=2048, system_prompt=READER_SYSTEM),
client,
)
async def run(self, inbox, outbox):
while True:
msg = await inbox.get()
if msg.msg_type == "documents":
context = "\n\n".join(
f"[{i}] {d['title']}\n{d['text']}\nURL: {d['url']}"
for i, d in enumerate(msg.payload["docs"])
)
raw = await self.generate(
f"Sub-question: {msg.payload['sub_question']}\n\nDocuments:\n{context}",
response_format=FACT_SCHEMA,
)
facts = json.loads(raw)["facts"]
await outbox.put(Message(
sender="reader",
recipient="critic",
msg_type="facts",
payload={"sub_question": msg.payload["sub_question"], "facts": facts},
))
inbox.task_done()Implementing the Critic and Writer Agents
The critic filters low-quality or unsupported claims before they reach the writer.
CRITIC_SYSTEM = (
"You evaluate research facts for accuracy, relevance, and "
"support. Return JSON with 'verified' (array of facts) and "
"'rejected' (array of facts with a 'reason' field)."
)
CRITIC_SCHEMA = {
"type": "object",
"properties": {
"verified": {
"type": "array",
"items": {"type": "object",
"properties": {"claim": {"type": "string"}, "citation": {"type": "string"}},
"required": ["claim", "citation"]},
},
"rejected": {
"type": "array",
"items": {"type": "object",
"properties": {"claim": {"type": "string"}, "reason": {"type": "string"}},
"required": ["claim", "reason"]},
},
},
"required": ["verified", "rejected"],
}
class CriticAgent(BaseAgent):
def __init__(self, client: AsyncOpenAI):
super().__init__(
AgentConfig(name="critic", model="meta-llama/Meta-Llama-3.1-8B-Instruct",
temperature=0.2, max_tokens=2048, system_prompt=CRITIC_SYSTEM),
client,
)
self.verified: list[dict] = []
async def run(self, inbox, outbox):
while True:
msg = await inbox.get()
if msg.msg_type == "facts":
raw = await self.generate(
f"Sub-question: {msg.payload['sub_question']}\n\nFacts: {json.dumps(msg.payload['facts'])}",
response_format=CRITIC_SCHEMA,
)
result = json.loads(raw)
self.verified.extend(result["verified"])
await outbox.put(Message(
sender="critic",
recipient="writer",
msg_type="verified_facts",
payload={"verified": result["verified"]},
))
inbox.task_done()The writer accumulates verified facts and produces the final report when signaled.
WRITER_SYSTEM = (
"You are a research report writer. Synthesize verified facts "
"into a coherent, well-structured markdown report with inline "
"citations. Do not invent facts not present in the input."
)
class WriterAgent(BaseAgent):
def __init__(self, client: AsyncOpenAI, original_question: str):
super().__init__(
AgentConfig(name="writer", model="meta-llama/Meta-Llama-3.1-8B-Instruct",
temperature=0.4, max_tokens=4096, system_prompt=WRITER_SYSTEM),
client,
)
self.original_question = original_question
self.all_facts: list[dict] = []
self.report: str | None = None
async def run(self, inbox, outbox):
while True:
msg = await inbox.get()
if msg.msg_type == "verified_facts":
self.all_facts.extend(msg.payload["verified"])
elif msg.msg_type == "finalize":
prompt = (
f"Original research question: {self.original_question}\n\n"
f"Verified facts:\n{json.dumps(self.all_facts, indent=2)}\n\n"
"Write the final report in markdown."
)
self.report = await self.generate(prompt)
await outbox.put(Message(
sender="writer",
recipient="orchestrator",
msg_type="report",
payload={"report": self.report},
))
inbox.task_done()Building the Orchestrator
The orchestrator wires agents together, routes messages by recipient, and signals the writer once all sub-questions have been processed.
class Orchestrator:
def __init__(self, client: AsyncOpenAI, question: str):
self.client = client
self.question = question
self.queues: dict[str, asyncio.Queue[Message]] = {}
self.agents: dict[str, BaseAgent] = {}
self.report: str | None = None
self._expected_sub_questions = 0
self._completed = 0
def setup(self):
names = ["planner", "searcher", "reader", "critic", "writer"]
for n in names:
self.queues[n] = asyncio.Queue()
self.agents["planner"] = PlannerAgent(self.client)
self.agents["searcher"] = SearcherAgent(self.client)
self.agents["reader"] = ReaderAgent(self.client)
self.agents["critic"] = CriticAgent(self.client)
self.agents["writer"] = WriterAgent(self.client, self.question)
async def _router(self):
"""Forward messages from a shared outbox to recipient inboxes."""
outbox: asyncio.Queue[Message] = asyncio.Queue()
self._outbox = outbox
while True:
msg = await outbox.get()
if msg.recipient == "orchestrator":
if msg.msg_type == "report":
self.report = msg.payload["report"]
return
elif msg.recipient in self.queues:
await self.queues[msg.recipient].put(msg)
# Track sub-question completion for finalize signal
if msg.msg_type == "verified_facts":
self._completed += 1
if self._completed >= self._expected_sub_questions:
await self.queues["writer"].put(Message(
sender="orchestrator",
recipient="writer",
msg_type="finalize",
payload={},
))
outbox.task_done()
async def run(self) -> str:
self.setup()
# Start agents
tasks = []
for name, agent in self.agents.items():
tasks.append(asyncio.create_task(agent.run(self.queues[name], self._outbox if hasattr(self, '_outbox') else asyncio.Queue())))
# We need the router to set _outbox first; restart with proper wiring
for t in tasks:
t.cancel()
self._outbox = asyncio.Queue()
tasks = [
asyncio.create_task(self.agents[n].run(self.queues[n], self._outbox))
for n in self.agents
]
router_task = asyncio.create_task(self._router())
# Kick off: send research question to planner
await self.queues["planner"].put(Message(
sender="user",
recipient="planner",
msg_type="research_question",
payload={"question": self.question},
))
# Wait for planner to emit sub-questions so we know the count
# (In a fuller implementation, planner would echo the count.)
# Here we poll the queue size heuristically:
await asyncio.sleep(2)
# A more robust approach: planner sends a 'plan' message to orchestrator.
# For brevity, assume expected = number of sub_question messages routed.
# We patch by intercepting in router; for this demo set a reasonable cap:
self._expected_sub_questions = max(1, self._completed + 1)
# In production, track via a dedicated 'plan_ready' message.
await router_task
for t in tasks:
t.cancel()
return self.report or "No report generated."The simplified orchestrator above illustrates the pattern. In production, the planner should emit a plan_ready message containing the sub-question count so the orchestrator knows exactly when to send finalize to the writer. This avoids race conditions and polling.
Running the Pipeline End to End
import asyncio
from openai import AsyncOpenAI
async def main():
client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="dummy")
orch = Orchestrator(client, "What are the trade-offs between RAG and long-context LLMs?")
report = await orch.run()
print(report)
if __name__ == "__main__":
asyncio.run(main())When you run this script, the planner decomposes the question, searchers retrieve documents for each sub-question, readers extract facts, the critic filters them, and the writer assembles a cited markdown report. All LLM calls hit the single vLLM server, which batches them transparently.
Scaling to Multiple vLLM Instances
For higher throughput, run multiple vLLM servers behind a load balancer. Because the OpenAI client supports round-robin across base URLs, you can implement a simple pooling client:
import itertools
from openai import AsyncOpenAI
class PooledClient:
def __init__(self, urls: list[str], api_key: str = "dummy"):
self._clients = [AsyncOpenAI(base_url=u, api_key=api_key) for u in urls]
self._cycle = itertools.cycle(self._clients)
@property
def chat(self):
return next(self._cycle).chatPass a PooledClient wherever an AsyncOpenAI instance is expected. For production, prefer a real reverse proxy (Nginx, Envoy, or LiteLLM) that also handles health checks and retries.
Best Practices
Use Constrained Decoding for Inter-Agent Messages
Every agent that emits structured data should use vLLM's guided JSON decoding. Unconstrained outputs cause parse failures that cascade through the pipeline. Define Pydantic models or JSON schemas for every agent output and pass them via extra_body={"guided_json": schema}.
Keep Prompts Focused and Short
Each agent should have a single, narrow responsibility. Resist the temptation to make the reader also summarize or the critic also rewrite. Narrow prompts reduce token usage, improve output quality, and make failures easier to diagnose.
Set Temperature Strategically
Use low temperature (0.0–0.2) for extraction and verification agents where determinism matters. Use higher temperature (0.4–0.7) for the planner and writer where creativity is valuable. Never use temperature above 1.0 in research pipelines — it increases hallucination risk.
Implement Retries with Backoff
vLLM can return 429 or 503 under load. Wrap every generate call with exponential backoff:
import asyncio
from openai import APIError, RateLimitError
async def generate_with_retry(agent, prompt, schema, retries=4):
for attempt in range(retries):
try:
return await agent.generate(prompt, response_format=schema)
except (RateLimitError, APIError) as e:
if attempt == retries - 1:
raise
await asyncio.sleep(2 ** attempt)Persist Intermediate Artifacts
Save every agent's input and output to disk or a database. This enables debugging, replay, and incremental re-runs when a single agent fails. A simple approach is to log each message as a JSONL line:
import time, json, pathlib
log_path = pathlib.Path("pipeline_log.jsonl")
def log_message(msg: Message):
with log_path.open("a") as f:
f.write(json.dumps({
"ts": time.time(),
"from": msg.sender,
"to": msg.recipient,
"type": msg.msg_type,
"payload": msg.payload,
}) + "\n")Monitor vLLM Metrics
vLLM exposes Prometheus metrics at /metrics. Track vllm:num_requests_running, vllm:num_requests_waiting, and vllm:time_to_first_token_seconds. If waiting requests grow unbounded, add GPU capacity or reduce concurrency in the orchestrator with a semaphore.
Cap Concurrency
Unbounded fan-out can overwhelm the server. Limit concurrent searcher calls with a semaphore sized to your GPU's capacity:
sem = asyncio.Semaphore(16)
async def bounded_search(query):
async with sem:
return await mock_retrieve(query)Use Different Models for Different Agents
A powerful advantage of vLLM is serving multiple models on one server (via --served-model-name and LoRA adapters, or separate processes). Use a smaller, faster model for the searcher (query generation is easy) and a larger model for the critic and writer (judgment and synthesis are hard). This optimizes cost and latency simultaneously.
Conclusion
Building a multi-agent research pipeline on vLLM combines the flexibility of role-specialized LLM agents with the throughput and memory efficiency of a production-grade inference engine. By decomposing research into planner, searcher, reader, critic, and writer roles, enforcing structured outputs through guided decoding, and orchestrating them with async message passing, you get a system that is more accurate, more debuggable, and more scalable than a monolithic prompt. Start with the single-server setup described here, instrument it with logging and metrics, then scale horizontally with pooled vLLM instances and concurrency controls as your research workload grows. The architecture patterns in this guide — constrained inter-agent schemas, narrow agent responsibilities, retry-aware orchestration, and persistent artifacts — form a durable foundation that adapts as models and retrieval backends evolve.