broadie 2.1.1

Broadie: Opinionated framework for building and serving AI agents with build on LangGraph + LangServe.

Broadie

Opinionated AI Agent Framework for Reducing Hallucination

Broadie is a production-ready framework for building AI agents with LangGraph and LangServe. It drastically reduces hallucination through structured contracts between agents, LLMs, and tools using Pydantic models.

🎯 Philosophy

Broadie is opinionated by design. We believe AI agents should:

• Use strict Pydantic schemas to enforce structured outputs and reduce hallucination
• Define clear contracts between agents, LLMs, and tools
• Leverage built-in memory and persistence for stateful conversations
• Support agent-to-agent (a2a) communication through a central registry
• Provide easy deployment with minimal configuration

This opinionated approach reduces hallucination by a huge margin compared to free-form text generation.

🚀 Getting Started

Step 1: Install Broadie

pip install broadie

Step 2: Set Up Google Cloud Authentication

Broadie requires access to Google Cloud Vertex AI. Set up Application Default Credentials:

1. Create or select a service account in your Google Cloud project with Vertex AI permissions (roles/aiplatform.user)

2. Generate and download a JSON key:

   gcloud iam service-accounts keys create key.json \
     --iam-account=YOUR_SA_NAME@YOUR_PROJECT.iam.gserviceaccount.com
   

3. Export the path to the JSON file:

   export GOOGLE_APPLICATION_CREDENTIALS="/absolute/path/to/key.json"
   

4. Set your GCP project and region:

   gcloud config set project YOUR_PROJECT_ID
   gcloud config set ai/region us-central1
   

5. Verify authentication works:

   gcloud auth application-default print-access-token
   

Step 3: Create Your First Agent

Create a file my_agent.py:

from broadie import create_agent

# Create a simple agent - no tools, just conversation
agent = create_agent(
    name="assistant",
    instruction="You are a helpful assistant. Be concise and friendly.",
)

That's it! Now chat with your agent interactively:

broadie chat my_agent.py:agent

You'll get an interactive session where the agent remembers your conversation history.

Step 4: Understanding What Just Happened

Your agent automatically gets:

• Conversation memory - Remembers everything you discuss in the session
• Google Vertex AI integration - Uses Gemini 2.0 Flash by default
• SQLite persistence - Saves conversations to broadie-agent.sqlite3
• Thread management - Each chat session gets a unique thread ID

Step 5: Serve as API

Expose your agent as an HTTP API:

broadie serve my_agent.py:agent --port 8000

📚 Core Concepts

Reducing Hallucination with Structured Outputs

Broadie's key feature is enforcing contracts using Pydantic models to prevent LLM hallucination:

from broadie import create_agent, BaseModel, Field

# Define strict output schema - LLM must follow this structure
class WeatherReport(BaseModel):
    location: str = Field(..., description="City name")
    temperature: str = Field(..., description="Temperature with unit")
    condition: str = Field(..., description="Weather condition")
    recommendation: str = Field(..., description="What to wear/do")

# Agent with structured output
weather_agent = create_agent(
    name="weather_assistant",
    instruction="Provide weather information and recommendations",
    output_schema=WeatherReport  # Enforces structure - no hallucination
)

The LLM cannot return unstructured text or make up fields - it must conform to the schema.

Adding Tools to Your Agent

Tools allow agents to interact with external systems using ToolResponse for structured results:

from broadie import create_agent, tool, ToolResponse

@tool("lookup_weather", description="Get current weather for a location")
def lookup_weather(location: str) -> ToolResponse:
    """Fetch weather data from API."""
    # In production, call real weather API
    return ToolResponse(
        status="success",
        message=f"Weather retrieved for {location}",
        data={
            "location": location,
            "temperature": "22°C",
            "condition": "sunny"
        }
    )

# Agent with tool
agent = create_agent(
    name="weather_bot",
    instruction="Help users check weather and provide advice",
    tools=[lookup_weather]
)

ToolResponse ensures tools return structured data that the LLM cannot hallucinate or modify.

Human-in-the-Loop with Approval Workflows

For sensitive operations, you can require human approval before tools execute. This is perfect for actions like deleting files, sending emails, or making API calls that can't be undone.

from broadie import create_agent, tool, ToolResponse

# Tool that requires approval
@tool(
    name="delete_file",
    description="Delete a file from the filesystem",
    approval_required=True,  # Requires human approval
    approval_message="⚠️ Delete file {filename}? This cannot be undone!",
    risk_level="high"  # Options: low, medium, high, critical
)
def delete_file(filename: str) -> ToolResponse:
    """Delete a file - requires approval before execution."""
    import os
    os.remove(filename)
    return ToolResponse(
        status="success",
        message=f"File {filename} deleted",
        data={"deleted": filename}
    )

# Agent with approval-required tool
agent = create_agent(
    name="file_manager",
    instruction="Help manage files. Always confirm before deleting.",
    tools=[delete_file]
)

How it works:

1. When the agent wants to use delete_file, execution pauses
2. You receive an approval request with the formatted message: "⚠️ Delete file report.txt? This cannot be undone!"
3. You approve or reject the action
4. If approved, the tool executes; if rejected, the agent is notified

Chat with approval workflow:

broadie chat file_manager.py:agent

# You: Delete the old report.txt file
# Agent: [Requests approval to delete report.txt]
#
# ⚠️ APPROVAL REQUIRED
# Tool: delete_file
# Message: Delete file report.txt? This cannot be undone!
# Risk Level: high
#
# Approve? (yes/no): yes
#
# Agent: I've deleted report.txt as requested.

Approval parameters:

• approval_required: Set to True to require approval
• approval_message: Custom message with {parameter} placeholders for tool arguments
• risk_level: Visual indicator of risk ("low", "medium", "high", "critical")
• approval_data: Additional context (dict or callable) to help with the decision

This ensures critical operations are always reviewed by a human before execution, adding a safety layer to your AI agents.

Combining Tools and Structured Outputs

The most powerful pattern combines both:

from broadie import create_agent, tool, ToolResponse
from enum import Enum
from broadie import BaseModel, Field

# Enum prevents invalid values
class Sentiment(str, Enum):
    positive = "positive"
    negative = "negative"
    neutral = "neutral"

# Structured output schema
class EmailAnalysis(BaseModel):
    summary: str = Field(..., description="Brief summary of email")
    sentiment: Sentiment = Field(..., description="Overall sentiment")
    requires_response: bool = Field(..., description="Does this need a reply?")
    priority: int = Field(..., ge=1, le=5, description="Priority 1-5")

# Tool with structured response
@tool("fetch_email", description="Retrieve email by ID")
def fetch_email(email_id: str) -> ToolResponse:
    """Fetch email content."""
    return ToolResponse(
        status="success",
        message=f"Email {email_id} retrieved",
        data={
            "from": "[e-mail]",
            "subject": "Meeting tomorrow",
            "body": "Can we meet at 2pm?"
        }
    )

# Agent combines tool + structured output
email_agent = create_agent(
    name="email_analyzer",
    instruction="Analyze emails and provide structured insights",
    tools=[fetch_email],
    output_schema=EmailAnalysis  # LLM must return this exact structure
)

This ensures:

• Tools return reliable, structured data (not hallucinated)
• LLM output conforms to your schema (no made-up fields)
• Type safety throughout the entire pipeline

Conversation Memory and State

Broadie automatically manages conversation history using thread IDs. Each chat session maintains its own conversation context:

# Start a chat session - gets automatic thread ID
broadie chat my_agent.py:agent

# You: My name is Alice
# Agent: Nice to meet you, Alice! How can I help you today?

# You: What's my name?
# Agent: Your name is Alice!

Use specific thread IDs to resume conversations or maintain separate contexts:

# Continue a specific conversation thread
broadie chat my_agent.py:agent --thread user_123

# Different thread = fresh conversation (no shared history)
broadie chat my_agent.py:agent --thread user_456

All conversation history is automatically persisted to broadie-agent.sqlite3 and managed by LangGraph's checkpointer system. You don't need to configure anything - just chat, and the agent remembers!

Using Sub-Agents for Delegation

Sub-agents handle specialized tasks:

from broadie import create_agent, create_sub_agent

# Specialized sub-agent for classification
spam_detector = create_sub_agent(
    name="spam_classifier",
    prompt="Classify if text is spam. Return 'spam' or 'not_spam'.",
)

# Main agent delegates to sub-agent
email_agent = create_agent(
    name="email_manager",
    instruction="Manage emails, delegate spam detection to sub-agent",
    subagents=[spam_detector]
)

Sub-agents automatically:

• Inherit parent's thread_id
• Share the same conversation context when needed
• Are orchestrated through LangGraph's swarm architecture

Complete Example: Threat Intelligence Agent

Here's a real-world example combining all concepts:

import asyncio
from enum import Enum
from broadie import create_agent, create_sub_agent, tool, ToolResponse
from broadie import BaseModel, Field

# Structured enums prevent hallucination
class ThreatLevel(str, Enum):
    critical = "critical"
    high = "high"
    medium = "medium"
    low = "low"
    none = "none"

class IndicatorType(str, Enum):
    ip = "ip"
    domain = "domain"
    url = "url"
    hash = "hash"

# Output schema enforces structure
class ThreatAnalysis(BaseModel):
    summary: str = Field(..., description="Executive summary of threats found")
    indicators: list[str] = Field(..., description="List of indicators found")
    threat_level: ThreatLevel = Field(..., description="Overall threat assessment")
    recommended_action: str = Field(..., description="What to do next")

# Tool with structured response
@tool("lookup_indicator", description="Check indicator reputation in threat intel database")
def lookup_indicator(indicator: str, indicator_type: IndicatorType) -> ToolResponse:
    """Query threat intelligence database."""
    # Simulated threat intel lookup
    is_malicious = "malware" in indicator.lower() or "phish" in indicator.lower()

    return ToolResponse(
        status="success",
        message=f"Looked up {indicator_type} indicator: {indicator}",
        data={
            "indicator": indicator,
            "type": indicator_type,
            "malicious": is_malicious,
            "confidence": 0.95 if is_malicious else 0.1
        },
        meta={"source": "threat_intel_db"}
    )

# Sub-agent for extraction
extractor = create_sub_agent(
    name="indicator_extractor",
    prompt="Extract all IPs, domains, URLs, and hashes from text. Return as list.",
)

# Main threat intelligence agent
threat_agent = create_agent(
    name="threat_analyzer",
    instruction="Analyze text for security threats. Extract indicators, check reputation, assess threat level.",
    tools=[lookup_indicator],
    subagents=[extractor],
    output_schema=ThreatAnalysis
)

async def main():
    email_text = """
    Please click this link to verify your account:
    http://evil-phishing-site.com/malware

    Contact us at: suspicious-domain.com
    """

    result = await threat_agent.run(email_text)
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

🔧 Configuration

Environment Variables

# Google Cloud (Required)
GOOGLE_APPLICATION_CREDENTIALS=/path/to/key.json
DATABASE_URL=sqlite+aiosqlite:///broadie-agent.sqlite3
# or your PostgreSQL URL postgresql://user:pass@localhost/broadie"

# LangSmith Tracing (Optional - for observability)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your-langsmith-api-key
LANGCHAIN_PROJECT=broadie

# Slack Notifications (Optional)
SLACK_BOT_TOKEN=xoxb-your-token
SLACK_SIGNING_SECRET=your-secret

# Email Notifications (Optional)
SMTP_HOST=smtp.sendgrid.net
SMTP_PORT=587
SMTP_USE_TLS=true
SMTP_USERNAME=apikey
SMTP_PASSWORD=your-sendgrid-api-key
EMAIL_FROM=[e-mail]

Database Configuration

Broadie uses SQLAlchemy for persistence. Default is SQLite, but PostgreSQL is recommended for production:

from broadie import create_agent

# SQLite (default - good for development)
agent = create_agent(
    name="assistant",
    instruction="You are helpful",
    # Uses broadie-agent.sqlite3 in current directory by default
)

# PostgreSQL (recommended for production)
agent = create_agent(
    name="assistant",
    instruction="You are helpful"
)

📡 Channels and Notifications

Agents can send notifications to Slack or email when tasks complete:

agent = create_agent(
    name="security_monitor",
    instruction="Monitor for security threats",
    channels=[
        {
            "type": "slack",
            "target": "#security-alerts",
            "instructions": "Format as Slack blocks with severity"
        },
        {
            "type": "email",
            "target": "[e-mail]",
            "instructions": "Send detailed email with all findings"
        }
    ]
)

Make sure to set the required environment variables (see Configuration section).

📊 Observability with LangSmith

Monitor agent performance and debug issues with LangSmith integration:

1. Sign up at smith.langchain.com (free tier available)
2. Create a project and get your API key
3. Set environment variables:

   export LANGCHAIN_TRACING_V2=true
   export LANGCHAIN_API_KEY=your-api-key
   export LANGCHAIN_PROJECT=broadie
   

All agent runs will be traced and visible in the LangSmith dashboard with:

• Detailed execution traces
• Tool calls and responses
• Token usage and costs
• Performance metrics
• Error debugging

🛠️ CLI Reference

Chat Command

# Interactive chat session
broadie chat my_agent.py:agent

# Chat with specific thread ID
broadie chat my_agent.py:agent --thread user123

Serve Command

# Serve agent as HTTP API
broadie serve my_agent.py:agent --port 8000

# Serve on all interfaces
broadie serve my_agent.py:agent --host 0.0.0.0 --port 8000

# Multiple workers for production
broadie serve my_agent.py:agent --port 8000 --workers 4

🚀 Deployment Best Practices

Development

# Use SQLite for local development
python my_agent.py

# Interactive testing
broadie chat my_agent.py:agent

# Serve API locally
broadie serve my_agent.py:agent --port 8000

Production

# Use PostgreSQL for persistence
export DATABASE_URL=postgresql://user:pass@db:5432/broadie

# Enable tracing
export LANGCHAIN_TRACING_V2=true

# Serve on all interfaces with multiple workers
broadie serve my_agent.py:agent --host 0.0.0.0 --port 8000 --workers 4

Docker Deployment

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY my_agent.py .
COPY key.json /secrets/

ENV GOOGLE_APPLICATION_CREDENTIALS=/secrets/key.json

CMD ["broadie", "serve", "my_agent.py:agent", "--host", "0.0.0.0", "--port", "8000"]

✨ Key Features Summary

• Structured Outputs - Pydantic schemas enforce contracts and eliminate hallucination
• ToolResponse Pattern - Type-safe tool outputs that LLMs cannot modify
• Built-in Persistence - SQLAlchemy-backed conversation history
• Thread Management - Automatic context handling across conversations
• Sub-Agents - Delegate specialized tasks to focused agents
• Channels - Built-in Slack and email notifications
• CLI Tools - Interactive chat and API serving
• Observability - LangSmith integration for tracing and debugging
• Production Ready - Database migrations, health checks, error handling

📞 Support

For questions, issues, or contributions:

Email: [e-mail]

🤝 Contributing

Broadie is developed by the Broad Institute. We welcome contributions!

---

Built with ❤️ by the Broad Institute