I build agentic products, real-time data systems, and production AI infrastructure.
End-to-end execution — backend, infrastructure, deployment, observability.

1from langgraph.graph import StateGraph, END
2from langgraph.prebuilt import ToolNode
3
4class AgentState(TypedDict):
5    messages: list[BaseMessage]
6    intent: str
7    tool_results: dict
8    needs_human: bool
9
10def classify_intent(state: AgentState) -> AgentState:
11    prompt = ChatPromptTemplate.from_messages([
12        ("system", "Classify: rag | code_exec | human_escalate"),
13        ("human", "{query}")
14    ])
15    result = llm.invoke(prompt.format(query=state["messages"][-1].content))
16    state["intent"] = result.content.strip().lower()
17    return state
18
19def route_by_intent(state: AgentState) -> str:
20    routes = {"rag": "retrieve", "code_exec": "execute",
21              "human_escalate": "handoff"}
22    return routes.get(state["intent"], "retrieve")
23
24def retrieve_context(state: AgentState) -> AgentState:
25    query = state["messages"][-1].content
26    embedding = embeddings.embed_query(query)
27    docs = vector_store.similarity_search_by_vector(embedding, k=8)
28    reranked = reranker.rerank(query, docs, top_k=3)
29    state["tool_results"]["retrieved"] = [d.page_content for d in reranked]
30    return state
31
32def execute_code(state: AgentState) -> AgentState:
33    sandbox = Sandbox(timeout=15, allow_imports=["numpy", "pandas"])
34    result = sandbox.run(state["tool_results"].get("code", ""))
35    state["tool_results"]["exec_output"] = result.stdout or result.stderr
36    return state
37
38graph = StateGraph(AgentState) \
39    .add_node("classify", classify_intent) \
40    .add_node("retrieve", retrieve_context) \
41    .add_node("execute", execute_code) \
42    .add_node("handoff", human_handoff) \
43    .add_conditional_edges("classify", route_by_intent, {
44        "retrieve": "retrieve", "execute": "execute",
45        "handoff": "handoff"
46    }) \
47    .add_edge("retrieve", END) \
48    .add_edge("execute", END) \
49    .compile()

1import asyncio
2from redis.asyncio import Redis
3from typing import AsyncGenerator
4
5class EventPipeline:
6    def __init__(self, redis: Redis, stream: str, group: str,
7                 consumer: str, max_pending: int = 500):
8        self.redis = redis
9        self.stream = stream
10        self.group = group
11        self.consumer = consumer
12        self.max_pending = max_pending
13        self.dlq = f"{stream}:dead"
14
15    async def ingest(self, events: AsyncGenerator[dict, None]) -> int:
16        count = 0
17        batch = []
18        async for event in events:
19            batch.append(("event", event))
20            if len(batch) >= 50:
21                await self.redis.xadd(self.stream, dict(batch))
22                count += len(batch); batch = []
23        if batch:
24            await self.redis.xadd(self.stream, dict(batch))
25            count += len(batch)
26        return count
27
28    async def consume(self, handler) -> None:
29        await self._ensure_group()
30        while True:
31            pending = await self.redis.xpending(self.stream, self.group)
32            if pending["pending"] >= self.max_pending:
33                await asyncio.sleep(0.5); continue  # backpressure
34
35            messages = await self.redis.xreadgroup(
36                self.group, self.consumer, {self.stream: ">"},
37                count=10, block=2000
38            )
39            for msg_id, data in messages:
40                try:
41                    await handler(data)
42                    await self.redis.xack(self.stream, self.group, msg_id)
43                except Exception as exc:
44                    await self.redis.xadd(self.dlq, {
45                        "original_id": msg_id, "data": str(data),
46                        "error": str(exc), "ts": str(time.time())
47                    })
48                    await self.redis.xack(self.stream, self.group, msg_id)
49
50    async def _ensure_group(self):
51        try:
52            await self.redis.xgroup_create(
53                self.stream, self.group, id="0", mkstream=True
54            )
55        except ResponseError:
56            pass  # group already exists