Emmanuel Cheruiyot

Project Setup & Requirements First, install the necessary dependencies: pip install fastapi uvicorn langgraph langchain-openai pydantic sse-starlette python-dotenv Set your API key in an .env file or environment variable: export OPENAI_API_KEY="your-api-key-here" File 1: agent.py (LangGraph Definition) This module defines the agent's state, tools, graph nodes, and workflow compilation. import os from typing import Annotated, TypedDict, Sequence from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver # 1. Define Agent State class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] # 2. Initialize LLM Model # Requires OPENAI_API_KEY in environment llm = ChatOpenAI(model="gpt-4o-mini", streaming=True) # 3. Define Graph Nodes async def call_model_node(state: AgentState): """Executes the primary model call with system prompt guidance.""" messages = state["messages"] system_prompt = SystemMessage( content="You are a helpful, concise AI assistant integrated into a FastAPI web backend." ) # Prepend system message if not present if not any(isinstance(m, SystemMessage) for m in messages): messages = [system_prompt] + list(messages) response = await llm.ainvoke(messages) return {"messages": [response]} # 4. Construct Graph builder = StateGraph(AgentState) # Add node(s) builder.add_node("agent", call_model_node) # Add edge(s) builder.add_edge(START, "agent") builder.add_edge("agent", END) # In-memory checkpointer for thread history (Use PostgresSaver for production) checkpointer = MemorySaver() # Compile the workflow agent_graph = builder.compile(checkpointer=checkpointer) File 2: main.py (FastAPI Server) This module initializes FastAPI, sets up CORS middleware, and provides both standard JSON and real-time streaming (SSE) endpoints. import json import asyncio from typing import AsyncGenerator, List, Optional from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from sse_starlette.sse import EventSourceResponse from langchain_core.messages import HumanMessage, AIMessage from agent import agent_graph app = FastAPI( title="LangGraph + FastAPI Integration Starter", version="1.0.0", ) # Enable CORS for local web clients app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Request & Response Schemas class ChatRequest(BaseModel): message: str = Field(..., description="User message input") thread_id: str = Field(default="default_thread", description="Unique session identifier for history tracking") class ChatResponse(BaseModel): thread_id: str response: str # ------------------------------------------------------------------ # Endpoint 1: Synchronous / Standard JSON Endpoint # ------------------------------------------------------------------ @app.post("/api/chat", response_model=ChatResponse) async def chat_endpoint(request: ChatRequest): """Executes graph and returns complete response in standard JSON.""" config = {"configurable": {"thread_id": request.thread_id}} inputs = {"messages": [HumanMessage(content=request.message)]} try: # Run graph to completion result = await agent_graph.ainvoke(inputs, config=config) last_message = result["messages"][-1] return ChatResponse( thread_id=request.thread_id, response=str(last_message.content) ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # ------------------------------------------------------------------ # Endpoint 2: Streaming SSE (Server-Sent Events) Endpoint # ------------------------------------------------------------------ async def generate_graph_stream(message: str, thread_id: str) -> AsyncGenerator[dict, None]: """Streams token events directly from the LangGraph execution pipeline.""" config = {"configurable": {"thread_id": thread_id}} inputs = {"messages": [HumanMessage(content=message)]} try: # Stream events from LangGraph async for event in agent_graph.astream_events(inputs, config=config, version="v2"): # Intercept raw tokens generated by the LLM inside the graph node if event["event"] == "on_chat_model_stream": chunk_content = event["data"]["chunk"].content if chunk_content: yield { "event": "message", "data": json.dumps({"content": chunk_content}) } # Send end marker event yield {"event": "done", "data": json.dumps({"status": "completed"})} except Exception as e: yield {"event": "error", "data": json.dumps({"error": str(e)})} @app.post("/api/chat/stream") async def stream_chat_endpoint(request: ChatRequest): """Streams response tokens line-by-line via SSE.""" return EventSourceResponse( generate_graph_stream(request.message, request.thread_id) ) # Health Check @app.get("/health") async def health_check(): return {"status": "ok"} if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True) How to Run & Test * Start the FastAPI Server: python main.py The interactive API docs will be live at http://localhost:8000/docs. * Test standard JSON Endpoint via curl: curl -X POST "http://localhost:8000/api/chat" \ -H "Content-Type: application/json" \ -d '{"message": "Hello, how does this architecture work?", "thread_id": "user_session_1"}' * Test Real-Time Streaming (SSE) via curl: curl -N -X POST "http://localhost:8000/api/chat/stream" \ -H "Content-Type: application/json" \ -d '{"message": "Write a short poem about building AI apps.", "thread_id": "user_session_1"}'