Healthcare Multi-Source Agentic RAG Platform: Part 2

Table of Contents

1. LangGraph Workflow & Agents
2. LangGraph Workflow Patterns Used
3. Core Steps in the Workflow
   • How question_graph/agents/prompts work together
   • RAG and GraphRAG
   • Postgres and Glue (metadata split)
   • Conversation Memory
   • Routers
   • Load Data
   • AWS Storage
   • OpenSearch, Embedding, and LLM Services
4. Resources

Part 1 of this guide covers environment setup, the two ingestion/question flows at a system level, code organization, and API usage. This part goes one level deeper into the LangGraph agentic core specifically.

---

LangGraph Workflow & Agents

The question-answering system is a single, compiled LangGraph StateGraph, built once at import time:

router_agent = RouterAgent()
sql_agent = SQLAgent()
s3_agent = S3Agent()
rag_agent = RAGAgent()
graph_rag_agent = GraphRAGAgent()
summarization_agent = SummarizationAgent()
classification_agent = ClassificationAgent()
final_answer_agent = FinalAnswerAgent()
memory_service = MemoryService()

question_graph = build_question_graph()

Every agent is a module-level singleton — constructed once, shared across every request, never recreated per-question. This mirrors the same pattern used for the module-level workflow singletons in api/routes.py (Part 1).

Nine agents exist in total. Eight participate in the graph (router, sql, s3, rag, graph_rag, summarization, classification, final_answer); the ninth, ChartAgent, powers the separate chart-explorer endpoints and is not part of this graph — there’s no chart route in RouterAgent’s ROUTES list, so asking for a chart in the main chat won’t currently reach it.

The graph shape

                              ┌─────────────┐
                    ┌────────►│    sql      ├────┐
                    │         └─────────────┘    │
                    │         ┌─────────────┐    │
                    ├────────►│     s3      ├────┤
                    │         └─────────────┘    │
                    │         ┌─────────────┐    │
  route_question ───┼────────►│    rag      ├────┤
   (RouterAgent)     │         └─────────────┘    │
                    │         ┌─────────────┐    │      ┌───────────────┐
                    ├────────►│  graph_rag  ├────┼─────►│ final_answer  ├───► END
                    │         └─────────────┘    │      └───────────────┘
                    │         ┌─────────────┐    │
                    ├────────►│summarization├────┤
                    │         └─────────────┘    │
                    │         ┌─────────────┐    │
                    └────────►│classification├───┘
                              └─────────────┘

One entry point (route_question), six mutually-exclusive tool nodes selected by a conditional edge, one convergence point (final_answer), one exit (END).

---

LangGraph Workflow Patterns Used

1. Shared, optional-everything state. QuestionState is a TypedDict with total=False — every field is optional, and each node only fills in the fields it’s responsible for:

class QuestionState(TypedDict, total=False):
    session_id: str
    question: str
    route: str
    available_tables: List[str]
    available_documents: List[str]
    available_document_records: List[Dict[str, str]]
    schema_description: str
    conversation_context: str
    tool_answer: str
    sources: List[Dict[str, Any]]
    sql: Optional[str]
    final_answer: str

2. Router + conditional-edge dispatch. One node decides a route; a separate, trivial function (select_route) reads that decision back out of state and returns it as a string key, because add_conditional_edges specifically wants a pure “read state, return a key” function, not the node that mutated state:

def route_question(state: QuestionState) -> QuestionState:
    route = router_agent.route(question=state["question"], ...)
    return {**state, "route": route}

def select_route(state: QuestionState) -> str:
    return state["route"]

graph.add_conditional_edges("route_question", select_route, {
    "sql": "sql", "s3": "s3", "rag": "rag",
    "graph_rag": "graph_rag", "summarization": "summarization",
    "classification": "classification"
})

3. State accumulation via spread, never mutation. Every node returns {**state, "new_field": value} — a new dict built from the old one plus updates — rather than mutating state in place. This is idiomatic for LangGraph’s state-merging model and keeps each node’s effect on state explicit and traceable.

4. Converge-to-one-exit-node. All six tool nodes connect to the same final_answer node before END:

for node in ["sql", "s3", "rag", "graph_rag", "summarization", "classification"]:
    graph.add_edge(node, "final_answer")
graph.add_edge("final_answer", END)

This means citation formatting, SQL-inclusion, and conversation-memory writes only need to be implemented once, in one place, regardless of which route actually ran.

5. Compile once, invoke many times. build_question_graph().compile() runs at import time; QuestionWorkflow.ask() calls .invoke(initial_state) on the already-compiled graph per request — the graph structure itself is never rebuilt.

---

Core Steps in the Workflow

How question_graph/agents/prompts work together

Three layers, each only talking to the layer directly below it:

question_graph.py (decides which agent runs, wires state between nodes)
        |
agents/*.py (decides which prompt to use, calls LLMService, shapes the result)
        |
prompts/*.py (pure string builders — a *_SYSTEM_PROMPT constant + a build_*_prompt() function)

No prompt file imports any service or agent — they’re pure functions, independently testable. sql_prompt.py is a concrete example of reuse: the exact same build_sql_prompt() function serves both SQLAgent (Postgres, dialect="PostgreSQL") and S3Agent (Athena, dialect="Amazon Athena (Presto) SQL"), parameterized rather than duplicated.

ROUTES itself lives in router_prompt.py, not in any agent — it’s the single source of truth both RouterAgent (to validate the model’s response) and question_graph.py (as node/edge names) depend on.

RAG and GraphRAG

RAGAgent.retrieve() embeds the question, runs a kNN search, then reranks the results with a second LLM call before returning them:

def retrieve(self, question, k=5, document_type=None):
    embedding = self.embedding_service.create_embedding(question)
    chunks = self.vector_store.search_chunks(embedding=embedding, k=k, document_type=document_type)
    return self.rerank_chunks(question, chunks)

If the rerank response can’t be parsed into a valid index list, it falls back to the original kNN order rather than raising — an optional quality improvement is never allowed to become a new failure mode. answer()’s sources list is deduplicated by file_id afterward, so a document that supplied several top-k chunks shows up once, not once per chunk.

GraphRAGAgent composes RAGAgent for document evidence and adds real graph traversal:

def answer(self, question: str, k: int = 5, hops: int = 2) -> Dict[str, Any]:
    extraction = extract_entities_and_relationships(question)
    graph_facts = self.graph_store.find_related(extraction["entities"], hops=hops)

    chunks = self.rag_agent.retrieve(question=question, k=k)

    prompt = build_graph_rag_prompt(question=question, graph_facts=graph_facts, chunks=chunks)
    answer = self.llm_service.generate(prompt=prompt, system_prompt=GRAPH_RAG_SYSTEM_PROMPT,
                                        max_tokens=800, temperature=0.2)
    return {"answer": answer, "sources": [...]}

The graph itself is built at ingestion time (spaCy noun-chunk + co-occurrence extraction, upserted into Postgres graph_nodes/graph_edges), then traversed at query time via an in-memory NetworkX ego_graph(). This is documented, deliberately, as a prototype-tier design — not Amazon Neptune — with known limitations: no entity resolution (so "diabetes" and "type 2 diabetes" can end up as separate nodes), and the entire graph is reloaded and rebuilt from Postgres on every single query.

Postgres and Glue (metadata split)

Two real metadata layers, not the three some earlier project notes described:

• Postgres documents table — file lifecycle: file_id, file_name, s3_uri, document_type, status, uploaded_at, processed_at, error_message. Notably, it does not store a glue_table link back to Glue — that value is returned once in the /upload API response and never persisted, a real gap if you ever need to programmatically recover which Glue table a given document created.
• AWS Glue Data Catalog — table/column/type/S3-location metadata for structured datasets, consumed by Athena.

Since the dual-write feature was added, Postgres also holds a third thing that isn’t strictly “metadata”: the actual structured data tables themselves (e.g. a table literally named test_patients with real rows), living alongside documents in the same database.

Conversation Memory

MemoryService sits behind exactly two points in the whole question-answering flow: QuestionWorkflow.ask() reads recent history once, at the very start, and compose_final_answer writes to it once, at the very end.

def ask(self, question: str, session_id: Optional[str] = None) -> Dict[str, Any]:
    ...
    initial_state = {
        "question": question,
        ...
        "conversation_context": self._format_conversation_context(session_id)
    }
    ...

def _format_conversation_context(self, session_id):
    if not session_id:
        return ""
    history = self.memory_service.get_recent_messages(session_id)
    return "\n".join(f"{turn['role']}: {turn['content']}" for turn in history)

def compose_final_answer(state: QuestionState) -> QuestionState:
    result = final_answer_agent.compose(...)

    session_id = state.get("session_id")
    if session_id:
        memory_service.add_message(session_id, "user", state["question"])
        memory_service.add_message(session_id, "assistant", result["answer"])

    return {**state, "final_answer": result["answer"]}

The scope of what memory actually influences is narrower than it might seem — worth being precise about, since it’s easy to assume conversation history feeds into everything. Grepping the graph confirms conversation_context is read in exactly one place: route_question(), where it’s passed into RouterAgent.route() alongside the question and the available tables/documents. It is not passed to any of the six tool nodes, and it is not passed to FinalAnswerAgent.compose() either. In practice this means conversation history can influence which route a follow-up question takes (e.g. helping the router recognize a short follow-up still belongs to the same topic), but it does not directly inform how SQLAgent, RAGAgent, or any other tool agent generates its actual answer, and it plays no role in how the final answer gets worded. session_id itself is generated client-side once (crypto.randomUUID()) and persisted in the browser’s localStorage, so it survives a page reload but resets if storage is cleared or a different browser/device is used — there’s no server-issued session concept.

Underneath, MemoryService composes CacheService (a generic Redis wrapper) for a fast recent-turns cache per session, backed by a durable Postgres conversation_messages table for full history:

def add_message(self, session_id, role, content) -> None:
    # ... INSERT into conversation_messages (Postgres, durable) ...
    self.cache_service.push_json(self._cache_key(session_id), message, max_length=self.recent_messages_limit)

def get_recent_messages(self, session_id) -> List[Dict[str, Any]]:
    cached = self.cache_service.get_list_json(self._cache_key(session_id))
    if cached:
        return cached
    return self.get_history(session_id, limit=self.recent_messages_limit)  # Postgres fallback

One real, verified gap here (covered in full in docs/cache_memory_service.md): the Postgres fallback path (get_history()) never writes its result back into Redis. A cache miss — say, resuming a session after Redis was restarted — stays a miss on every subsequent read until the next add_message() call happens to push something new into the list. It’s a correct read-through, just not a self-healing one.

Routers

api/routes.py is a thin HTTP-to-workflow adapter — it never touches Bedrock, OpenSearch, SQL, or S3 directly. Six endpoints, five singleton workflow/service objects, and nothing else:

document_ingestion_workflow = DocumentIngestionWorkflow()
structured_ingestion_workflow = StructuredIngestionWorkflow()
question_workflow = QuestionWorkflow()
visualization_workflow = VisualizationWorkflow()
document_store = DocumentStore()

Method	Path	Delegates to
POST	/upload	StructuredIngestionWorkflow or DocumentIngestionWorkflow, by detect_file_type()
GET	/documents	DocumentStore.list_all_documents()
DELETE	/documents/{file_id}	OpenSearch chunk delete + S3 file delete + Postgres row delete (does not clean up Glue tables or graph entries)
POST	/ask	QuestionWorkflow.ask()
GET	/visualizations/suggestions	VisualizationWorkflow.suggest()
POST	/visualizations	VisualizationWorkflow.chart()

Every endpoint wraps its workflow call in try/except and maps failures to HTTPException — a 500 for unexpected errors, a 400 specifically for /visualizations when the generated SQL fails the read-only safety check, and a 404 for deleting a document that doesn’t exist.

Load Data

tools/data_loader.py’s detect_file_type() is the very first decision made about any upload:

STRUCTURED_EXTENSIONS = {".csv", ".xlsx", ".xls"}
DOCUMENT_EXTENSIONS = {".pdf", ".docx", ".txt"}
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg"}

Only "structured" is checked explicitly by api/routes.py to branch workflows — "image" falls into the document workflow’s branch, where load_document() re-checks the extension itself and dispatches to load_pdf() / load_docx() / load_txt() / load_image_ocr():

def load_document(file_path: str) -> str:
    ext = os.path.splitext(file_path.lower())[1]
    if ext == ".pdf": return load_pdf(file_path)
    if ext == ".docx": return load_docx(file_path)
    if ext == ".txt": return load_txt(file_path)
    if ext in IMAGE_EXTENSIONS: return load_image_ocr(file_path)
    raise ValueError(f"Unsupported document type: {ext}")

Real, verified gap: .xlsx/.xls are listed in STRUCTURED_EXTENSIONS but have no working loader — load_csv() is the only structured loader, and StructuredIngestionWorkflow.ingest() rejects anything that isn’t .csv with a clean ValueError before ever calling it.

AWS Storage

Covered in depth in Part 1 — AWSStorage (S3 upload/delete) and OpenSearchVectorStore (index/search/delete chunks, get a document’s full text back) are two classes in one file, services/aws_storage_service.py, each the sole owner of its respective raw client. The one detail worth repeating here in a workflow context: index_chunks() auto-creates the OpenSearch index on first use via create_index_if_not_exists() — no separate “provision the index” step exists anywhere else in the codebase.

OpenSearch, Embedding, and LLM Services

The layering that every agent and every ingestion workflow ultimately funnels through:

Agents (9 of them)                    Ingestion workflows
        |                                     |
LLMService.generate()          EmbeddingService.create_embedding()
        |                                     |
ModelService.invoke_text_model()   ModelService.invoke_embedding_model()
        |                                     |
        └─────────────────┬────────────────────┘
                           v
                boto3 bedrock-runtime client (ModelService only)

OpenSearchVectorStore.search_chunks() is what RAGAgent.retrieve() calls after embedding a question — the vector search itself is a kNN query, optionally filtered by document_type:

def search_chunks(self, embedding, k=5, document_type=None):
    knn_clause = {"knn": {"embedding": {"vector": embedding, "k": k}}}
    query = knn_clause if not document_type else {
        "bool": {"must": [knn_clause], "filter": [{"term": {"document_type": document_type}}]}
    }
    response = self.client.search(index=settings.opensearch_index, body={"size": k, "query": query})
    return [...]

The embedding model configured is Titan Embeddings v2 (amazon.titan-embed-text-v2:0); the text-generation model is Claude Haiku 4.5, invoked through a cross-region inference profile ID (us.anthropic.claude-haiku-4-5-20251001-v1:0) rather than a bare model ID, since this particular model requires it.

---

Resources

Every doc in this repository, current as of this guide, grouped by what it covers:

Architecture & system-level

• docs/system_design_updated.md — the full honest system-design writeup: business/data/security/AI/evaluation/production/maintenance questions answered against the real codebase, known gaps, real folder structure
• docs/design-code-general-structure.md, docs/design-code-abstract.md — the AI service layering principle in more depth

LLM/embeddings layer

• docs/bedrock-service.md — line-by-line ModelService/LLMService/EmbeddingService walkthrough
• docs/model-service.md — the architecture diagram of which agent talks to which service

RAG / ingestion / storage

• docs/rag-utils.md — chunk_documents()/create_embeddings_for_chunks() and aws_storage_service.py’s two classes
• docs/aws-storage.md — bulk indexing into OpenSearch, in depth
• docs/load-data.md — every file loader, the dispatch flow, the .xlsx gap
• docs/RDS-glue.md — the Postgres/Glue metadata split, the un-persisted glue_table gap
• docs/workflows.md — both ingestion workflows, step by step, with every field in their return values

Question-answering / agents

• docs/question-graph-agents-prompts.md — the most detailed doc in the repo: every agent, every method, every prompt file, cross-referenced
• docs/routers.md — every API endpoint, what it delegates to, and what it doesn’t clean up

Reliability, memory, testing

• docs/cache_memory_service.md — CacheService/MemoryService, including the cache-miss-doesn’t-repopulate gap
• docs/testing.md — the three-tier testing plan, what’s actually implemented and verified
• docs/evaluation-and-automation.md — the real retrieval-evaluation harness and an honest tooling inventory
• docs/test-methods.md — what mocking is and how it’s used here
• docs/details-settings-clouds.md — dev vs. cloud configuration notes

Narrative / portfolio series

• docs/post-1.md, docs/post-2.md, docs/post-3.md — a three-part blog-style series covering the same ground as this guide, written for a general/portfolio audience rather than as a technical reference

GitHub Repository

GitHub Code: Healthcare Multi-Source Agentic RAG Platform