Top 5 Python Frameworks for AI Engineers

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition

# Cyclic StateGraph definition
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools))

workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", tools_condition)
workflow.add_edge("tools", "agent")  # Cyclic feedback loop

app = workflow.compile(checkpointer=PostgresSaver.from_conn_string(DB_URI))

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

class FinancialReport(BaseModel):
    revenue_growth_pct: float = Field(..., gt=-100)
    risk_factors: list[str] = Field(..., min_length=1)

agent = Agent('openai:gpt-6-sol', result_type=FinancialReport)

@agent.tool
async def fetch_sec_filing(ctx: RunContext[DatabaseConn], ticker: str) -> str:
    return await ctx.deps.query_sec(ticker)

result = await agent.run('Analyze Q3 results for NVDA', deps=db_connection)
print(result.data.revenue_growth_pct)  # 100% Type-Safe Float!

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.agent import ReActAgent

# 1. Load multi-source enterprise documents
documents = SimpleDirectoryReader("./enterprise_docs").load_data()
index = VectorStoreIndex.from_documents(documents)
engine = index.as_query_engine(similarity_top_k=5)

# 2. Expose as structured Agent Tool with citations
tool = QueryEngineTool(
    query_engine=engine,
    metadata=ToolMetadata(name="hr_policy", description="HR internal policies")
)
agent = ReActAgent.from_tools([tool], verbose=True)

import dspy
from dspy.teleprompt import BootstrapFewShot

# 1. Define clean declarative signature
class SummarizeWithFacts(dspy.Signature):
    """Summarize the article while preserving key statistical metrics."""
    article: str = dspy.InputField()
    summary: str = dspy.OutputField()

# 2. Define Module
class Pipeline(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate = dspy.ChainOfThought(SummarizeWithFacts)

    def forward(self, article):
        return self.generate(article=article)

# 3. DSPy compiles & optimizes few-shot examples automatically!
teleprompter = BootstrapFewShot(metric=accuracy_metric)
compiled_program = teleprompter.compile(Pipeline(), trainset=trainset)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
# Points directly to local self-hosted vLLM engine instance
vllm_client = AsyncOpenAI(base_url="http://localhost:8000/v1", api_key="token-vllm")

@app.post("/api/agent/stream")
async def stream_agent(prompt: str):
    async def event_generator():
        stream = await vllm_client.chat.completions.create(
            model="meta-llama/Llama-3.3-70B-Instruct",
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )
        async for chunk in stream:
            token = chunk.choices[0].delta.content or ""
            yield f"data: {token}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")