FastAPI backends and data pipelines, built to be tested and explained.
Python · APIs · data pipelines · open-source stat-arb research

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

class AgentState(TypedDict):
    messages: list[BaseMessage]
    intent: str
    tool_results: dict
    needs_human: bool

def classify_intent(state: AgentState) -> AgentState:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Classify: rag | code_exec | human_escalate"),
        ("human", "{query}")
    ])
    result = llm.invoke(prompt.format(query=state["messages"][-1].content))
    state["intent"] = result.content.strip().lower()
    return state

def route_by_intent(state: AgentState) -> str:
    routes = {"rag": "retrieve", "code_exec": "execute",
              "human_escalate": "handoff"}
    return routes.get(state["intent"], "retrieve")

def retrieve_context(state: AgentState) -> AgentState:
    query = state["messages"][-1].content
    embedding = embeddings.embed_query(query)
    docs = vector_store.similarity_search_by_vector(embedding, k=8)
    reranked = reranker.rerank(query, docs, top_k=3)
    state["tool_results"]["retrieved"] = [d.page_content for d in reranked]
    return state

def execute_code(state: AgentState) -> AgentState:
    sandbox = Sandbox(timeout=15, allow_imports=["numpy", "pandas"])
    result = sandbox.run(state["tool_results"].get("code", ""))
    state["tool_results"]["exec_output"] = result.stdout or result.stderr
    return state

graph = StateGraph(AgentState) \
    .add_node("classify", classify_intent) \
    .add_node("retrieve", retrieve_context) \
    .add_node("execute", execute_code) \
    .add_node("handoff", human_handoff) \
    .add_conditional_edges("classify", route_by_intent, {
        "retrieve": "retrieve", "execute": "execute",
        "handoff": "handoff"
    }) \
    .add_edge("retrieve", END) \
    .add_edge("execute", END) \
    .compile()

import asyncio
from redis.asyncio import Redis
from typing import AsyncGenerator

class EventPipeline:
    def __init__(self, redis: Redis, stream: str, group: str,
                 consumer: str, max_pending: int = 500):
        self.redis = redis
        self.stream = stream
        self.group = group
        self.consumer = consumer
        self.max_pending = max_pending
        self.dlq = f"{stream}:dead"

    async def ingest(self, events: AsyncGenerator[dict, None]) -> int:
        count = 0
        batch = []
        async for event in events:
            batch.append(("event", event))
            if len(batch) >= 50:
                await self.redis.xadd(self.stream, dict(batch))
                count += len(batch); batch = []
        if batch:
            await self.redis.xadd(self.stream, dict(batch))
            count += len(batch)
        return count

    async def consume(self, handler) -> None:
        await self._ensure_group()
        while True:
            pending = await self.redis.xpending(self.stream, self.group)
            if pending["pending"] >= self.max_pending:
                await asyncio.sleep(0.5); continue  # backpressure

            messages = await self.redis.xreadgroup(
                self.group, self.consumer, {self.stream: ">"},
                count=10, block=2000
            )
            for msg_id, data in messages:
                try:
                    await handler(data)
                    await self.redis.xack(self.stream, self.group, msg_id)
                except Exception as exc:
                    await self.redis.xadd(self.dlq, {
                        "original_id": msg_id, "data": str(data),
                        "error": str(exc), "ts": str(time.time())
                    })
                    await self.redis.xack(self.stream, self.group, msg_id)

    async def _ensure_group(self):
        try:
            await self.redis.xgroup_create(
                self.stream, self.group, id="0", mkstream=True
            )
        except ResponseError:
            pass  # group already exists