Build a Verified Research Crew with CrewAI
Build a multi-agent AI crew that fact-checks claims with programmatic guardrails.
Introduction
30 Second Summary
You ask an AI to research a topic, and it hands you a confident-looking briefing. But when you click the source links, half of them go nowhere. The AI hallucinated its citations.
In this project, you will build a multi-agent research system that doesn't just generate research. It verifies it. Three specialized AI agents collaborate in sequence: one researches, one compresses the findings into atomic facts, and one cross-checks every claim against its source using a programmatic guardrail that rejects anything it can't verify.
What You'll Build
You feed in a company or topic, and your crew produces a verified briefing where every claim traces back to a real source. Unverified claims get rejected and re-researched automatically.
By the end of this project, you'll have:
- A multi-agent research crew powered by CrewAI and Groq that produces fact-checked briefings with source citations on any topic you give it.
- Custom CrewAI tools with full Pydantic typed input/output schemas, including a context compressor that extracts only the atomic facts worth keeping from verbose research.
- A programmatic Python guardrail that structurally validates every claim, rejects hallucinated sources, and forces the agent to retry with actionable feedback.
- Secret Mission: Wrap your crew in an event-driven CrewAI Flow with conditional routing that sends failed verifications back to the researcher automatically.
Are there any prerequisites?
This project assumes intermediate Python knowledge. You should be comfortable with classes, decorators, and type hints.
You do not need prior experience with CrewAI or multi-agent systems. The project teaches those from scratch.
Before We Start
Before We Start...
Before we dive in, let's lock in what you're about to build and why it matters. You're going to build a multi-agent research crew where a programmatic guardrail rejects unverified claims automatically. That's a different league from telling an LLM "please double-check your sources" in a prompt.
Set Up Your Environment
Before you can build agents that verify research, you need the right tools in place. CrewAI orchestrates your agents, and Groq gives them a powerful LLM brain for free.
In this step, you will install the CrewAI framework, connect it to Groq's free API, and scaffold the project files your agents will live in.
In this step, get ready to:
- Install uv and CrewAI with its LiteLLM integration.
- Create a free Groq account and wire up your API key.
- Scaffold the project file structure and verify the Groq connection.
Install uv and CrewAI
uv is a fast Python package manager that handles virtual environments and dependencies. You will use it to create an isolated environment for your CrewAI project.
- Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Terminal and pressing Enter.
- Move to your Desktop by running this command:
cd ~/Desktop
- Install uv by running the appropriate command for your operating system:
macOS
curl -LsSf https://astral.sh/uv/install.sh | sh
Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
What is uv?
uv is an extremely fast Python package manager written in Rust. It replaces pip and venv with faster equivalents while staying compatible with the same commands you already know.
- Create your project directory and move into it by running these commands:
mkdir verified-research-crew
cd verified-research-crew
- Create a virtual environment by running:
uv venv
- Activate the virtual environment:
macOS
source .venv/bin/activate
Windows
.venv\Scripts\activate
You should see (.venv) appear at the start of your terminal prompt.
Don't see (.venv) in your prompt?
If you see a permission error on macOS, make sure you ran the source command (not just the path). On Windows, if execution policy blocks activation, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned first.
help me activate a Python virtual environment
- Install CrewAI with the LiteLLM extra by running:
pip install "crewai[litellm]"
Why the litellm extra?
CrewAI uses LiteLLM under the hood to connect to different LLM providers. The [litellm] extra installs the adapter that lets CrewAI talk to Groq using one consistent interface.
Before you run the verification, what do you expect to see if the install succeeded?
- Verify the installation works by running:
python -c "from crewai import Agent; print('CrewAI ready')"
You should see CrewAI ready printed in your terminal.
Seeing ModuleNotFoundError?
Make sure your virtual environment is active. Look for (.venv) at the start of your terminal prompt.
If it disappeared, reactivate it with the activate command for your OS, then re-run the pip install command.
help me debug a ModuleNotFoundError for crewai after pip install
Create your Groq account and API key
Groq provides free access to powerful LLMs including the 70-billion parameter Llama 3.3 model. No credit card required.
- Open console.groq.com in your browser.
- Sign up for a free account (no credit card needed).
- Once signed in, click API Keys in the left sidebar.
- Click Create API Key.
- Give it a name like crewai-research and click Submit.
- Copy the key immediately. You will not be able to see it again after closing the dialog.
- Back in your terminal, install python-dotenv by running:
pip install python-dotenv
- Create a .env file in your project folder by running:
echo "GROQ_API_KEY=your-key-here" > .env
- Open the .env file and replace your-key-here with the API key you just copied from the Groq console.
- Save the file.
Keep your key safe
Never commit .env files to version control. The .env file holds your credentials locally so they never appear in your code.
Create the project files and test the Groq connection
Your project needs four Python files that will hold different responsibilities: data models, custom tools, guardrails, and the main orchestration logic.
- Create all four project files by running:
echo "# placeholder" > models.py
echo "# placeholder" > tools.py
echo "# placeholder" > guardrails.py
echo "# placeholder" > main.py
- Confirm the files exist by running:
ls
You should see main.py, tools.py, guardrails.py, models.py, and .env listed in the output.
What will each file do?
- models.py defines the Pydantic data shapes your agents produce (claims, reports).
- tools.py holds custom tools your agents use (compressor, verifier).
- guardrails.py contains the programmatic validation that rejects bad output.
- main.py wires agents, tasks, and the crew together.
Now test that your Groq connection works end-to-end by creating a minimal agent.
- Test the Groq connection by running this command:
python -c "
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, LLM
load_dotenv()
llm = LLM(model='groq/llama-3.3-70b-versatile')
agent = Agent(role='Tester', goal='Say hello', backstory='You test connections.', llm=llm)
task = Task(description='Say hello world', expected_output='A greeting', agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()
print(result.raw)
"
You should see a short greeting printed by the LLM, confirming your Groq API key works and CrewAI can orchestrate an agent.
Seeing an authentication error?
Double-check your .env file has GROQ_API_KEY set to the full key you copied (no extra spaces or quotes around the value).
Make sure you saved the .env file before running the test command.
help me debug a Groq API authentication error in CrewAI
- Open VS Code by pressing Cmd+Space (macOS) or the Windows key (Windows), typing VS Code, and pressing Enter.
- Click File in the top menu bar.
- Click Open Folder.
- Navigate to Desktop and select the verified-research-crew folder, then click Open.
You should see your four Python files and the .env file in the VS Code Explorer sidebar.
✔️ Awesome, I've got everything!
Great. Double check you have all five files visible in your VS Code sidebar: main.py, tools.py, guardrails.py, models.py, and .env.
ⓧ I'd like to double check the full code
Each file should contain just a single comment for now. Here is what each file looks like at this point:
.env:
GROQ_API_KEY=your-key-here
models.py, tools.py, guardrails.py, and main.py each contain:
# placeholder
Your environment is wired up and talking to Groq. Next up, you will build a basic research crew that exposes the exact problems verified research solves.
Build a Basic Research Crew
You have CrewAI installed and your Groq connection working. Now you need a crew that actually does something.
In this step, you'll build a naive research crew with two agents: a Researcher and a Writer. You'll run it on a real topic and watch the output. The goal is to see exactly where a basic crew falls short so you know what to fix.
In this step, get ready to:
- Define Pydantic models for structured agent output.
- Create a Researcher and Writer agent wired into a sequential crew.
- Run the crew and observe its verification gaps.
Define your data models
Before your agents can produce structured output, you need to define what that structure looks like. Pydantic models act as schemas that enforce the shape of your agent's response.
- Open models.py in VS Code and add the following code:
from pydantic import BaseModel, Field
from typing import List
class ResearchClaim(BaseModel):
"""A single research claim with its source."""
claim: str = Field(description="A specific factual claim")
source_url: str = Field(description="URL of the source supporting this claim")
confidence: float = Field(default=0.5, description="Confidence score 0-1")
class ResearchReport(BaseModel):
"""Structured research report with verified claims."""
topic: str = Field(description="The research topic")
claims: List[ResearchClaim] = Field(description="List of researched claims with sources")
summary: str = Field(description="Executive summary of findings")
What does this code do?
You defined two models that enforce structure on your agents' output:
- ResearchClaim holds a single factual claim, the URL where it was found, and a confidence score.
- ResearchReport groups multiple claims under a topic with a summary. When you attach this to a CrewAI task via output_pydantic, the agent is forced to produce output matching this exact shape.
- Save models.py.
- Verify the models import correctly by running this in your terminal:
python -c "from models import ResearchReport; print('Models ready')"
You should see Models ready printed.
Seeing a ModuleNotFoundError?
- Make sure your terminal is in the project directory (the folder containing models.py).
- Confirm your virtual environment is activated. You should see the environment name in your terminal prompt.
- Check for typos in models.py. The class names must be exactly ResearchClaim and ResearchReport.
Why isn't my models.py importing correctly?
Create your agents and crew
Now you'll build the actual agents and wire them into a sequential crew. The Researcher finds claims about a topic. The Writer synthesizes those claims into a structured briefing.
- Open main.py and add the imports and LLM configuration:
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process, LLM
from models import ResearchReport
# Load environment variables so CrewAI can access GROQ_API_KEY
load_dotenv()
# Configure the LLM to use Groq's free API
llm = LLM(
model="groq/llama-3.3-70b-versatile",
temperature=0.7
)
What does this code do?
- load_dotenv() reads your .env file so CrewAI can access your GROQ_API_KEY automatically.
- The LLM object tells CrewAI to route all agent calls through Groq using the llama-3.3-70b-versatile model (70 billion parameters, free on Groq).
- Save main.py.
- Verify the LLM configuration loads correctly by running:
python -c "from main import llm; print(llm.model)"
You should see groq/llama-3.3-70b-versatile printed.
- Now add the two agent definitions below the LLM config in main.py:
# Agent 1: Researches the topic and produces claims
researcher = Agent(
role="Senior Research Analyst",
goal="Research {topic} thoroughly and produce claims with specific source URLs",
backstory=(
"You are a meticulous researcher who always cites specific sources. "
"Every claim you make must have a real, verifiable URL."
),
llm=llm,
verbose=True
)
# Agent 2: Turns raw research into a polished briefing
writer = Agent(
role="Research Writer",
goal="Synthesize research into a clear briefing",
backstory="You turn raw research into polished briefings.",
llm=llm,
verbose=True
)
What does this code do?
- Each agent has a role, goal, and backstory that shape how the LLM behaves. The {topic} placeholder gets filled at runtime when the crew kicks off.
- verbose=True prints each agent's reasoning in your terminal so you can watch the crew think.
- The Researcher's backstory tells it to cite real URLs. You'll soon see whether that instruction alone is enough to prevent fabrication.
- Save main.py.
- Now add the tasks, crew, and entry point below the writer agent in main.py:
# Define what each agent does as a Task
research_task = Task(
description="Research the topic: {topic}. Find at least 5 specific factual claims with source URLs.",
expected_output="A list of claims with source URLs.",
agent=researcher
)
# The writing task receives context from research and enforces Pydantic output
writing_task = Task(
description="Synthesize the research into a briefing.",
expected_output="A JSON object with topic, claims, and summary.",
agent=writer,
context=[research_task],
output_pydantic=ResearchReport
)
# Wire agents and tasks into a sequential crew
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
verbose=True
)
def main():
topic = input("Enter a research topic: ") or "Tesla Q2 2025 earnings"
result = crew.kickoff(inputs={"topic": topic})
print(result.raw)
if __name__ == "__main__":
main()
What does this code do?
- context=[research_task] passes the Researcher's output as input to the Writer. This is how agents share information in a sequential crew.
- output_pydantic=ResearchReport forces the final task's output to match your Pydantic schema. CrewAI instructs the LLM to produce JSON matching that shape.
- Process.sequential means agents run one after another: Researcher first, then Writer. No parallelism.
- crew.kickoff(inputs={"topic": topic}) starts the crew and fills in the {topic} placeholder in all agent goals and task descriptions.
- Save main.py.
Seeing red underlines in VS Code?
- If VS Code shows import errors for crewai, make sure VS Code is using your virtual environment's Python interpreter. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), type "Python: Select Interpreter", and choose the one inside your virtual environment.
Help me fix my CrewAI import errors in VS Code.
✔️ Awesome, I've got everything!
Great. Make sure you have saved main.py with all the code above.
ⓧ I'd like to double check the full code
Your main.py should contain all three chunks you added above. Scroll up and confirm you have:
- The imports and LLM config at the top (ending with the llm = LLM(...) block).
- Both agent definitions (researcher and writer) in the middle.
- The tasks, crew, and main() function at the bottom.
Your models.py should contain the ResearchClaim and ResearchReport classes from the first substep.
Run the crew and observe the problem
Your crew is assembled. Two agents, two tasks, structured output enforced. Time to see if a simple "please cite sources" instruction in the agent's backstory actually produces reliable research.
Before you run this, think about what you expect. The Researcher agent's backstory says "every claim must have a real, verifiable URL." Do you think that prompt instruction alone will guarantee real source URLs in the output?
- Run your crew by executing this command:
python main.py
- When prompted, press Enter to use the default topic ("Tesla Q2 2025 earnings"), or type your own topic.
What should I see?
Your terminal fills with verbose output as each agent reasons through its task. The Researcher thinks, produces claims, then the Writer receives that context and synthesizes a briefing. This takes 30-60 seconds depending on Groq's response time.
At the end, you'll see the final output printed: a JSON-like structure with a topic, a list of claims, and a summary.
Look carefully at the output. You'll notice the problems:
- Some claims have empty source_url fields or obviously fabricated URLs (like https://www.example.com or generic domains).
- The research context passed from the Researcher to the Writer is bloated with verbose reasoning that wastes the Writer's context window.
- There is no mechanism rejecting bad claims. The crew happily outputs whatever the LLM generates, hallucinated sources and all.
Why does this happen?
Prompt instructions ("always cite real sources") are suggestions, not enforcement. The LLM can and will ignore them when it doesn't have a real URL to provide. It fills in a plausible-looking URL rather than admitting it doesn't know.
This is exactly why you need programmatic guardrails: code that structurally validates the output and rejects it when claims lack real sources. Prompts ask nicely. Guardrails enforce.
Crew erroring out instead of producing output?
- If you see an authentication error, double-check that your .env file contains GROQ_API_KEY= followed by your actual key (no spaces around the =).
- If you see a rate limit error, wait 60 seconds and try again. Groq's free tier has per-minute limits.
- If the output looks malformed or the crew hangs, press Ctrl+C to stop it and run again. LLM responses are non-deterministic, and occasionally one attempt produces better structure than another.
My CrewAI crew is failing with an error when I run it. Help me debug.
Your crew runs. It produces output. But that output cannot be trusted. Some sources are fabricated, the context between agents is bloated, and nothing stops bad claims from reaching the final briefing. In the next step, you'll build a custom tool that solves the context bloat problem by compressing verbose research into atomic facts.
Build the Context Compressor Tool
In the previous step, you ran the crew and saw the problem firsthand: the research output is bloated with filler sentences, and whatever the Researcher passes to the Writer floods the context window with noise.
CrewAI has a built-in respect_context_window=True setting, but all it does is auto-summarize when you hit the model's token limit. That's lossy and gives you zero control over what survives. In this step, you'll build a custom tool that uses deterministic Python logic to extract only the atomic facts worth keeping.
In this step, get ready to:
- Define Pydantic models for atomic facts and compression output.
- Build a custom CrewAI tool with deterministic sentence scoring and ranking.
- Test the tool standalone to verify it compresses verbose text into concise facts.
Define the compression data models
The compressor tool needs structured output types. You'll define two new Pydantic models that describe what an atomic fact looks like and what the compressor returns.
- In VS Code, open models.py and add these two classes below the existing ResearchReport class:
class AtomicFact(BaseModel):
"""A single atomic fact extracted from text."""
fact: str = Field(description="The atomic fact")
importance_score: float = Field(description="Importance score 0-1")
class CompressorOutput(BaseModel):
"""Output from the context compressor."""
facts: List[AtomicFact] = Field(description="Extracted atomic facts")
original_length: int = Field(description="Character count of original text")
compressed_length: int = Field(description="Character count of compressed output")
compression_ratio: float = Field(description="Ratio of compression achieved")
What do these models represent?
- AtomicFact holds a single distilled fact and a score indicating how information-dense it is.
- CompressorOutput wraps the list of extracted facts along with compression metrics, so you can see exactly how much noise was cut.
- Save models.py.
- Confirm the models import cleanly by running this command in your terminal:
python -c "from models import AtomicFact, CompressorOutput; print('Models ready')"
You should see Models ready printed with no errors.
Seeing an ImportError?
- Make sure you added the new classes below the existing ResearchReport class, not inside it. Each class should be at the top level of the file.
- Check that List is still imported from typing at the top of the file.
Help me debug my models.py import error.
Create the tool skeleton in tools.py
CrewAI custom tools subclass BaseTool and declare a Pydantic args_schema so the agent knows what inputs the tool accepts. You'll set up the class structure first, then add the scoring logic.
- Open tools.py (currently empty) and paste the following imports and class definitions:
import re
from typing import Type
from pydantic import BaseModel, Field
from crewai.tools import BaseTool
from models import CompressorOutput, AtomicFact
class CompressorInput(BaseModel):
"""Input schema for the ContextCompressorTool."""
text: str = Field(description="The text to compress into atomic facts")
max_facts: int = Field(default=10, description="Maximum number of facts to extract")
class ContextCompressorTool(BaseTool):
name: str = "Context Compressor"
description: str = (
"Compresses verbose research text into ranked atomic facts. "
"Use this to reduce long research output before passing to the next agent."
)
args_schema: Type[BaseModel] = CompressorInput
What does this code do?
- CompressorInput defines the tool's input contract: a block of text and an optional cap on how many facts to extract.
- ContextCompressorTool subclasses BaseTool from CrewAI. The args_schema tells the agent exactly what parameters this tool needs.
- The name and description fields are what the agent reads when deciding whether to use this tool.
- Save tools.py.
- Verify the tool class loads by running:
python -c "from tools import ContextCompressorTool; print('Tool class loaded')"
You should see Tool class loaded printed. The tool won't do anything yet because it has no _run method.
Seeing a crewai import error?
- Confirm your virtual environment is active. The crewai package must be installed in the environment you're running Python from.
- Double-check that the import path is from crewai.tools import BaseTool (not from crewai import BaseTool).
Help me fix my CrewAI tool import.
Implement the scoring and compression logic
The core of this tool is pure Python. No LLM call, no API request. It splits text into sentences, scores each one by information density, and returns only the top-ranked facts.
- In tools.py, add the _run method inside the ContextCompressorTool class, directly below the args_schema line:
def _run(self, text: str, max_facts: int = 10) -> CompressorOutput:
# Split input text into individual sentences
sentences = [s.strip() for s in re.split(r'[.!?]+', text) if s.strip()]
scored_facts = []
for sentence in sentences:
score = self._score_sentence(sentence)
# Only keep sentences above the minimum density threshold
if score > 0.3:
scored_facts.append(AtomicFact(fact=sentence, importance_score=score))
# Rank by importance and take the top N
scored_facts.sort(key=lambda f: f.importance_score, reverse=True)
top_facts = scored_facts[:max_facts]
compressed_text = ". ".join(f.fact for f in top_facts)
return CompressorOutput(
facts=top_facts,
original_length=len(text),
compressed_length=len(compressed_text),
compression_ratio=round(len(compressed_text) / max(len(text), 1), 2)
)
What does this method do?
- It splits the input on sentence boundaries (periods, exclamation marks, question marks).
- Each sentence gets a density score from _score_sentence (which you'll add next). Sentences below 0.3 are discarded as noise.
- The survivors are ranked highest-score-first, capped at max_facts, and returned as a structured CompressorOutput with compression metrics.
- Now add the _score_sentence helper and the format_output_for_agent override below the _run method:
def _score_sentence(self, sentence: str) -> float:
score = 0.0
# Numbers suggest concrete data
if re.search(r'\d+', sentence):
score += 0.3
# Multi-word proper nouns suggest named entities
if re.search(r'[A-Z][a-z]+(?:\s[A-Z][a-z]+)+', sentence):
score += 0.2
# Action verbs common in factual reporting
if any(word in sentence.lower() for word in ['revenue', 'growth', 'increased', 'decreased', 'launched', 'acquired', 'reported', 'announced']):
score += 0.2
# Dollar amounts are high-value facts
if re.search(r'\$[\d,.]+[BMK]?', sentence):
score += 0.3
# Goldilocks sentence length (not too short, not too long)
word_count = len(sentence.split())
if 8 <= word_count <= 30:
score += 0.1
return min(score, 1.0)
def format_output_for_agent(self, raw_result: object) -> str:
# Present results as a clean numbered list instead of raw JSON
result = CompressorOutput.model_validate(raw_result)
lines = [f"Compressed {result.original_length} chars to {result.compressed_length} chars ({result.compression_ratio:.0%} ratio)"]
lines.append(f"Extracted {len(result.facts)} atomic facts:")
for i, fact in enumerate(result.facts, 1):
lines.append(f" {i}. [{fact.importance_score:.1f}] {fact.fact}")
return "\n".join(lines)
What do these methods do?
- _score_sentence assigns points based on five heuristics: contains numbers (+0.3), named entities (+0.2), reporting verbs (+0.2), dollar amounts (+0.3), and appropriate length (+0.1). A sentence maxes out at 1.0.
- format_output_for_agent overrides how the agent sees the tool's output. Instead of raw JSON, the agent gets a clean numbered list with scores. This helps the agent understand what it's working with.
Why deterministic scoring instead of an LLM call?
An LLM call here would burn tokens and introduce randomness. The scoring logic is fast, repeatable, and free. It catches the patterns that matter in research text (numbers, entities, dollar figures) without needing a model.
- Save tools.py.
Before you test, what do you think will happen if you pass in a paragraph with a mix of data-heavy and filler sentences? Which ones will survive the 0.3 threshold?
- Test the tool standalone by running this command:
python -c "
from tools import ContextCompressorTool
tool = ContextCompressorTool()
test_text = (
'Tesla reported Q2 2025 revenue of \$25.5B, a 12% increase year-over-year. '
'The weather in Austin was sunny. '
'The company announced plans to expand Gigafactory Texas. '
'Elon Musk stated that production increased by 40 percent. '
'The office had nice furniture. '
'Tesla acquired a robotics startup for \$200M.'
)
result = tool.run(text=test_text)
print(result)
"
You should see a numbered list of atomic facts, with the data-rich sentences (revenue figures, acquisition price, growth percentages) scoring highest. The filler sentences ("weather was sunny", "nice furniture") should be absent. The compressed output is shorter than the input.
Seeing a NotImplementedError or empty output?
- If you see NotImplementedError, the _run method is not indented correctly inside the ContextCompressorTool class. All three methods must be indented one level inside the class body.
- If you see zero facts extracted, check that the regex patterns in _score_sentence use raw strings (r'...'). Without the r prefix, backslashes get eaten.
- Make sure _score_sentence is defined before the closing of the class (not outside it as a standalone function).
Help me debug my ContextCompressorTool output.
✔️ Awesome, I've got everything!
Great. Double check you've saved both models.py and tools.py.
ⓧ I'd like to double check the full code
Here is the full models.py at the end of this step:
from pydantic import BaseModel, Field
from typing import List
class ResearchClaim(BaseModel):
"""A single research claim with its source."""
claim: str = Field(description="A specific factual claim")
source_url: str = Field(description="URL of the source supporting this claim")
confidence: float = Field(default=0.5, description="Confidence score 0-1")
class ResearchReport(BaseModel):
"""Structured research report with verified claims."""
topic: str = Field(description="The research topic")
claims: List[ResearchClaim] = Field(description="List of researched claims with sources")
summary: str = Field(description="Executive summary of findings")
class AtomicFact(BaseModel):
"""A single atomic fact extracted from text."""
fact: str = Field(description="The atomic fact")
importance_score: float = Field(description="Importance score 0-1")
class CompressorOutput(BaseModel):
"""Output from the context compressor."""
facts: List[AtomicFact] = Field(description="Extracted atomic facts")
original_length: int = Field(description="Character count of original text")
compressed_length: int = Field(description="Character count of compressed output")
compression_ratio: float = Field(description="Ratio of compression achieved")
For tools.py, confirm these sections are present in order:
- Imports (re, Type, BaseModel, Field, BaseTool, CompressorOutput, AtomicFact).
- CompressorInput class with text and max_facts fields.
- ContextCompressorTool class with name, description, and args_schema.
- _run method that splits, scores, ranks, and returns CompressorOutput.
- _score_sentence method with five heuristic checks.
- format_output_for_agent method returning a numbered list string.
Your compression tool is working. It takes verbose research text and extracts only the information-dense facts. Next up, you'll build the Source Verifier tool and the guardrail that rejects unverified claims.
Build the Source Verifier and Guardrail
In the last step, you saw that the basic crew produces claims with fabricated or missing source URLs. A prompt telling the agent to "please verify your sources" does nothing to prevent this structurally.
You need a programmatic guardrail that inspects the output after the agent finishes. It checks every claim for a non-empty, well-formed URL and rejects the entire output if any claim fails. The agent is forced to retry with specific feedback about what to fix.
In this step, get ready to:
- Build a SourceVerifierTool that checks URL format validity and detects placeholder URLs.
- Write a programmatic guardrail function that rejects task output when claims lack valid sources.
- Test the guardrail in isolation to prove it catches bad output before any agent sees it.
Build the SourceVerifierTool
The SourceVerifierTool is a custom CrewAI tool that an agent can call to check whether a URL is legitimate. It uses deterministic Python logic (regex pattern matching) to detect malformed URLs and known placeholder patterns that LLMs fabricate.
- The tool needs a VerificationResult model to structure its output. Add this model to models.py below your existing CompressorOutput class:
class VerificationResult(BaseModel):
"""Result of verifying a single claim."""
claim: str = Field(description="The claim that was verified")
source_url: str = Field(description="The source URL checked")
is_valid_url: bool = Field(description="Whether the URL is well-formed")
is_placeholder: bool = Field(description="Whether the URL appears to be a placeholder")
verification_status: str = Field(description="PASS, FAIL, or WARN")
reason: str = Field(description="Explanation of the verification result")
What does this model do?
Each field captures one dimension of the verification check. is_valid_url tracks format validity, is_placeholder catches fabricated domains like example.com, and verification_status gives a clean PASS/FAIL/WARN verdict the agent can act on.
- Save models.py.
- Verify the new model imports correctly by running:
python -c "from models import VerificationResult; print('VerificationResult ready')"
You should see VerificationResult ready printed.
Seeing a syntax error?
Make sure the new class is at the same indentation level as your other models. Check that you have a blank line separating it from CompressorOutput above.
Why can't Python import my VerificationResult model?
Now build the tool itself. The SourceVerifierTool follows the same BaseTool pattern you used for the compressor: a Pydantic input schema, a _run method with the core logic, and a format_output_for_agent override.
- In tools.py, find the import line at the top that says:
from models import CompressorOutput, AtomicFact
- Add VerificationResult to that import so it looks like this:
from models import CompressorOutput, AtomicFact, VerificationResult
- Below the ContextCompressorTool class (after its last method), add the VerifierInput schema and the beginning of the SourceVerifierTool class:
class VerifierInput(BaseModel):
"""Input schema for the SourceVerifierTool."""
claim: str = Field(description="The claim to verify")
source_url: str = Field(description="The URL cited as the source")
class SourceVerifierTool(BaseTool):
name: str = "Source Verifier"
description: str = (
"Verifies that a claim has a valid, non-placeholder source URL. "
"Checks URL format and flags obviously fabricated sources."
)
args_schema: Type[BaseModel] = VerifierInput
def _run(self, claim: str, source_url: str) -> VerificationResult:
# Check if URL has valid format (scheme + domain)
is_valid_url = bool(re.match(
r'^https?://[a-zA-Z0-9][-a-zA-Z0-9]*(\.[a-zA-Z0-9][-a-zA-Z0-9]*)+(/\S*)?$',
source_url
))
# Known placeholder URL patterns that LLMs fabricate
placeholder_patterns = [
r'example\.com', r'placeholder', r'source\.com',
r'url\.com', r'website\.com', r'link\.com',
r'^https?://www\.[a-z]+\.com/?$'
]
is_placeholder = any(re.search(p, source_url.lower()) for p in placeholder_patterns)
What does this code do?
The regex in is_valid_url checks that the URL starts with http:// or https:// followed by a valid domain structure. This catches completely mangled strings.
The placeholder_patterns list catches LLMs' favourite fake domains. When an LLM doesn't know a real URL, it defaults to patterns like example.com or generic single-word domains.
- Directly below the code you just added (still inside the _run method), add the decision logic and the format_output_for_agent override:
# Determine verification status based on checks
if not source_url or source_url.strip() == "":
status = "FAIL"
reason = "No source URL provided"
elif not is_valid_url:
status = "FAIL"
reason = f"URL is malformed: {source_url}"
elif is_placeholder:
status = "FAIL"
reason = f"URL appears to be a placeholder: {source_url}"
else:
status = "PASS"
reason = "URL is well-formed and does not match known placeholder patterns"
return VerificationResult(
claim=claim,
source_url=source_url,
is_valid_url=is_valid_url,
is_placeholder=is_placeholder,
verification_status=status,
reason=reason
)
def format_output_for_agent(self, raw_result: object) -> str:
result = VerificationResult.model_validate(raw_result)
icon = "PASS" if result.verification_status == "PASS" else "FAIL"
return f"[{icon}] Claim: {result.claim[:60]}... | Source: {result.source_url} | {result.reason}"
What does this code do?
The if/elif chain assigns a FAIL status for empty URLs, malformed URLs, or placeholder URLs. Only URLs that pass all three checks get a PASS.
The format_output_for_agent override converts the raw Pydantic object into a human-readable line the agent can parse at a glance, instead of dumping raw JSON.
- Save tools.py.
- Test the tool by running it directly on a known-bad URL:
python -c "from tools import SourceVerifierTool; t = SourceVerifierTool(); r = t._run(claim='Revenue grew 50%', source_url='https://example.com'); print(r.verification_status, r.reason)"
You should see FAIL URL appears to be a placeholder: https://example.com printed. The tool correctly flags a placeholder domain.
Seeing an IndentationError?
The decision logic (the if/elif/else block) and return statement must be indented inside the _run method (two levels: class + method). The format_output_for_agent method must be at the same level as _run (one level inside the class).
Help me fix an indentation error in my SourceVerifierTool class.
Write the verification guardrail
The tool checks individual URLs, but you need something that inspects the entire task output and decides whether to accept or reject it. This is what CrewAI calls a guardrail: a function that runs after a task completes and gates whether the output moves forward.
What is a CrewAI guardrail?
A guardrail is a plain Python function that receives the task's output and returns a tuple. (True, output) means the output passed validation and moves to the next task. (False, "reason") means the output failed, and CrewAI sends the rejection reason back to the agent as feedback for a retry.
- Open guardrails.py (currently empty) and add the imports and the JSON parsing logic:
import json
import re
from typing import Tuple, Union
def verify_research_output(output) -> Tuple[bool, Union[str, object]]:
"""
Programmatic guardrail that validates research output.
Checks that every claim has a non-empty, well-formed source URL.
Returns (True, output) if valid, or (False, rejection_reason) if invalid.
"""
# Extract raw text from TaskOutput object
raw_text = output.raw if hasattr(output, 'raw') else str(output)
# Parse JSON, handling cases where it's embedded in other text
try:
data = json.loads(raw_text)
except (json.JSONDecodeError, TypeError):
json_match = re.search(r'\{[\s\S]*\}', raw_text)
if json_match:
try:
data = json.loads(json_match.group())
except json.JSONDecodeError:
return (False, "Output is not valid JSON. Provide your response as a valid JSON object matching the required schema.")
else:
return (False, "Output is not valid JSON. Provide your response as a valid JSON object matching the required schema.")
Why the double-try JSON parsing?
LLMs sometimes wrap JSON in markdown code fences or conversational text. The first json.loads tries to parse the raw output directly. If that fails, the regex extracts the first JSON object from the surrounding text and tries again. This makes the guardrail robust to messy LLM formatting.
- Directly below the code you just added (still inside the function), add the claim validation logic:
# Check that claims exist
claims = data.get("claims", [])
if not claims:
return (False, "No claims found in output. Include at least 3 researched claims with source URLs.")
# Validate each claim's source URL
failures = []
for i, claim in enumerate(claims, 1):
source = claim.get("source_url", "")
claim_text = claim.get("claim", f"Claim {i}")
if not source or source.strip() == "":
failures.append(f"Claim {i} ('{claim_text[:50]}...') has no source URL.")
elif not re.match(r'^https?://[a-zA-Z0-9]', source):
failures.append(f"Claim {i} has malformed URL: '{source}'. Must start with http:// or https://")
elif re.search(r'example\.com|placeholder|source\.com|url\.com', source.lower()):
failures.append(f"Claim {i} has placeholder URL: '{source}'. Provide a real source.")
# Return pass or fail with actionable feedback
if failures:
rejection = "VERIFICATION FAILED. Fix these issues:\n" + "\n".join(f"- {f}" for f in failures)
rejection += "\n\nRe-do your research and provide real, specific source URLs for every claim."
return (False, rejection)
return (True, output)
What does this code do?
The loop iterates every claim in the parsed JSON. For each claim, it checks three conditions in order: is the source URL empty, is it malformed, or does it match a placeholder pattern.
If any claim fails, the function builds a rejection string naming exactly which claims failed and why. This feedback goes back to the agent so it knows precisely what to fix on retry.
- Save guardrails.py.
Seeing unexpected indentation?
Everything from claims = data.get("claims", []) onwards must be indented one level inside the verify_research_output function. If you pasted it at the top level (no indent), Python will throw an error.
Help me fix indentation in my guardrails.py file.
Test the guardrail in isolation
Before wiring the guardrail into the crew, you need to prove it actually catches bad output. You will create mock objects that simulate what CrewAI's TaskOutput looks like and run them through the function.
- Run this test directly in your terminal to check the guardrail rejects bad output:
python -c "
import json
from guardrails import verify_research_output
class MockOutput:
def __init__(self, data):
self.raw = json.dumps(data)
# Test with one claim missing a source URL
bad_output = MockOutput({
'topic': 'Test',
'claims': [
{'claim': 'Revenue grew 50%', 'source_url': 'https://reuters.com/article/123'},
{'claim': 'Stock rose 10%', 'source_url': 'https://bloomberg.com/news/456'},
{'claim': 'New product launched', 'source_url': ''},
],
'summary': 'Test summary'
})
passed, reason = verify_research_output(bad_output)
print(f'Passed: {passed}')
print(f'Reason: {reason}')
"
Before you hit Enter, what do you think the guardrail will return for claim 3 (the one with an empty source URL)?
You should see Passed: False followed by a rejection message that specifically names Claim 3 as having no source URL. The guardrail caught the problem and told the agent exactly what to fix.
- Now test the happy path to confirm valid output passes through by running:
python -c "
import json
from guardrails import verify_research_output
class MockOutput:
def __init__(self, data):
self.raw = json.dumps(data)
# Test with all claims having valid source URLs
good_output = MockOutput({
'topic': 'Test',
'claims': [
{'claim': 'Revenue grew 50%', 'source_url': 'https://reuters.com/article/123'},
{'claim': 'Stock rose 10%', 'source_url': 'https://bloomberg.com/news/456'},
{'claim': 'New product launched', 'source_url': 'https://techcrunch.com/post/789'},
],
'summary': 'Test summary'
})
passed, result = verify_research_output(good_output)
print(f'Passed: {passed}')
"
You should see Passed: True. The guardrail lets valid output through without interference.
Why test in isolation first?
The guardrail is pure Python logic. Testing it with mock objects proves correctness without burning API tokens or waiting for agents to run. When you wire it into the crew in the next step, you already know the validation logic works.
Getting an ImportError?
Make sure you are running the command from your project directory (the folder containing guardrails.py). If you see ModuleNotFoundError, run ls to confirm the file is in the current directory.
Help me debug why my guardrail import is failing.
✔️ Awesome, I've got everything!
Great. Double check you have saved both tools.py and guardrails.py.
ⓧ I'd like to double check the full code
Both files are longer than 30 lines, so scroll up and confirm each section is present in order.
In models.py, confirm you have (in order): ResearchClaim, ResearchReport, AtomicFact, CompressorOutput, VerificationResult.
In tools.py, confirm you have (in order): imports including VerificationResult, then CompressorInput, ContextCompressorTool, VerifierInput, SourceVerifierTool (with both _run and format_output_for_agent).
In guardrails.py, confirm you have: imports (json, re, typing), then the verify_research_output function with JSON parsing, claims iteration, and the pass/fail return logic.
You now have a verification tool and a programmatic guardrail that catches hallucinated sources with zero ambiguity. In the next step, you will wire both into the full crew so the guardrail gates the final output and forces automatic retries when claims fail verification.
Assemble the Verified Research Crew
You now have all three pieces: a ContextCompressorTool that extracts atomic facts, a SourceVerifierTool that validates URLs, and a programmatic guardrail that rejects bad output.
This step wires them into a single crew so the Research Agent compresses its findings before passing them on, the Verifier checks every source URL, and the guardrail automatically forces retries when claims lack valid sources. This is the payoff.
In this step, get ready to:
- Replace the basic crew with a verified research crew that uses your custom tools.
- Attach the programmatic guardrail to the verification task so it rejects bad output.
- Run the full crew and watch the guardrail reject unverified claims in real time.
Wire in imports, LLM config, and tool instances
The current main.py has a basic two-agent crew with no custom tools and no guardrail. You will replace it entirely with the verified version.
- Open main.py in VS Code.
- Select all the existing code and delete it.
- Paste in the following imports, LLM configuration, and tool instances:
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process, LLM
from tools import ContextCompressorTool, SourceVerifierTool
from guardrails import verify_research_output
from models import ResearchReport
load_dotenv()
llm = LLM(
model="groq/llama-3.3-70b-versatile",
temperature=0.7
)
# Instantiate your custom tools
compressor_tool = ContextCompressorTool()
verifier_tool = SourceVerifierTool()
What does this code do?
The imports pull in both custom tools you built, the guardrail function, and your Pydantic output model. The LLM object connects to Groq via LiteLLM under the hood.
Creating tool instances at the module level means every agent can reference the same objects without re-instantiating them.
- Save main.py and confirm the imports resolve by running:
python -c "from main import compressor_tool, verifier_tool; print('Tools imported')"
You should see Tools imported with no errors.
Seeing an ImportError?
- If it says cannot import name 'ContextCompressorTool', make sure tools.py is in the same folder as main.py and contains the class from the previous step.
- If you see ModuleNotFoundError: No module named 'crewai', make sure your virtual environment is activated.
help me debug this import error in main.py
- Below the tool instances in main.py, add the researcher agent:
researcher = Agent(
role="Senior Research Analyst",
goal="Research {topic} thoroughly and produce claims with specific source URLs",
backstory=(
"You are a meticulous researcher who always cites specific sources. "
"Every claim you make must have a real, verifiable URL. "
"Never use placeholder URLs like example.com."
),
llm=llm,
tools=[compressor_tool],
verbose=True,
max_iter=15
)
What changed from the basic crew?
The researcher now has tools=[compressor_tool] so it can compress its own findings into atomic facts before passing them along. The max_iter=15 parameter gives the agent up to 15 iterations to complete its task, which is important because guardrail rejections force retries.
- Below the researcher, add the verifier agent:
verifier = Agent(
role="Fact Verification Specialist",
goal="Verify that every research claim has a valid, non-placeholder source URL",
backstory=(
"You are a rigorous fact-checker. You examine each claim and its source URL. "
"You use the Source Verifier tool to validate URLs. "
"You reject any claim with a missing, malformed, or placeholder URL."
),
llm=llm,
tools=[verifier_tool],
verbose=True,
max_iter=15
)
Why replace the writer with a verifier?
The basic crew had a writer that just synthesized output. The verifier actively checks each claim's source URL using the SourceVerifierTool. This is the structural difference between decoration and engineering.
- Save main.py and confirm both agents load by running:
python -c "from main import researcher, verifier; print('2 agents ready')"
You should see 2 agents ready printed.
Seeing a validation error?
- If you see a Pydantic validation error, check that tools=[compressor_tool] uses the variable name, not the class name. The tool must be instantiated before the agent references it.
help me fix this Pydantic validation error on my CrewAI agent
Define the tasks and attach the guardrail
Agents need tasks that tell them what to do. The verification task is where the guardrail lives, automatically rejecting output that fails structural validation.
- Below the agent definitions in main.py, add the research task:
research_task = Task(
description=(
"Research the topic: {topic}\n\n"
"Find at least 5 specific, factual claims about this topic. "
"For each claim, provide the exact URL where you found this information. "
"Use your Context Compressor tool to distill your findings into atomic facts. "
"Do NOT use placeholder URLs like example.com or made-up URLs."
),
expected_output=(
"A JSON object with: topic (string), claims (array of objects with "
"claim, source_url, and confidence fields), and summary (string). "
"Every claim must have a real source_url starting with https://."
),
agent=researcher
)
Why does the task mention the tool by name?
CrewAI agents decide when to use their tools based on the task description. Explicitly saying "Use your Context Compressor tool" increases the chance the agent actually invokes it rather than skipping it.
- Below the research task, add the verification task with the guardrail attached:
verification_task = Task(
description=(
"Verify the research output from the previous task. "
"Use the Source Verifier tool to check each claim's source URL. "
"Produce a final verified report containing only claims that pass verification. "
"Remove any claims with invalid or placeholder URLs."
),
expected_output=(
"A JSON object with: topic (string), claims (array of verified claims with "
"claim, source_url, and confidence fields), and summary (string). "
"Only include claims that passed URL verification."
),
agent=verifier,
context=[research_task],
output_pydantic=ResearchReport,
guardrail=verify_research_output,
guardrail_max_retries=3
)
What do these task parameters do?
- context=[research_task] passes the research output as input to the verifier so it knows what to check.
- output_pydantic=ResearchReport forces the final output into your Pydantic model structure.
- guardrail=verify_research_output attaches your programmatic validator. If it returns (False, reason), the task automatically retries.
- guardrail_max_retries=3 limits retries before accepting whatever the agent produces.
- Save main.py and confirm the tasks wire together by running:
python -c "from main import research_task, verification_task; print('Tasks ready')"
You should see Tasks ready with no errors.
Seeing a NameError?
- If you see NameError: name 'researcher' is not defined, check that the agent definitions appear above the task definitions. Order matters at module level.
- If the error mentions guardrail, confirm your import reads from guardrails import verify_research_output and the file is named guardrails.py (plural).
help me fix this NameError in main.py
- Below the verification task, add the crew assembly and main function:
crew = Crew(
agents=[researcher, verifier],
tasks=[research_task, verification_task],
process=Process.sequential,
verbose=True
)
def main():
topic = input("Enter a research topic: ") or "OpenAI revenue growth 2025"
print(f"\nResearching: {topic}\n{'=' * 50}")
result = crew.kickoff(inputs={"topic": topic})
print(f"\n{'=' * 50}")
print("VERIFIED RESEARCH BRIEFING")
print(f"{'=' * 50}")
print(result.raw)
if __name__ == "__main__":
main()
What does this code do?
The Crew orchestrates both agents sequentially: researcher first, then verifier. The verbose=True flag prints each agent's reasoning so you can watch the guardrail in action.
The main() function prompts for a topic, kicks off the crew with crew.kickoff(), and prints the final verified briefing.
- Save main.py and confirm the full crew assembles by running:
python -c "from main import crew; print(f'{len(crew.agents)} agents, {len(crew.tasks)} tasks')"
You should see 2 agents, 2 tasks confirming everything is wired.
Seeing an AttributeError on crew?
- Make sure the crew variable is defined after both tasks. If research_task or verification_task is not yet defined when Crew() runs, you will get a NameError.
help me fix this error when assembling my CrewAI crew
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py and that the import check printed 2 agents, 2 tasks.
ⓧ I'd like to double check the full code
Your main.py should contain these blocks in order. Scroll up and confirm each is present:
- Imports: os, dotenv, crewai, tools, guardrails, models.
- load_dotenv() and LLM with groq/llama-3.3-70b-versatile.
- compressor_tool and verifier_tool instances.
- researcher agent with tools=[compressor_tool] and max_iter=15.
- verifier agent with tools=[verifier_tool] and max_iter=15.
- research_task assigned to researcher.
- verification_task with context=[research_task], guardrail=verify_research_output, guardrail_max_retries=3.
- crew with Process.sequential and verbose=True.
- main() function with crew.kickoff() and the if __name__ guard.
Run the verified research crew
Everything is wired. Time to see your programmatic guardrail do real work.
- Before you run the crew, predict: do you think the researcher's claims will pass verification on the first try, or will the guardrail reject some?
- Run the full crew by executing:
python main.py
- When prompted for a topic, type OpenAI revenue 2025 and press Enter.
What should I see?
Because verbose=True is set, you will see each agent's thought process in real time. Watch for these stages:
- The researcher gathers claims and uses the Context Compressor to distill them into atomic facts.
- The verifier receives the compressed research and checks each claim's source URL.
- If any claims have missing or placeholder URLs, you will see the guardrail print a rejection starting with VERIFICATION FAILED and the agent retries.
- After passing verification (or hitting the 3-retry limit), the final briefing prints under VERIFIED RESEARCH BRIEFING.
The guardrail rejection is the key moment. It proves your programmatic validator does structural work that a prompt instruction could never reliably enforce.
Execution takes 30-60 seconds
Multi-agent crews make several LLM calls. With Groq's fast inference you should see results within a minute. If it seems stuck, the agent is likely retrying after a guardrail rejection.
Seeing a rate limit error from Groq?
- Groq's free tier has per-minute rate limits. Wait 60 seconds and run the script again.
- If you see AuthenticationError, confirm your .env file contains a valid GROQ_API_KEY value and you have saved it.
help me debug this Groq API error when running my CrewAI crew
Your multi-agent research crew now produces verified briefings where every claim has a structurally validated source. Next up, you will wrap this crew in a Flow with conditional re-research routing in the Secret Mission.
Secret mission
Add a Flow with Conditional Re-Research
Your crew verifies research, but what happens when verification fails repeatedly? Right now it retries within a single task. In this mission, you'll wrap the entire crew in a CrewAI Flow that conditionally re-routes failed verifications back to the researcher with a refined query, adding stateful event-driven orchestration on top of your verification system.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally with the Groq free tier, so there are no ongoing costs.
Resources you used:
- Groq API key (free tier, no charges).
- Local Python virtual environment with CrewAI installed.
- Project directory with Python files (main.py, tools.py, guardrails.py, models.py, flow_main.py, .env).
Keep everything running
No action needed. Choose this if you're still actively building or plan to extend the project further.
- Your code is local. Nothing is running in the background.
- Groq's free tier has no ongoing charges. Your API key remains active for future use.
- The virtual environment stays on disk but consumes no resources unless you activate it.
Pause - I'll come back to this later
There are no running services to pause. Your code and virtual environment sit idle on disk until you return.
- No processes to stop. The crew only runs when you execute python main.py or python flow_main.py.
- Your Groq API key stays valid. You can pick up where you left off anytime.
Delete - I don't want to use this again
Remove all project resources and revoke your API key so nothing remains.
- Delete your project directory (including the virtual environment) by running this command from your Desktop:
macOS
cd ~/Desktop
rm -rf verified-research-crew
Windows
cd ~\Desktop
Remove-Item -Recurse -Force verified-research-crew
- Revoke your Groq API key by visiting console.groq.com, navigating to your API Keys page, and deleting the key you created for this project.
Nice Work!
Nice Work!
You just built a production-grade multi-agent research system that doesn't just generate information. It verifies it.
You've learned how to:
- Build a multi-agent research crew with CrewAI and Groq, orchestrating specialized agents that collaborate sequentially to research, compress, and verify claims.
- Create custom tools with Pydantic typed I/O and format_output_for_agent overrides, including a deterministic ContextCompressorTool that extracts atomic facts using pure Python logic.
- Implement a programmatic Python guardrail that structurally validates agent output, rejects hallucinated or unsourced claims, and forces automatic retries until the research meets your verification standards.
- Secret Mission: Wrapped the crew in an event-driven Flow with @router() to conditionally re-route failed verifications back to the researcher for refined re-research.
Ready to quiz yourself?