Capstone: Multi-Document Agentic RAG With Structured Output

Everything this course built, in one script

This is the last lesson in the course, and it doesn't introduce a new concept so much as wire together concepts from all three tiers:

Two document sets, one agent that picks

The Nimbus policy documents (vacation, remote work, expense) and the Nimbus engineering documentation (engineering handbook + Cobalt-1 product FAQ, reused from 02_intermediate/13_multi_document_indexes/data/) are genuinely different domains, HR questions and engineering questions shouldn't search the same index. Each gets its own small VectorStoreIndex, wrapped as its own QueryEngineTool with a description telling the agent what it's for:

policy_tool = build_query_engine_tool(POLICY_DATA_DIR, name="hr_policies", description="...")
engineering_tool = build_query_engine_tool(ENGINEERING_DATA_DIR, name="engineering_docs", description="...")

FunctionAgent receives both tools and decides, per question, whether it needs one, the other, or both, the same tool-selection reasoning Lesson 14's predict_and_call() did for a single tool, now scaled to multiple tools and (if needed) multiple calls in sequence.

Structured output at the agent level

Lesson 10 showed output_cls on a single query engine, coercing one synthesized answer into a Pydantic model. FunctionAgent accepts the same output_cls keyword, applied to its *final* answer, after however many tool calls it took to get there:

class NimbusAnswer(BaseModel):
    answer: str
    document_sets_used: list[str]
    needs_human_followup: bool

agent = FunctionAgent(tools=[...], llm=Settings.llm, output_cls=NimbusAnswer, ...)

document_sets_used and needs_human_followup aren't things a plain text answer would reliably self-report, having the schema force those fields is what makes them dependable to consume downstream (log which tools fired, route flagged answers to a human) rather than something you'd have to regex out of prose.

The code, piece by piece

result = await agent.run(question)

FunctionAgent.run() is async (it's a Workflow under the hood, Lesson 19's mechanics), so this lesson wraps it in a small run_all() coroutine and calls asyncio.run() once at the bottom, rather than sprinkling await through a sync main().

parsed: NimbusAnswer = result.get_pydantic_model(NimbusAnswer)

agent.run() returns an AgentOutput event, not the Pydantic model directly. .get_pydantic_model(NimbusAnswer) validates AgentOutput.structured_response against the model class and returns a real NimbusAnswer instance, or None with a warning if validation failed, the same "coerced and validated, not just text" guarantee Lesson 10 established.

def build_query_engine_tool(data_dir, name, description) -> QueryEngineTool:
    ...
    return QueryEngineTool.from_defaults(query_engine=query_engine, name=name, description=description)

Factored into a function since this lesson needs the same build-index-then-wrap-as-tool steps twice, once per document set, identical to what Lesson 14 did for a single FunctionTool, just built from a QueryEngine instead of a plain Python function.