Overview
The DocMindState TypedDict defines the shared state that flows through all nodes in the DocMind workflow. It tracks the query, intermediate results, retry logic, and execution history.
Type Definition
from typing import TypedDict, List, Optional, Dict
class DocMindState(TypedDict):
query: str
decomposition: Optional[Dict]
retrieved_sections: List[Dict]
generated_response: Optional[str]
judge_verdict: Optional[Dict]
final_output: Optional[str]
retry_count: int
node_history: List[str]
State Fields
query
The original user query that initiated the workflow
decompose_node, retrieve_node
Example:
state["query"] = "How do I configure authentication in DocMind?"
decomposition
Structured breakdown of the query into sub-queries and intent
decompose_node
Used by: retrieve_node
Initial value: None
Example:
state["decomposition"] = {
"intent": "configuration",
"sub_queries": [
"authentication setup",
"auth configuration options"
]
}
retrieved_sections
List of documentation sections retrieved based on the query
retrieve_node
Used by: generate_node, judge_node
Initial value: []
Example:
state["retrieved_sections"] = [
{
"title": "Authentication Setup",
"content": "To configure authentication...",
"source": "auth.md",
"relevance_score": 0.95
},
{
"title": "OAuth Configuration",
"content": "DocMind supports OAuth...",
"source": "oauth.md",
"relevance_score": 0.87
}
]
generated_response
Response generated from the retrieved documentation sections
generate_node
Used by: judge_node, output_node
Initial value: None
Example:
state["generated_response"] = "To configure authentication in DocMind, you need to..."
judge_verdict
Evaluation verdict from the LLM judge assessing response quality
judge_node
Used by: should_retry (conditional logic), output_node
Initial value: None
Structure:
{
"is_hallucinated": bool, # Whether response contains hallucinations
"should_return": bool, # Whether response is acceptable
"reasoning": str # Explanation of the verdict
}
state["judge_verdict"] = {
"is_hallucinated": False,
"should_return": True,
"reasoning": "Response is grounded in retrieved sections"
}
final_output
The final output returned to the user
output_node
Used by: Workflow caller
Initial value: None
Possible values:
- The
generated_response if judge_verdict.should_return is True
"Unable to provide a confident response. Please rephrase your query." otherwise
Example:
state["final_output"] = "To configure authentication in DocMind, you need to..."
retry_count
Number of times the workflow has retried due to hallucination detection
judge_node (incremented when hallucination detected)
Used by: should_retry (conditional logic)
Initial value: 0
Maximum value: 2 (workflow stops retrying after 2 attempts)
Increment logic:
if verdict.get("is_hallucinated", False):
state["retry_count"] = state.get("retry_count", 0) + 1
# Initial state
state["retry_count"] = 0
# After first hallucination
state["retry_count"] = 1 # Will retry
# After second hallucination
state["retry_count"] = 2 # Will NOT retry (max reached)
node_history
Ordered list of node names that have been executed
[]
Example without retries:
state["node_history"] = [
"decompose",
"retrieve",
"generate",
"judge",
"output"
]
state["node_history"] = [
"decompose",
"retrieve",
"generate",
"judge", # First attempt - hallucination detected
"retrieve", # Retry begins
"generate",
"judge", # Second attempt - success
"output"
]
State Flow Example
Successful Flow (No Retries)
# Initial state
state = {
"query": "How do I setup DocMind?",
"decomposition": None,
"retrieved_sections": [],
"generated_response": None,
"judge_verdict": None,
"final_output": None,
"retry_count": 0,
"node_history": []
}
# After decompose_node
state["decomposition"] = {"intent": "setup", "sub_queries": [...]}
state["node_history"] = ["decompose"]
# After retrieve_node
state["retrieved_sections"] = [{"title": "Setup Guide", ...}]
state["node_history"] = ["decompose", "retrieve"]
# After generate_node
state["generated_response"] = "To setup DocMind, first..."
state["node_history"] = ["decompose", "retrieve", "generate"]
# After judge_node
state["judge_verdict"] = {"is_hallucinated": False, "should_return": True}
state["node_history"] = ["decompose", "retrieve", "generate", "judge"]
# After output_node
state["final_output"] = "To setup DocMind, first..."
state["node_history"] = ["decompose", "retrieve", "generate", "judge", "output"]
Flow with Retry
# After first judge_node (hallucination detected)
state["judge_verdict"] = {"is_hallucinated": True, "should_return": False}
state["retry_count"] = 1
state["node_history"] = ["decompose", "retrieve", "generate", "judge"]
# should_retry returns "retry" -> back to retrieve_node
# After second retrieve_node
state["retrieved_sections"] = [{"title": "Different sections", ...}]
state["node_history"] = ["decompose", "retrieve", "generate", "judge", "retrieve"]
# After second generate_node
state["generated_response"] = "Updated response..."
state["node_history"] = ["decompose", "retrieve", "generate", "judge", "retrieve", "generate"]
# After second judge_node (success)
state["judge_verdict"] = {"is_hallucinated": False, "should_return": True}
state["retry_count"] = 1 # Still 1 (only incremented on hallucination)
state["node_history"] = [..., "generate", "judge"]
# should_retry returns "output" -> proceeds to output_node
Usage
Creating Initial State
from state_types import DocMindState
initial_state: DocMindState = {
"query": "How do I use the API?",
"decomposition": None,
"retrieved_sections": [],
"generated_response": None,
"judge_verdict": None,
"final_output": None,
"retry_count": 0,
"node_history": []
}
Accessing State in Nodes
async def custom_node(state: DocMindState) -> DocMindState:
# Read from state
query = state["query"]
retry_count = state.get("retry_count", 0)
# Modify state
state["custom_field"] = "value"
state["node_history"] = state.get("node_history", []) + ["custom"]
return state