How It All Works

Uses Response(stream_with_context(generate()), content_type='text/event-stream'). The inner generator yields JSON-encoded token chunks as SSE events.

def generate():
    stream = groq.create(..., stream=True)
    for chunk in stream:
        content = chunk.choices[0].delta.content
        yield f"data: {json.dumps({'token':content})}\n\n"
    yield f"data: {json.dumps({'done':True})}\n\n"

return Response(
    stream_with_context(generate()),
    content_type='text/event-stream',
    headers={'Cache-Control': 'no-cache',
             'X-Accel-Buffering': 'no'})

The browser uses the Fetch API with a ReadableStream reader rather than EventSource, allowing POST requests with JSON bodies. Tokens are parsed and appended in real-time.

const resp = await fetch('/api/chat/stream', {
  method: 'POST',
  body: JSON.stringify({message, model, session_id})
});
const reader = resp.body.getReader();
while (true) {
  const {done, value} = await reader.read();
  if (done) break;
  // parse SSE lines, extract .token, append
}

Chunks are accumulated during streaming. After done, the full assembled text is saved to SQLite, token usage is logged (with estimated fallback if the API doesn't return usage), and the session ID is emitted.

# After streaming loop ends:
full_text = ''.join(chunks)
_add(session_id, 'user',      user_msg)
_add(session_id, 'assistant', full_text)
log_usage(model, provider,
           tokens_in, tokens_out, latency_ms)

Even blocked responses stream word-by-word for consistent UX. Off-topic detection runs before the LLM call; the canned reply is split on spaces and yielded as token events.

if is_off_topic(message, has_history):
    def _off():
        for word in OFF_TOPIC_REPLY.split(' '):
            yield (
                f"data: "
                + json.dumps({'token': word+' '})
                + "\n\n"
            )
        yield f"data: {json.dumps({'done':True})}\n\n"
    return Response(stream_with_context(_off()), ...)

A typed TypedDict state flows through nodes. Conditional edges route execution based on state values. Checkpointers enable pause-and-resume and human-in-the-loop.

from langgraph.graph import StateGraph

graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("tools",    ToolNode(tools))
graph.add_conditional_edges(
    "router", route_fn,
    {"research": "research",
     "analysis": "analysis"})

Each node binds tools to the LLM with .bind_tools(). The model returns tool_calls in its message; the graph routes to a ToolNode, executes, and feeds the result back into state.

llm_with_tools = llm.bind_tools([
    web_search,
    code_executor,
    retriever,
])

# ToolNode handles execution automatically
tool_node = ToolNode(tools=[
    web_search, code_executor, retriever
])

Agents are defined with role, goal, and backstory. Tasks are assigned per-agent and executed sequentially. The crew orchestrator manages handoffs and shared context.

researcher = Agent(
  role="Research Analyst",
  goal="Find accurate information",
  backstory="Expert at gathering data",
  tools=[search_tool])

crew = Crew(
  agents=[researcher, analyst, writer],
  tasks=[task1, task2, task3],
  process=Process.sequential)

All agents follow Reason → Act → Observe. The LLM reasons about the goal, picks a tool action, receives the observation, and loops until it decides the task is complete.

# ReAct cycle (conceptual)
while not done:
    thought = llm.think(state, tools)
    if thought.is_final_answer:
        break
    obs = thought.tool.invoke(thought.args)
    state.update(observation=obs)
    # loop: reason again with observation