Integrating Open Source LLMs

Complete training — Reference guide with code examples, diagrams, and best practices
Covers all three modules: security & reliability, RAG & Agents SDK, conversation management

Table of Contents

1. Module 1 — Reliability and Security in LLM Integration

   • 1.1 — Overview: Integrating an LLM
   • 1.2 — Environment Setup and Configuration
   • 1.3 — Credential Management and Authentication
   • 1.4 — LLM Comparison: Generation Tasks
   • 1.5 — API Usage, Rate Limits and Error Codes

2. Module 2 — RAG: Customizing and Enhancing Outputs

   • 2.1 — OpenAI Agents SDK: Introduction (Part 1/2)
   • 2.2 — OpenAI Agents SDK: Full Workflow (Part 2/2)
   • 2.3 — Leveraging LLM Power with RAG
   • 2.4 — Creating a Vector Store
   • 2.5 — Optimizing Model Outputs
   • 2.6 — Tracing LLM Outputs and Usage Monitoring

3. Module 3 — Conversation State Management

   • 3.1 — Adding a Moderation Filter
   • 3.2 — History Management (Session)

4. Final Project Architecture

5. Quick Reference

Module 1

Module 1 — Reliability and Security in LLM Integration

Language models have the ability to understand and reason from human input to generate content from a simple natural language prompt. They are very large deep learning models pre-trained on a vast amount of data from different sources: websites, online books, research articles, and even code repositories.

graph TD
    A[Training Data] --> B[Pre-trained LLM]
    A --> |Websites| B
    A --> |Online books| B
    A --> |Research articles| B
    A --> |Code repositories| B
    B --> C[Text generation]
    B --> D[Summarization]
    B --> E[Code generation]
    B --> F[Chatbot / Assistant]

Open Source LLM Limitations

• Are not connected to the internet
• Have no real-time awareness
• Require advanced techniques (RAG, fine-tuning) to extend their capabilities

What We Build in This Training

graph LR
    U[User] --> PM[Project Manager\nOrchestrator]
    PM --> D[Developer\nAgent]
    PM --> CR[Code Reviewer\nAgent]
    PM --> LD[Lead Developer\nAgent]
    D --> LLM1[OpenAI GPT]
    CR --> LLM1
    LD --> LLM1
    CR --> VS[Vector Store\nRAG]
    PM --> MOD[Moderation\nFilter]
    PM --> SES[SQLite\nSession]

1.1 — Overview: Integrating an LLM

As an AI engineer, your role is to integrate an open source LLM into a client-facing application. The goal is to ensure that AI responses are safe, reliable, and efficient, while maintaining scalable API usage.

The Three Providers Compared

graph LR
    subgraph OpenAI [" OpenAI (2015)"]
        O1[ChatGPT]
        O2[GPT-3.5-turbo]
        O3[GPT-4.1]
    end
    subgraph Anthropic [" Anthropic (2021)"]
        A1[Claude AI]
        A2[claude-sonnet-4]
    end
    subgraph Mistral [" Mistral (2023)"]
        M1[mistral-large-latest]
        M2[Open-weight models]
    end
    OpenAI --> |API| App[Application]
    Anthropic --> |API| App
    Mistral --> |API| App

Typical use cases:

• OpenAI: Chatbots, virtual assistants, customer support automation
• Anthropic: Research, AI safety experiments, content generation with guardrails
• Mistral: Applications requiring flexibility, transparency, and control

1.2 — Environment Setup and Configuration

Setup Steps

┌─────────────────────────────────────────────────┐
│             SETUP STEPS                         │
│                                                 │
│  1. Install Python                              │
│  2. Create a virtual environment (.venv)        │
│  3. Install packages (requirements.txt)         │
│  4. Create API keys                             │
│  5. Configure environment variables             │
└─────────────────────────────────────────────────┘

Installing Python

macOS:

# Check if Python is installed
python3 --version

# Install Homebrew if needed
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

# Install Python
brew install python3

Windows:

# Download from https://www.python.org/downloads/
# Check "Add Python to PATH" during installation

python --version

Virtual Environment

macOS:

# Navigate to the project
cd /path/to/your/project

# Create the virtual environment
python3 -m venv .venv

# Activate the virtual environment
source .venv/bin/activate

# Deactivate
deactivate

Windows:

# Navigate to the project
cd C:\path\to\your\project

# Create the virtual environment
python -m venv .venv

# Activate the virtual environment
.venv\Scripts\activate

# Deactivate
deactivate

Installing Packages

macOS:

pip3 install -r requirements.txt

Windows:

pip install -r requirements.txt

requirements.txt File

python-dotenv==1.1.0
rich>=13.5.0
colorama==0.4.6
openai>=1.8.0             # OpenAI API
anthropic>=0.0.1          # Anthropic Claude API
mistralai>=0.2.0          # Mistral AI SDK

Package descriptions:

• python-dotenv: Loads environment variables from a .env file
• rich: Visually rich and colored output in the terminal
• colorama: Color support in console output
• openai: Official OpenAI client
• anthropic: Official Anthropic (Claude) client
• mistralai: Official Mistral AI client

Running the Application

macOS:

python3 main.py

Windows:

python main.py

1.3 — Credential Management and Authentication

Most LLMs require authentication via a secret key to access their APIs.

sequenceDiagram
    participant Dev as Developer
    participant Provider as API Provider
    participant App as Application

    Dev->>Provider: Create an account
    Provider-->>Dev: Account activated
    Dev->>Provider: Generate secret API key
    Provider-->>Dev: API key (copy it NOW)
    Dev->>App: Add key to .env
    App->>Provider: API request + secret key
    Provider-->>App: LLM response

Creating API Keys

⚠️ Important: Once the key is generated, copy it immediately. It will not be accessible again.

💳 Billing: You need to provide payment information and purchase credits to access the APIs in production.

.env.example File

OPENAI_API_KEY="YOUR_OPENAI_API_KEY"
ANTHROPIC_API_KEY="YOUR_ANTHROPIC_API_KEY"
MISTRAL_API_KEY="YOUR_MISTRAL_API_KEY"

Rename this file to .env and replace the values with your actual API keys.

API Key Verification — main.py (exercise 1.1)

import os
from colorama import Fore
from dotenv import load_dotenv

from openai import OpenAI
import anthropic
from mistralai import Mistral

from rich.console import Console
from rich.pretty import Pretty

load_dotenv()

console = Console()
console.rule("[bold blue]Pretty Object")

SYSTEM_PROMPT = "You are a skilled web development assistant who generates and explains coding tasks clearly in 1 sentence."
PROMPT = "Explain large language models pretrained vs fine-tuned in 1 sentence."

client_openai = OpenAI()
client_anthropic = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
client_mistral = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))

print(f"{Fore.YELLOW}OpenAI secret key is {os.getenv('OPENAI_API_KEY')}!{Fore.RESET}")
print(f"{Fore.GREEN}Anthropic secret key is {os.getenv('ANTHROPIC_API_KEY')}!{Fore.RESET}")
print(f"{Fore.BLUE}Mistral secret key is {os.getenv('MISTRAL_API_KEY')}!{Fore.RESET}")

1.4 — LLM Comparison: Generation Tasks

Sending Requests to All Three Providers — main.py (exercise 1.2)

import os
from colorama import Fore
from dotenv import load_dotenv

from openai import OpenAI
import anthropic
from mistralai import Mistral

from rich.console import Console
from rich.pretty import Pretty

load_dotenv()

console = Console()

SYSTEM_PROMPT = "You are a skilled web development assistant who generates and explains coding tasks clearly in 1 sentence."
PROMPT = "Explain large language models pretrained vs fine-tuned in 1 sentence."

client_openai = OpenAI()
client_anthropic = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
client_mistral = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))

# ── OpenAI ──────────────────────────────────────────────────────────────────
completion = client_openai.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user",   "content": PROMPT}
    ],
    max_tokens=1000,
)

print(Fore.BLUE + "\nOpenAI LLM Output" + Fore.RESET)
print(completion.choices[0].message.content)

# ── Anthropic (Claude) ───────────────────────────────────────────────────────
response = client_anthropic.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": PROMPT}]
)

print(Fore.CYAN + "\nAnthropic LLM Output" + Fore.RESET)
print(response.content[0].text)

# ── Mistral ───────────────────────────────────────────────────────────────────
print(Fore.MAGENTA + "\nMistral LLM Output" + Fore.RESET)
chat_response = client_mistral.chat.complete(
    model="mistral-large-latest",
    messages=[{"role": "user", "content": PROMPT}]
)
print(chat_response.choices[0].message.content)

Inspecting Response Objects — main.py (exercise 1.3)

To debug and understand the structure of responses, display the full object with rich.pretty:

# Display the full OpenAI response (enriched format)
print(Fore.BLUE + "\nOpenAI LLM Output" + Fore.RESET)
print(completion.choices[0].message.content)
console.print(Pretty(completion))   # ← Formatted display of the full object

# Display the full Anthropic response
print(Fore.CYAN + "\nAnthropic LLM Output" + Fore.RESET)
print(response.content[0].text)
console.print(Pretty(response))     # ← Formatted display of the full object

# Display the full Mistral response
print(Fore.MAGENTA + "\nMistral LLM Output" + Fore.RESET)
print(chat_response.choices[0].message.content)
console.print(Pretty(chat_response)) # ← Formatted display of the full object

API Structure Comparison

┌──────────────────────────────────────────────────────────────────────┐
│                       API COMPARISON                                 │
├────────────────┬────────────────┬────────────────┬───────────────────┤
│  Parameter     │   OpenAI       │  Anthropic     │   Mistral         │
├────────────────┼────────────────┼────────────────┼───────────────────┤
│ Method         │ chat.          │ messages.      │ chat.complete()   │
│                │ completions.   │ create()       │                   │
│                │ create()       │                │                   │
├────────────────┼────────────────┼────────────────┼───────────────────┤
│ Model          │ gpt-3.5-turbo  │ claude-sonnet- │ mistral-large-    │
│                │ gpt-4.1        │ 4-20250514     │ latest            │
├────────────────┼────────────────┼────────────────┼───────────────────┤
│ Text retrieval │ completion.    │ response.      │ chat_response.    │
│                │ choices[0].    │ content[0].    │ choices[0].       │
│                │ message.       │ text           │ message.content   │
│                │ content        │                │                   │
├────────────────┼────────────────┼────────────────┼───────────────────┤
│ System role    │ role: "system" │ N/A            │ N/A               │
│                │ in messages    │ (separate)     │                   │
└────────────────┴────────────────┴────────────────┴───────────────────┘

1.5 — API Usage, Rate Limits and Error Codes

Why Rate Limits?

Rate limits are implemented to:

• Protect against API abuse
• Avoid server overload
• Guarantee fair access to all users

How Rate Limits Work

graph TD
    A[API Request] --> B{RPM limit\nreached?}
    B -->|No| C{TPM limit\nreached?}
    C -->|No| D[Request processed ✅]
    C -->|Yes| E[Error 429\nRate Limit Exceeded ❌]
    B -->|Yes| E

• RPM: Requests Per Minute
• TPM: Tokens Per Minute

⚠️ The first limit hit (RPM or TPM) cuts off API access.

Common Error Codes

Inspecting Token Usage

from rich.console import Console
from rich.pretty import Pretty

console = Console()

# Display the full response to see token usage
console.print(Pretty(completion))    # OpenAI
console.print(Pretty(response))      # Anthropic
console.print(Pretty(chat_response)) # Mistral

Best Practices

• Limit the max_tokens parameter to control costs
• Implement a retry system with exponential backoff
• Monitor usage via provider dashboards
• Use gpt-3.5-turbo (less expensive) for testing

Module 2

Module 2 — RAG: Customizing and Enhancing Outputs

LLMs have transformed how developers build intelligent software. The OpenAI Agents SDK enables building multi-agent systems capable of performing professional tasks comparable to those performed by humans.

Key SDK Features

mindmap
  root((OpenAI Agents SDK))
    Agent Loop
      Workflow automation
      LLM connection
      Result processing
    Handoffs
      Task transfer
      Specialized agents
      Data sharing
    Guardrails
      Output validation
      Expectation compliance
    Session Management
      Interaction history
      Preserved context
      SQLite backend
    Function Tools
      External APIs
      Backends
      Real-time data
    RAG
      Vector Store
      Semantic search
      Augmented context
    Tracing
      Debugging
      Monitoring
      Observability

2.1 — OpenAI Agents SDK: Introduction (Part 1/2)

Agent Architecture

graph TD
    User[User] --> PM[project_manager_orchestrator]
    
    PM -->|developer_tool| Dev[Developer Agent]
    PM -->|code_reviewer_tool| CR[Code Reviewer Agent]
    PM -->|lead_developer_tool| LD[Lead Developer Agent]
    
    Dev -->|handoff| CR
    CR -->|handoff| LD
    
    Dev --> GPT[OpenAI GPT]
    CR --> GPT
    LD --> GPT

    style PM fill:#f9f,stroke:#333
    style Dev fill:#bbf,stroke:#333
    style CR fill:#bfb,stroke:#333
    style LD fill:#fbf,stroke:#333

Agent Instructions — agents-instructions.txt

Agent: Developer
─────────────────────────────────────────────────────────────
You are a skilled software developer. You write code
based on the project guidelines and the client's request,
help with feature development, and fix bugs.

Agent: Developer and Reviewer
─────────────────────────────────────────────────────────────
You specialize in peer programming and code review.
Your role is to review code written by other developers,
provide constructive feedback, and suggest improvements
based on coding standards and best practices.
Handoff: When the code review is done, transfer to the
lead developer for final approval.

Agent: Lead Developer
─────────────────────────────────────────────────────────────
You are the lead developer. Oversee the project and guide
other developers. You give final approval before code
is merged into the main codebase.

Agent: project_manager_orchestrator
─────────────────────────────────────────────────────────────
You are a project manager agent. You use the provided
tools to run code writing, review, and approval.
You ALWAYS use the provided tools, never answer directly.

Basic Implementation — main.py (exercise 2.1)

import asyncio
from colorama import Fore
from dotenv import load_dotenv

from openai import OpenAI
from agents import Agent, ItemHelpers, MessageOutputItem, Runner, trace

from rich.console import Console

load_dotenv()

console = Console()
client_openai = OpenAI()

# ── Agent definitions ────────────────────────────────────────────────────
developer = Agent(
    name="Developer",
    instructions="You are a skilled software developer. You write code based on the "
                 "Project Guidelines and Client's request, assist with software features "
                 "development, and fix bugs.",
    handoff_description="When the code task is complete, hand off to code reviewer.",
)

code_reviewer = Agent(
    name="Developer and Reviewer",
    instructions="You specialize in peer programming and code review. Your role is to "
                 "review code written by other developers, provide constructive feedback, "
                 "and suggest improvements based on the coding standards, project "
                 "guidelines and best practices.",
    handoff_description="When the code review is complete, hand off to the lead developer "
                        "for final approval.",
)

lead_developer = Agent(
    name="Lead Developer",
    instructions="You are the lead developer. Oversee the project and provide guidance "
                 "to other developers based on the code review checklist and project "
                 "guidelines. You give final approval on code before it is merged into "
                 "the main codebase.",
)

# ── Orchestrator (Project Manager) ─────────────────────────────────────────
project_manager_orchestrator = Agent(
    name="project_manager_orchestrator",
    instructions=(
        "You are a project manager agent. You use the tools given to you to run code "
        "writing, review and approval. You always use the provided tools for the tasks, "
        "never answer directly."
    ),
    tools=[
        developer.as_tool(
            tool_name="developer_tool",
            tool_description="A tool to write code, assist with software features "
                             "development, and fix bugs.",
        ),
        code_reviewer.as_tool(
            tool_name="code_reviewer_tool",
            tool_description="A tool to assist with code review and provide feedback.",
        ),
        lead_developer.as_tool(
            tool_name="lead_developer_tool",
            tool_description="A tool to oversee the project, provide guidance to the "
                             "development team and approve code.",
        ),
    ],
)

console.rule("[bold Green]START OF CHAT")

async def main():
    msg = input("Hi! What feature would you want to develop? ")

    with trace("Orchestrator evaluator"):
        orchestrator_result = Runner.run_streamed(project_manager_orchestrator, input=msg)

        async for event in orchestrator_result.stream_events():
            if event.type == "raw_response_event":
                continue
            elif event.type == "agent_updated_stream_event":
                print(Fore.MAGENTA + f"Agent updated: {event.new_agent.name}" + Fore.RESET)
            elif event.type == "run_item_stream_event":
                if event.item.type == "tool_call_item":
                    print(Fore.YELLOW + "-- Tool was called" + Fore.RESET)
                elif event.item.type == "tool_call_output_item":
                    print(Fore.YELLOW + f"-- Tool output: {event.item.output}" + Fore.RESET)
                elif event.item.type == "message_output_item":
                    print(Fore.YELLOW + f"-- Message output:\n "
                          f"{ItemHelpers.text_message_output(event.item)}" + Fore.RESET)

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

2.2 — OpenAI Agents SDK: Full Workflow (Part 2/2)

Complete Workflow with Streaming

The full workflow runs the orchestrator in streaming mode, then waits for the final response:

async def main():
    msg = input("Hi! What feature would you want to develop? ")

    with trace("Orchestrator Code Review Process"):
        # Phase 1: Streaming execution
        orchestrator_result = Runner.run_streamed(project_manager_orchestrator, input=msg)

        async for event in orchestrator_result.stream_events():
            if event.type == "raw_response_event":
                continue
            elif event.type == "agent_updated_stream_event":
                print(Fore.MAGENTA + f"Agent updated: {event.new_agent.name}" + Fore.RESET)
            elif event.type == "run_item_stream_event":
                if event.item.type == "tool_call_item":
                    print(Fore.YELLOW + "-- Tool was called" + Fore.RESET)
                elif event.item.type == "tool_call_output_item":
                    print(f"-- Tool output: {event.item.output}")
                elif event.item.type == "message_output_item":
                    print(f"-- Message output:\n "
                          f"{ItemHelpers.text_message_output(event.item)}")

        # Phase 2: Final result
        code_review_result = await Runner.run(
            project_manager_orchestrator, orchestrator_result.to_input_list()
        )

    print(Fore.GREEN + f"\n\n ====== Final response:\n"
          f"{code_review_result.final_output} ======" + Fore.RESET)
    console.rule("[bold Green]END OF CHAT")

Streaming Event Types

┌─────────────────────────────────────────────────────────┐
│              STREAMING EVENT TYPES                      │
│                                                         │
│  raw_response_event      → Ignore (raw deltas)          │
│  agent_updated_stream_event → Display active agent      │
│  run_item_stream_event:                                 │
│    ├── tool_call_item    → "-- Tool was called"         │
│    ├── tool_call_output_item → "-- Tool output: ..."    │
│    └── message_output_item → "-- Message output: ..."  │
└─────────────────────────────────────────────────────────┘

2.3 — Leveraging LLM Power with RAG

Why RAG?

LLMs are pre-trained on static data and do not have access to:

• Organization-specific data
• Real-time information
• Proprietary documents (coding standards, internal procedures)

RAG (Retrieval-Augmented Generation) solves this problem by allowing agents to query a vector knowledge base before responding.

graph LR
    Q[User\nquestion] --> EMB[Text\nembedding]
    EMB --> VS[Vector Store\nSemantic search]
    VS --> |Relevant documents| CTX[Augmented context]
    CTX --> LLM[LLM]
    Q --> LLM
    LLM --> R[Contextualized\nresponse]

Utility Function for Logging — utils.py

from agents import ItemHelpers
from colorama import Fore
from rich.console import Console
from rich.pretty import Pretty

console = Console()

async def log_streaming_events(result):
    """Logs streaming events from the orchestrator result."""
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            continue
        elif event.type == "agent_updated_stream_event":
            print(Fore.MAGENTA + f"Agent updated: {event.new_agent.name}" + Fore.RESET)
            continue
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print(Fore.CYAN + f"-- Tool was called: " + Fore.RESET)
            elif event.item.type == "tool_call_output_item":
                print(f"-- Tool output: {event.item.output}")
            elif event.item.type == "message_output_item":
                print(f"-- Message output:\n "
                      f"{ItemHelpers.text_message_output(event.item)}")
            else:
                pass

2.4 — Creating a Vector Store

Concept: Vector Embeddings

Vector embeddings are numerical representations of textual data (words, sentences, documents).

┌────────────────────────────────────────────────────────────────┐
│                    VECTOR EMBEDDINGS                           │
│                                                                │
│  Text → [0.23, -0.17, 0.89, 0.42, ..., -0.11]  (vector)      │
│                                                                │
│  Short distance = high semantic similarity                     │
│  Long distance  = low semantic similarity                      │
│                                                                │
│     "dog"  ●──────────●  "animal"  (close)                     │
│                                                                │
│     "dog"  ●────────────────────────●  "car"  (distant)        │
└────────────────────────────────────────────────────────────────┘

Full RAG Architecture

graph TD
    subgraph Preparation
        TXT[code-review_checklist.txt] --> PREP[Dataset\npreparation]
        PREP --> EMB[Embedding\ncreation]
        EMB --> VS[(Vector Store\nOpenAI)]
    end

    subgraph Runtime
        Q[Agent question] --> SEARCH[Semantic search]
        VS --> SEARCH
        SEARCH --> RESULTS[Relevant documents]
        RESULTS --> CONTEXT[Enriched context]
        CONTEXT --> LLM[GPT-4.1]
        LLM --> RESP[Contextualized response]
    end

Vector Store Implementation — main.py (exercise 2.3)

import asyncio
from colorama import Fore
from typing import Literal
from dotenv import load_dotenv
from dataclasses import dataclass

from openai import OpenAI
from agents import Agent, FileSearchTool, Runner, trace
from rich.console import Console
from utils import log_streaming_events

load_dotenv()
console = Console()
client = OpenAI()

# ── Step 1: Create the Vector Store ─────────────────────────────────────────
vector_store = client.vector_stores.create(name="Project Guidelines")

# ── Step 2: Load and index the guidelines file ───────────────────────────────
with open("code-review_checklist.txt", "r") as f:
    text = f.read()
    file_upload = client.files.create(
        file=("code-review_checklist.txt", text.encode("utf-8")),
        purpose="assistants",
    )

indexed = client.vector_stores.files.create_and_poll(
    vector_store_id=vector_store.id,
    file_id=file_upload.id
)

print(Fore.BLUE + f"Stored files in vector store: {indexed.to_dict()}" + Fore.RESET)

# ── Step 3: Configure agents with FileSearchTool ─────────────────────────────
developer_agent = Agent(
    name="Developer",
    instructions="You are a skilled software developer. You write code based on the "
                 "Project Guidelines and Client's request.",
)

code_reviewer_agent = Agent(
    name="Developer and Reviewer",
    instructions="You specialize in peer programming and code review. Your role is to "
                 "review code written by other developers, provide constructive feedback, "
                 "and suggest improvements based on the coding standards, project "
                 "guidelines and best practices.",
    handoff_description="When the code review is complete, hand off to the lead developer.",
    tools=[FileSearchTool(        # ← RAG: search in the vector store
        max_num_results=3,
        vector_store_ids=[vector_store.id],
        include_search_results=True
    )]
)

lead_developer_agent = Agent(
    name="Lead Developer",
    instructions="You are the lead developer. Oversee the project and provide guidance "
                 "to other developers based on the code review checklist and project "
                 "guidelines. You approve code before sending to Q&A.",
)

Code Review Checklist — code-review_checklist.txt

This file is the dataset used for RAG. It provides agents with the context needed to perform quality code reviews.

=== Checklist: Code Review Recommendations and Best Practices ===

Reviewer Feedback :

- Code Style & Standards: Code follows the project's style guide.
    ✅ Best Practice: Consistent naming (camelCase / PascalCase).
    ⚠️ Issue Example: Mixing snake_case and camelCase in the same file.

- Formatting: Indentation, spacing, and naming are consistent.
    ✅ Best Practice: Auto-formatted using Prettier/ESLint.
    ⚠️ Issue Example: Inconsistent indentation (2 vs. 4 spaces).

- Architectural Integrity: Codebase is scalable and maintainable.
    ✅ Best Practice: Separation of concerns (controllers, services, data layers).
    ⚠️ Issue Example: Business logic directly embedded in route handlers.

- Performance: Queries and logic are optimized.
    ✅ Best Practice: Proper indexing on frequently queried DB columns.
    ⚠️ Issue Example: SELECT * queries without filters.

- Security: Sensitive information not hardcoded.
    ✅ Best Practice: Passwords hashed with bcrypt or Argon2.
    ⚠️ Issue Example: JWT secret stored in plain text.

- Error Handling & Logging: Proper error handling implemented.
    ✅ Best Practice: Centralized error middleware.
    ⚠️ Issue Example: Silent failure — errors logged only to console.

- Testing: Adequate unit, integration, and/or end-to-end tests.
    ✅ Best Practice: Tests cover core business logic and edge cases.
    ⚠️ Issue Example: No tests for API error states.

=== Code Review Approval Report (Q&A) ===
✅ Approved
⚠️ Approved with Minor Changes
❌ Rejected

2.5 — Optimizing Model Outputs

LLM as a Judge — Structured Evaluation

The idea is to use an LLM to evaluate the outputs of other agents and ensure they meet expectations.

graph TD
    User[User] --> PM[Project Manager\nOrchestrator]
    PM --> Dev[Developer\nAgent]
    PM --> CR[Code Reviewer\nAgent + RAG]
    PM --> LD[Lead Developer\nAgent]
    LD -->|EvaluationFeedback| Result{Result}
    Result -->|✅ pass| Approved[Code approved]
    Result -->|⚠️ needs_improvement| Improve[Improvements required]
    Result -->|❌ fail| Rejected[Code rejected]

Structured Output Format with dataclass — main.py (exercise 2.4)

from dataclasses import dataclass
from typing import Literal

@dataclass
class EvaluationFeedback:
    feedback: str
    score: Literal["✅ pass", "⚠️ needs_improvement", "❌ fail"]

# Apply this format to the Lead Developer agent
lead_developer_agent = Agent[None](
    name="Lead Developer",
    instructions="You are the lead developer. Oversee the project and provide guidance "
                 "to other developers based on the code review checklist and project "
                 "guidelines. You approve code before sending to Q&A.",
    output_type=EvaluationFeedback,   # ← Structured output format
)

# The orchestrator includes this format in its instructions
project_manager_orchestrator = Agent(
    name="project_manager_orchestrator",
    instructions=(
        "You are a project manager agent. You use the tools given to you to run code "
        "writing, review and approval. You always use the provided tools for the tasks, "
        "never answer directly. Include the validation feedback format in the final "
        "response: Literal['✅ pass', '⚠️ needs_improvement', '❌ fail']"
    ),
    tools=[
        developer_agent.as_tool(
            tool_name="developer_tool",
            tool_description="A tool to write code, assist with software features "
                             "development, and fix bugs.",
        ),
        code_reviewer_agent.as_tool(
            tool_name="code_reviewer_tool",
            tool_description="A tool to assist with code review and provide feedback.",
        ),
        lead_developer_agent.as_tool(
            tool_name="lead_developer_tool",
            tool_description="A tool to oversee the project, provide guidance to the "
                             "development team and approve code.",
        ),
    ],
)

Running the Full Workflow

async def main():
    msg = input("Hi! What feature would you want to develop? ")

    with trace("Orchestrator Code Review Process"):
        orchestrator_result = Runner.run_streamed(project_manager_orchestrator, input=msg)
        await log_streaming_events(orchestrator_result)   # ← Utility function utils.py

        code_review_result = await Runner.run(
            project_manager_orchestrator, orchestrator_result.to_input_list()
        )

    print(Fore.GREEN + f"\n\n ====== Final response:\n"
          f"{code_review_result.final_output} ======" + Fore.RESET)
    console.rule("[bold Blue]END OF CHAT")

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

2.6 — Tracing LLM Outputs and Usage Monitoring

Available Tracing Platforms

Several options are available to observe and debug your agents:

┌──────────────────────────────────────────────────────────────┐
│                    TRACING PLATFORMS                         │
│                                                              │
│  1. OpenAI Platform Logs                                     │
│     → platform.openai.com → Logs                             │
│     → See models used, tokens, instructions                  │
│                                                              │
│  2. Logfire (Pydantic)                                       │
│     → Advanced tracing with logfire.configure()              │
│     → Custom metrics                                         │
│                                                              │
│  3. LangSmith                                                │
│  4. Arize Phoenix                                            │
│  5. Weights & Biases                                         │
└──────────────────────────────────────────────────────────────┘

Logfire Integration — main.py (exercise 2.5)

import logfire
import os

# Logfire configuration for tracing
logfire.configure(
    service_name="code review workflow",
    send_to_logfire=True,
    token=os.getenv("LOGFIRE_TOKEN")
)

logfire.info('Hello, {place}!', place='World')

Add LOGFIRE_TOKEN to your .env file to enable cloud tracing.

What You Can Observe in OpenAI Logs

• Model used (e.g.: gpt-4.1)
• Token count: input tokens, output tokens, total tokens
• Instructions given to the agent
• Outputs generated by the assistant
• Estimated cost of the request

Token Optimization

┌──────────────────────────────────────────────────────────────┐
│               OPTIMIZATION STRATEGIES                        │
│                                                              │
│  Input tokens  ──► Reduce instruction length                 │
│  Output tokens ──► Limit max_tokens in requests              │
│                                                              │
│  completion = client_openai.chat.completions.create(         │
│      model="gpt-3.5-turbo",                                  │
│      max_tokens=1000,    ← Limits output tokens              │
│      messages=[...]                                          │
│  )                                                           │
└──────────────────────────────────────────────────────────────┘

Module 3

Module 3 — Conversation State Management

In this module, we add two essential features to our multi-agent workflow:

1. Moderation: Filter content that does not comply with usage policies
2. Session history: Allow agents to remember past interactions

3.1 — Adding a Moderation Filter

Why Moderation?

Content moderation is essential for:

• Maintaining a safe and respectful environment
• Filtering content that violates usage policies
• Protecting users and the organization

graph TD
    U[User input] --> MOD{Moderation\nOpenAI API}
    MOD -->|flagged = false ✅| WF[Workflow\nexecution]
    MOD -->|flagged = true ❌| STOP[Workflow stopped\n+ Alert message]
    WF --> PM[Project Manager\nOrchestrator]
    PM --> D[Developer]
    PM --> CR[Code Reviewer]
    PM --> LD[Lead Developer]

OpenAI Moderation Endpoint

The OpenAI API provides a moderation endpoint that automatically detects problematic content and classifies it by category.

Moderation Function — utils.py

from openai import OpenAI

client = OpenAI()

async def run_moderation(text: str):
    """Check text for unsafe or disallowed content."""
    response = client.moderations.create(
        model="omni-moderation-latest",
        input=text
    )
    results = response.results[0]
    flagged = results.flagged         # True if problematic content
    categories = results.categories   # Violation categories detected
    return flagged, categories

Integration in the Workflow — main.py (exercise 3.1)

from utils import log_streaming_events, run_moderation

async def main():
    msg = input("Hi! What feature would you want to develop? ")

    # ── User input moderation ───────────────────────────────────────────────
    flagged, categories = await run_moderation(msg)
    if flagged:
        console.print(Fore.RED + "⚠️ !MODERATION ALERT! Your input is flagged" + Fore.RESET)
        console.print(f"Categories: {categories}")
        return   # ← Stop the workflow immediately
    else:
        console.print(Fore.GREEN + "✅ Your input passed moderation" + Fore.RESET)

    # ── Normal workflow execution ────────────────────────────────────────────
    with trace("Orchestrator Code Review Process"):
        orchestrator_result = Runner.run_streamed(project_manager_orchestrator, input=msg)
        await log_streaming_events(orchestrator_result)

        code_review_result = await Runner.run(
            project_manager_orchestrator, orchestrator_result.to_input_list()
        )

    print(Fore.GREEN + f"\n\n ====== Final response:\n"
          f"{code_review_result.final_output} ======" + Fore.RESET)
    console.rule("[bold Blue]END OF CHAT")

Detected Moderation Categories

┌──────────────────────────────────────────────────────────────┐
│             OPENAI MODERATION CATEGORIES                     │
│                                                              │
│  hate              - Hate speech                             │
│  hate/threatening  - Threats                                 │
│  harassment        - Harassment                              │
│  self-harm         - Self-harm                               │
│  sexual            - Sexual content                          │
│  violence          - Violence                                │
│  violence/graphic  - Graphic violence                        │
└──────────────────────────────────────────────────────────────┘

3.2 — History Management (Session)

How Session Memory Works

The Agents SDK integrates SQLite-based session management that allows agents to automatically maintain context and history.

sequenceDiagram
    participant User as User
    participant Runner as Runner
    participant SQLite as SQLiteSession
    participant Agent as Agent

    User->>Runner: run(input, session=session)
    Runner->>SQLite: Retrieve history (session_id)
    SQLite-->>Runner: Previous messages
    Runner->>Agent: Input + history
    Agent-->>Runner: New response
    Runner->>SQLite: Store new response
    Note over SQLite: Context preserved for the next run

How It Works

┌──────────────────────────────────────────────────────────────┐
│             SQLITE SESSION LIFECYCLE                         │
│                                                              │
│  1. Before each run:                                         │
│     Runner retrieves history from SQLite                     │
│     → Attached to input items                                │
│                                                              │
│  2. During the run:                                          │
│     The agent processes input with historical context        │
│                                                              │
│  3. After each run:                                          │
│     New items are automatically stored                       │
│     → Context preserved for the next run                     │
└──────────────────────────────────────────────────────────────┘

SQLite Session Implementation — main.py (exercise 3.2)

import asyncio
from colorama import Fore
from typing import Literal
from dotenv import load_dotenv
from dataclasses import dataclass

from openai import OpenAI
from agents import Agent, FileSearchTool, Runner, trace, SQLiteSession
from rich.console import Console
from utils import log_streaming_events, run_moderation

load_dotenv()
console = Console()
client = OpenAI()

# ── Create the SQLite session ────────────────────────────────────────────────
session = SQLiteSession("code_review_process_123")   # ← Unique session ID

# [... vector store and agent creation as before ...]

async def main():
    msg = input("Hi! What feature would you want to develop? ")

    # Input moderation
    flagged, categories = await run_moderation(msg)
    if flagged:
        console.print(Fore.RED + "⚠️ !MODERATION ALERT! Your input is flagged" + Fore.RESET)
        console.print(f"Categories: {categories}")
        return
    else:
        console.print(Fore.GREEN + "✅ Your input passed moderation" + Fore.RESET)

    with trace("Orchestrator Code Review Process"):
        orchestrator_result = Runner.run_streamed(project_manager_orchestrator, input=msg)
        await log_streaming_events(orchestrator_result)

        # ── Attach the session to preserve context ───────────────────────────
        code_review_result = await Runner.run(
            project_manager_orchestrator,
            input=str(orchestrator_result.final_output),
            session=session     # ← The session preserves history
        )

    print(Fore.GREEN + f"\n\n ====== Final response:\n"
          f"{code_review_result.final_output} ======" + Fore.RESET)
    console.rule("[bold Blue]END OF CHAT")

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

Contextual Memory Demonstration

Test scenario:

The SQLite session preserves context without having to specify it explicitly at each request.

Final Project Architecture

Complete System Overview

graph TD
    subgraph Input
        U[User] --> MSG[Message]
    end

    subgraph Security
        MSG --> MOD[Moderation\nomni-moderation-latest]
        MOD -->|❌ Flagged| STOP[Stop]
        MOD -->|✅ OK| WF
    end

    subgraph Multi-Agent Workflow
        WF[Workflow] --> PM[project_manager_orchestrator]
        PM -->|developer_tool| DEV[Developer\nAgent]
        PM -->|code_reviewer_tool| CR[Code Reviewer\nAgent]
        PM -->|lead_developer_tool| LD[Lead Developer\nAgent]
    end

    subgraph RAG Context
        CR --> FS[FileSearchTool]
        FS --> VS[(Vector Store\ncode-review_checklist.txt)]
    end

    subgraph Evaluation
        LD --> EF[EvaluationFeedback\n✅ pass / ⚠️ needs_improvement / ❌ fail]
    end

    subgraph Memory
        WF --> SES[(SQLiteSession\nConversation History)]
        SES --> WF
    end

    subgraph Observability
        WF --> TRACE[Tracing\nOpenAI Logs / Logfire]
    end

    EF --> OUT[Final output]

    style STOP fill:#f88,stroke:#c00
    style MOD fill:#ff9,stroke:#990
    style VS fill:#9cf,stroke:#06c
    style SES fill:#9cf,stroke:#06c
    style EF fill:#9f9,stroke:#090

Exercise Progression

┌────────────────────────────────────────────────────────────────────────────┐
│                        EXERCISE PROGRESSION                                │
├────────┬──────────────┬────────────────────────────────────────────────────┤
│ Ex     │ Folder       │ Feature added                                      │
├────────┼──────────────┼────────────────────────────────────────────────────┤
│ 1.1    │ 01/1/1.1     │ API key verification                               │
│ 1.2    │ 01/1/1.2     │ Requests to all 3 LLMs (OpenAI, Anthropic, Mistral)│
│ 1.3    │ 01/1/1.3     │ Response object inspection (rich.pretty)           │
│ 2.1    │ 01/2/2.1     │ Agents SDK: basic workflow with streaming          │
│ 2.2    │ 01/2/2.2     │ Full workflow with final result                    │
│ 2.3    │ 01/2/2.3     │ RAG: Vector Store + FileSearchTool                 │
│ 2.4    │ 01/2/2.4     │ Structured evaluation (EvaluationFeedback)         │
│ 2.5    │ 01/2/2.5     │ Tracing with Logfire                               │
│ 3.1    │ 01/3/3.1     │ User input moderation                              │
│ 3.2    │ 01/3/3.2     │ SQLite session (conversation history)              │
└────────┴──────────────┴────────────────────────────────────────────────────┘

Quick Reference

Project File Structure

project/
├── main.py                    # Main entry point
├── utils.py                   # Utility functions (logging, moderation)
├── code-review_checklist.txt  # RAG dataset
├── agents-instructions.txt    # Agent instructions (reference)
├── requirements.txt           # Python dependencies
├── .env                       # API keys (do NOT commit!)
└── .env.example               # .env template file

Environment Variables

OPENAI_API_KEY="sk-..."
ANTHROPIC_API_KEY="sk-ant-..."
MISTRAL_API_KEY="..."
LOGFIRE_TOKEN="..."       # Optional (for Logfire tracing)

Recommended Models

Deployment Checklist

☐ API keys are in .env (never in code)
☐ .env is in .gitignore
☐ requirements.txt is up to date
☐ Vector store is created and indexed
☐ Moderation is enabled on user inputs
☐ Tracing is configured for monitoring
☐ Rate limits are handled (max_tokens defined)
☐ SQLite session is configured for persistence

Full Import List for the Final Project

import asyncio
import os
from colorama import Fore
from typing import Literal
from dotenv import load_dotenv
from dataclasses import dataclass

from openai import OpenAI
from agents import (
    Agent,
    FileSearchTool,
    Runner,
    trace,
    SQLiteSession,
    ItemHelpers
)
from rich.console import Console
from rich.pretty import Pretty

Additional Resources:

Search Terms

integrating · open · source · llms · llm · application · development · artificial · intelligence · generative · ai · main.py · moderation · api · openai · workflow · architecture · environment · limits · rag · rate · sdk · session · usage