Building a Single-Purpose LangGraph Agent with FastAPI and Streamlit

-

This article shows how to build a compact, single-purpose agent stack in Python. The agent’s role and tools are defined in a configuration file, the reasoning core runs on LangGraph, a FastAPI service exposes a clean JSON API, and a Streamlit page provides a minimal “ask a question → get an answer” console.

1. High-Level Architecture

The system is organised into three layers:

  • LangGraph Core – connects to Groq or OpenAI, attaches tools (web search, plus placeholders for vector search and database queries), enforces guardrails, and returns a final answer string.
  • FastAPI Service – validates incoming requests, selects the correct model/provider, calls the core agent, and returns a JSON response.
  • Streamlit Console – a focused UI that lets the user ask a question and see the answer. The agent’s role is fixed by configuration; there is no editable system-prompt field and no conversation history.
User in Streamlit
    ↓
Streamlit console (frontend.py)
    • User types a question
    • UI sends JSON to POST /chat

FastAPI service (backend.py)
    • Validates JSON using ChatRequest (Pydantic)
    • Confirms model is in SUPPORTED_MODELS
    • Calls run_graph_agent(...)

LangGraph core (ai_agent.py)
    • Loads role and tool settings from a config file
    • Selects Groq or OpenAI client
    • Attaches Tavily search + optional vector/db placeholders
    • Runs create_react_agent(...)
    • Applies output guardrails and returns safe text

Response flows back:
LangGraph → FastAPI → Streamlit → shown to the user

Before wiring the Streamlit UI, you can test the backend directly via the automatically generated FastAPI docs at /docs. This is useful to confirm your JSON payloads and responses look correct.

2. Phase 1 – LangGraph Core (ai_agent.py)

2.1 Role of the core layer

The core layer is responsible for three things:

  1. Loading the agent role and tool configuration from a file.
  2. Building a LangGraph ReAct agent with the requested model and tools.
  3. Applying simple guardrails to the answer before returning it as text.

2.2 Core flow

run_graph_agent(...)
    ↓
load agent config (role + tool toggles)
    ↓
select Groq or OpenAI client based on provider
    ↓
assemble tool list:
    • web search (Tavily)
    • optional vector search placeholder
    • optional database query placeholder
    ↓
build ReAct graph with create_react_agent(...)
    ↓
invoke agent with {"messages": messages}
    ↓
extract assistant messages and apply output guardrails
    ↓
return final safe answer text

2.3 Imports, configuration, and placeholders

import json
import os
from pathlib import Path
from typing import List, Optional

from langchain_groq import ChatGroq
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent


# -------- agent configuration --------

DEFAULT_CONFIG = {
    "system_prompt": (
        "You are a domain-specific assistant. "
        "Answer with clear, structured explanations that stay on topic."
    ),
    "enable_web_search": True,
    "enable_vector_search": False,
    "enable_database_query": False,
}


def load_agent_config() -> dict:
    """Read agent configuration from JSON; fall back to defaults if missing."""
    config_path = Path(
        os.getenv("AGENT_CONFIG_PATH", "config/agent_profile.json")
    )
    if not config_path.exists():
        return DEFAULT_CONFIG.copy()

    try:
        return json.loads(config_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        # If the file is present but invalid, do not crash the service.
        return DEFAULT_CONFIG.copy()


AGENT_CONFIG = load_agent_config()


# -------- tool placeholders --------

@tool
def vector_search_tool(query: str) -> str:
    """
    Placeholder tool for vector search over your own documents.

    Replace the body of this function with a real call to your vector store
    (for example, a similarity search over embeddings in a database).
    """
    return (
        "Vector search placeholder reached. "
        "Integrate your document store here."
    )


@tool
def database_query_tool(sql: str) -> str:
    """
    Placeholder tool for read-only database queries.

    Replace the body of this function with a safe query runner that only
    allows parameterised, read-only SQL statements.
    """
    return (
        "Database query placeholder reached. "
        "Wire this up to your reporting database with proper safeguards."
    )


# -------- simple output guardrails --------

FORBIDDEN_SNIPPETS = ["DROP TABLE", "FORMAT DISK"]


def apply_output_guardrails(text: str) -> str:
    """
    Apply basic safety checks to the agent's answer.

    This example trims extremely long outputs and masks a few hard-coded
    patterns. In a production system you would do more comprehensive checks.
    """
    cleaned = text.strip()

    # Limit very large responses
    if len(cleaned) > 8000:
        cleaned = cleaned[:8000] + "\n\n[Answer truncated for length.]"

    # Mask a couple of dangerous phrases
    for snippet in FORBIDDEN_SNIPPETS:
        if snippet in cleaned:
            cleaned = cleaned.replace(snippet, "[redacted]")

    return cleaned

Why this structure is used

  • The configuration file defines the agent’s role and which tools are enabled without changing code.
  • The vector and database tools are clearly marked placeholders, encouraging readers to plug in their own logic.
  • The guardrail function centralises simple safety checks; it can later be expanded into a richer validation step.

2.4 Core agent implementation

def run_graph_agent(
    model_name: str,
    messages: List[str],
    provider: str,
    allow_search: bool,
    system_prompt_override: Optional[str] = None,
) -> str:
    """
    Run the LangGraph ReAct agent and return a guarded answer string.
    """

    provider_lower = provider.lower()

    if provider_lower == "groq":
        llm = ChatGroq(model=model_name)
    elif provider_lower == "openai":
        llm = ChatOpenAI(model=model_name)
    else:
        raise ValueError(f"Unsupported provider: {provider}")

    tool_list = []

    # Web search is allowed only if both config and request permit it.
    if AGENT_CONFIG.get("enable_web_search", True) and allow_search:
        tool_list.append(TavilySearchResults(max_results=2))

    if AGENT_CONFIG.get("enable_vector_search", False):
        tool_list.append(vector_search_tool)

    if AGENT_CONFIG.get("enable_database_query", False):
        tool_list.append(database_query_tool)

    system_prompt = (
        system_prompt_override
        if system_prompt_override is not None
        else AGENT_CONFIG.get("system_prompt", DEFAULT_CONFIG["system_prompt"])
    )

    agent = create_react_agent(
        model=llm,
        tools=tool_list,
        state_modifier=system_prompt,
    )

    state = {"messages": messages}
    result_state = agent.invoke(state)

    assistant_messages = [
        msg
        for msg in result_state.get("messages", [])
        if isinstance(msg, AIMessage)
    ]

    if not assistant_messages:
        return ""

    raw_answer = assistant_messages[-1].content
    safe_answer = apply_output_guardrails(raw_answer)
    return safe_answer

Key points in this function

  • The configuration file decides which tools are available; the request flag only controls web search.
  • The system prompt normally comes from configuration, but an explicit override parameter is available if needed for testing.
  • The graph structure is simple (no router or planner nodes) to keep the behaviour easy to reason about.
  • Guardrails are applied to the final answer only, keeping all intermediate reasoning internal to LangGraph.

3. Phase 2 – FastAPI Service (backend.py)

3.1 What the service does

The FastAPI layer receives a JSON payload, validates it against a Pydantic model, checks that the requested model is allowed, calls the core agent, and sends back a JSON response. This is the same endpoint used by Streamlit and by the Swagger UI.

3.2 Service flow

POST /chat
    ↓
FastAPI parses body into ChatRequest
    ↓
Check that model is in SUPPORTED_MODELS
    ↓
Call run_graph_agent(...)
    ↓
Return {"answer": "..."} or {"error": "..."}

3.3 Data model and imports

from typing import List, Literal

from fastapi import FastAPI
from pydantic import BaseModel

from ai_agent import run_graph_agent


class ChatRequest(BaseModel):
    model: str
    provider: Literal["Groq", "OpenAI"]
    messages: List[str]
    enable_search: bool = False

Why this shape is chosen

  • The provider field is constrained to the two expected values, which keeps error handling simple.
  • The system prompt is not exposed here because the agent’s behaviour is fixed by configuration.
  • The messages list allows you to move to multi-turn conversations later, even though this article uses single questions.

3.4 Endpoint logic

SUPPORTED_MODELS = {
    "llama-3.3-70b-versatile",
    "mixtral-8x7b-32768",
    "llama3-70b-8192",
    "gpt-4o-mini",
}

app = FastAPI(title="Single-Purpose LangGraph Agent")


@app.post("/chat")
def chat_with_agent(request: ChatRequest):
    """
    Forward chat requests to the LangGraph agent and return its answer.
    """

    if request.model not in SUPPORTED_MODELS:
        return {"error": "Requested model is not enabled on this service."}

    reply = run_graph_agent(
        model_name=request.model,
        messages=request.messages,
        provider=request.provider,
        allow_search=request.enable_search,
    )

    return {"answer": reply}

3.5 Running the service and using Swagger

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=9999)

Once this file is running, visit http://127.0.0.1:9999/docs to open the interactive API documentation. You can select the /chat endpoint, click “Try it out”, fill in a JSON payload, and verify the agent response directly from the browser before connecting the Streamlit UI.

4. Phase 3 – Streamlit Console (frontend.py)

4.1 Purpose of the UI

The Streamlit page is intentionally minimal. It does not expose system instructions or conversation history. Instead, it presents a single-purpose agent with a short description, a place to choose provider/model, and a large text box where users can ask their question.

4.2 UI flow

User opens Streamlit page
    ↓
Reads what the agent is specialised in
    ↓
Enters a question and clicks the button
    ↓
Streamlit sends JSON to /chat
    ↓
Shows answer or error message

4.3 UI imports and layout

import requests
import streamlit as st

API_URL = "http://127.0.0.1:9999/chat"

st.set_page_config(page_title="Specialised LangGraph Agent", layout="centered")
st.header("Specialised LangGraph Agent")

st.write(
    "This console connects to a single-purpose agent whose role and tools "
    "are defined in a configuration file. The agent does not keep history; "
    "each question is handled independently."
)

st.markdown(
    "For advanced testing, you can also call the backend directly via the "
    "FastAPI docs at http://127.0.0.1:9999/docs."
)

4.4 Provider, model, and question widgets

GROQ_MODELS = ["llama-3.3-70b-versatile", "mixtral-8x7b-32768", "llama3-70b-8192"]
OPENAI_MODELS = ["gpt-4o-mini"]

provider = st.radio("Model provider:", ["Groq", "OpenAI"])

if provider == "Groq":
    model_name = st.selectbox("Groq model:", GROQ_MODELS)
else:
    model_name = st.selectbox("OpenAI model:", OPENAI_MODELS)

allow_search = st.checkbox("Allow web search (if enabled in config)", value=False)

user_question = st.text_area(
    "Your question:",
    placeholder="Describe your issue or ask a domain-specific question...",
    height=160,
)

4.5 Sending the request and displaying the answer

if st.button("Ask the agent"):
    question = user_question.strip()

    if not question:
        st.warning("Please enter a question before sending it to the agent.")
    else:
        payload = {
            "model": model_name,
            "provider": provider,
            "messages": [question],
            "enable_search": allow_search,
        }

        try:
            response = requests.post(API_URL, json=payload, timeout=60)
        except requests.RequestException as exc:
            st.error(f"Request to backend failed: {exc}")
        else:
            if response.status_code != 200:
                st.error(f"Backend returned HTTP {response.status_code}.")
            else:
                data = response.json()
                if "error" in data:
                    st.error(data["error"])
                else:
                    st.subheader("Agent answer")
                    st.markdown(data.get("answer", "").strip())

Why the UI remains simple

  • No system prompt field appears: the agent’s behaviour is locked by configuration.
  • No conversation history is stored or shown; every request is stateless.
  • The only toggles are model choice, provider, and whether web search is allowed for that call.

5. Glossary of Functions, Classes, and Key Syntax

5.1 LangGraph and tools

  • create_react_agent(model, tools, state_modifier) – builds a ReAct-style agent graph that can call tools while reasoning.
  • AIMessage – message type representing assistant replies in the LangChain message schema.
  • @tool – decorator that turns a Python function into a LangChain tool that the agent can call.
  • TavilySearchResults(max_results=n) – tool wrapper for web search; max_results limits the number of results returned.

5.2 Configuration and guardrails

  • os.getenv("AGENT_CONFIG_PATH", default) – reads the path to the configuration file from an environment variable with a default fallback.
  • Path(...).read_text() – reads the entire content of a text file using pathlib.
  • json.loads(...) – converts a JSON string into a Python dictionary.
  • apply_output_guardrails(text) – function that trims overly long answers and masks certain substrings before returning the final text.

5.3 FastAPI and Pydantic

  • BaseModel – base class for defining request/response schemas in Pydantic.
  • class ChatRequest(BaseModel): ... – specifies the JSON structure expected by the /chat endpoint.
  • FastAPI(title="...") – creates an application object for defining routes.
  • @app.post("/chat") – registers a handler for POST requests to /chat.
  • uvicorn.run(app, host, port) – runs the FastAPI application on the given host and port.

5.4 Streamlit and HTTP client

  • st.set_page_config(...) – configures page title and layout.
  • st.header / st.write / st.markdown – functions for rendering text elements.
  • st.radio / st.selectbox / st.checkbox / st.text_area / st.button – input widgets used to collect parameters and questions from the user.
  • requests.post(url, json=payload, timeout=...) – sends a JSON POST request to the backend.
  • response.json() – parses the body of the HTTP response as JSON.

You now have a focused, single-purpose agent stack: configuration-driven behaviour, a clean LangGraph core, a small but expressive FastAPI interface, and a Streamlit console that lets users ask questions.

Comments

Post a Comment

Popular posts from this blog

Vector Databases

-
Vector Databases – Three-Phase Learning Roadmap This index links to the three phases of the series: foundations (vectors and similarity), embeddings, and vector databases / indexing. Phase 1 – Vector Spaces and Similarity Foundations What Is a Vector and Why Do We Use Vectors? Vector Notation and Dimensionality Basic Operations on Vectors (add, scale, dot product) Distance vs Similarity (L2, inner product, cosine) Geometric Intuition: Angles, Projections, Subspaces Mapping Simple Data Into Vectors (one-hot, numeric features) From “Just Numbers” to Semantic Meaning Phase 2 – Embeddings and Representation Learning What Is an Embedding and Why We Need It Text Preprocessing and Tokenization Classical Word Embeddings (word2vec, CBOW, Skip-gram) Sentence and Document Embeddings Training Objectives an...
Read more

Best Python Learning Resources for Data Engineering Job

-
s The best overall starting resource is the University of Helsinki Python Programming MOOC . The best data-engineering-specific follow-on is Python and Pandas for Data Engineering from Pragmatic AI Labs. The best polished fundamentals course is CS50P . The best fast hands-on supplement is Kaggle Python . What this guide covers What this category is really about What actually matters in a Python resource for data engineering Weighted decision framework Compared lineup Comparison table and weighted scores Critical narrative analysis Best overall, runner-up, and best value picks Sources checked What this category is really about This is not a generic “best Python course” roundup. The brief from the conversation was narrower and more useful: what are the best Python learning resources for data engineering jobs? That changes the ranking immediately. A resource can be excellent for hobby ...
Read more

Best SQL Learning Resources for Data Engineering Jobs and Interviews

-
The best overall SQL learning resource for data engineering jobs and interviews is Mode SQL Tutorial . The runner-up and best interview-practice platform is StrataScratch . The best fundamentals anchor is PostgreSQL Tutorial and Docs . The best zero-friction beginner start is SQLBolt . What this guide covers What this category is really about What actually matters in a SQL resource for data engineering Weighted decision framework Compared lineup Comparison table and weighted scores Critical narrative analysis Best overall, runner-up, and best value picks Sources checked What this category is really about This is not a generic “best SQL course” roundup. The brief from the conversation was narrower and more useful: what are the best SQL learning resources for data engineering jobs and interviews? That distinction matters. A resource can be perfectly good for absolute beginners and stil...
Read more