9. FastAPI Multi-Agent AI Hindi – LangGraph Zero to Hero

from typing import TypedDict, List, Annotated
from operator import add

class AgentState(TypedDict):
    messages: Annotated[List, add]  # add operator se history accumulate hogi
    current_task: str
    result: str

def research_node(state: AgentState):
    # State se data read karo
    task = state["current_task"]
    
    # Kaam karo (e.g., call LLM, search web)
    result = perform_research(task)
    
    # Updated state return karo
    return {"result": result}

def router(state: AgentState) -> str:
    if "research" in state["current_task"]:
        return "research_node"
    else:
        return "default_node"

graph.add_conditional_edges("start", router)

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)

# Add nodes
graph.add_node("research", research_node)
graph.add_node("summarize", summarize_node)

# Add edges
graph.add_edge(START, "research")
graph.add_edge("research", "summarize")
graph.add_edge("summarize", END)

# Compile graph
app = graph.compile()

from langgraph.func import entrypoint, task

@entrypoint()
def my_workflow(state):
    result = my_task(state)
    return result

# Virtual environment banao
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Dependencies install karo
pip install fastapi uvicorn langgraph langchain langchain-openai python-dotenv sse-starlette

OPENAI_API_KEY="your-api-key-here"

from typing import TypedDict, List, Annotated, Literal
from operator import add

class MultiAgentState(TypedDict):
    messages: Annotated[List, add]
    query: str
    next_agent: str
    research_result: str
    math_result: str
    final_answer: str

from langchain_openai import ChatOpenAI
from langgraph.types import Command
import operator

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Research Agent
def research_agent(state: MultiAgentState):
    query = state["query"]
    
    # Simulate research (real implementation mein Tavily ya other search use karo)
    research_result = f"Research findings for '{query}': Found 5 relevant sources..."
    
    return Command(
        update={"research_result": research_result},
        goto="supervisor"  # Wapas supervisor ke paas jao
    )

# Math Agent
def math_agent(state: MultiAgentState):
    query = state["query"]
    
    # Simple math extraction
    import re
    numbers = re.findall(r'\d+', query)
    if numbers:
        result = sum(int(n) for n in numbers)
        math_result = f"Sum of numbers: {result}"
    else:
        math_result = "No math operations found"
    
    return Command(
        update={"math_result": math_result},
        goto="supervisor"
    )

# Summarize Agent
def summarize_agent(state: MultiAgentState):
    research = state.get("research_result", "No research found")
    math = state.get("math_result", "")
    
    combined = f"Research: {research}\nMath: {math}"
    
    response = llm.invoke([
        {"role": "system", "content": "You are a summarizer. Create a concise final answer from the provided information."},
        {"role": "user", "content": combined}
    ])
    
    return Command(
        update={"final_answer": response.content},
        goto="__end__"  # Graph end
    )

from langgraph.graph import StateGraph, START, END
from langgraph_supervisor import create_supervisor  # Requires pip install langgraph-supervisor

# Ya manually implement with conditional routing

def supervisor_router(state: MultiAgentState) -> str:
    """Decide which agent to call next"""
    query = state["query"].lower()
    
    # Check if research needed
    if "search" in query or "find" in query or "research" in query:
        if not state.get("research_result"):
            return "research_agent"
    
    # Check if math needed
    if any(word in query for word in ["sum", "add", "calculate", "math", "number"]):
        if not state.get("math_result"):
            return "math_agent"
    
    # If both done, go to summarize
    if state.get("research_result") or state.get("math_result"):
        return "summarize_agent"
    
    return "summarize_agent"

# Create graph
builder = StateGraph(MultiAgentState)

# Add nodes
builder.add_node("supervisor", lambda state: state)  # Router node
builder.add_node("research_agent", research_agent)
builder.add_node("math_agent", math_agent)
builder.add_node("summarize_agent", summarize_agent)

# Add edges
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", supervisor_router, {
    "research_agent": "research_agent",
    "math_agent": "math_agent",
    "summarize_agent": "summarize_agent"
})

# Worker agents wapas supervisor ko route karte hain via Command (already handled)
builder.add_edge("research_agent", "supervisor")
builder.add_edge("math_agent", "supervisor")
builder.add_edge("summarize_agent", END)

# Compile
agent_graph = builder.compile()

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager

app = FastAPI(title="Multi-Agent AI Assistant")

# Request/Response models
class QueryRequest(BaseModel):
    query: str

class QueryResponse(BaseModel):
    final_answer: str
    research_result: str = None
    math_result: str = None

@app.post("/ask", response_model=QueryResponse)
async def ask_agent(request: QueryRequest):
    try:
        initial_state = {
            "messages": [],
            "query": request.query,
            "next_agent": "",
            "research_result": "",
            "math_result": "",
            "final_answer": ""
        }
        
        result = await agent_graph.ainvoke(initial_state)
        
        return QueryResponse(
            final_answer=result.get("final_answer", "No answer generated"),
            research_result=result.get("research_result"),
            math_result=result.get("math_result")
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

from fastapi.responses import StreamingResponse
import json
import asyncio

@app.post("/ask/stream")
async def ask_agent_stream(request: QueryRequest):
    async def event_generator():
        try:
            initial_state = {
                "messages": [],
                "query": request.query,
                "next_agent": "",
                "research_result": "",
                "math_result": "",
                "final_answer": ""
            }
            
            # Send start event
            yield f"data: {json.dumps({'type': 'start', 'query': request.query})}\n\n"
            await asyncio.sleep(0.1)
            
            # Send planning event
            yield f"data: {json.dumps({'type': 'planning', 'message': '🤔 Analyzing your query...'})}\n\n"
            await asyncio.sleep(0.5)
            
            # Stream each step (simplified – actual implementation use astream_events)
            if "search" in request.query.lower():
                yield f"data: {json.dumps({'type': 'research', 'message': '🔍 Research Agent searching the web...'})}\n\n"
                await asyncio.sleep(1)
            
            if "sum" in request.query.lower() or "add" in request.query.lower():
                yield f"data: {json.dumps({'type': 'math', 'message': '🧮 Math Agent calculating...'})}\n\n"
                await asyncio.sleep(0.5)
            
            # Execute agent and get final result
            result = await agent_graph.ainvoke(initial_state)
            
            # Stream final answer token by token
            final_answer = result.get("final_answer", "")
            words = final_answer.split()
            for i, word in enumerate(words):
                yield f"data: {json.dumps({'type': 'token', 'content': word + (' ' if i < len(words)-1 else '')})}\n\n"
                await asyncio.sleep(0.03)  # Typing effect
            
            # Send done event
            yield f"data: {json.dumps({'type': 'done'})}\n\n"
            
        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"
        }
    )

from langgraph.checkpoint.sqlite import SqliteSaver

# Create checkpointer
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# Compile graph with checkpointer
agent_graph = builder.compile(checkpointer=checkpointer)

# Use with thread_id for conversation management
config = {"configurable": {"thread_id": "user_123"}}

# Invoke with config
result = await agent_graph.ainvoke(initial_state, config=config)

# Resume from checkpoint (for HITL)
result = await agent_graph.ainvoke(None, config=config)

from langgraph.types import interrupt, Command

def human_approval_node(state: MultiAgentState):
    """Node that asks for human approval before proceeding"""
    
    # Create interrupt – graph execution pause ho jayegi
    user_approval = interrupt({
        "question": "Shall I send the email?",
        "draft": state.get("draft_email", ""),
        "options": ["yes", "no", "edit"]
    })
    
    if user_approval == "yes":
        return Command(
            update={"status": "approved", "user_decision": "yes"},
            goto="send_email_node"
        )
    elif user_approval == "no":
        return Command(
            update={"status": "rejected", "user_decision": "no"},
            goto="__end__"
        )
    else:
        return Command(
            update={"status": "needs_edit", "user_decision": "edit"},
            goto="edit_email_node"
        )

# Add to graph
builder.add_node("human_approval", human_approval_node)

class ResumeRequest(BaseModel):
    thread_id: str
    user_decision: str  # "yes", "no", or custom value

@app.post("/agent/resume")
async def resume_agent(request: ResumeRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    
    # Resume execution with user input
    result = await agent_graph.ainvoke(
        Command(resume=request.user_decision),
        config=config
    )
    
    return {"result": result}

research_assistant/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── graph.py          # Agent graph definition
│   ├── agents.py         # Worker agent implementations
│   ├── models.py         # Pydantic models
│   ├── streaming.py      # SSE event handling
│   └── checkpoints.py    # Checkpointer setup
├── .env
└── requirements.txt

async with asyncio.timeout(30):
    result = await agent.ainvoke(state)

# ---------- BASIC GRAPH ----------
from langgraph.graph import StateGraph, START, END

graph = StateGraph(MyState)
graph.add_node("node_a", func_a)
graph.add_edge(START, "node_a")
graph.add_edge("node_a", END)
app = graph.compile()

# ---------- CONDITIONAL EDGES ----------
def router(state):
    if state["condition"]: return "node_b"
    return "node_c"
graph.add_conditional_edges("node_a", router)

# ---------- COMMAND PATTERN ----------
from langgraph.types import Command
def my_node(state):
    return Command(
        update={"field": value},
        goto="next_node"
    )

# ---------- HUMAN-IN-THE-LOOP ----------
from langgraph.types import interrupt
def approval_node(state):
    user_input = interrupt({"prompt": "Approve?"})
    if user_input == "yes":
        return Command(goto="next")
    return Command(goto="__end__")

# ---------- FASTAPI ENDPOINT ----------
@app.post("/ask")
async def ask(request: Request):
    result = await graph.ainvoke({"query": request.query})
    return result