from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
class AgentState(TypedDict):
question: str
documents: List[Document]
generation: str
retry_count: int
rewritten_question: str
def retrieve(state: AgentState) -> dict:
"""Retrieve documents for the current question."""
query = state.get("rewritten_question") or state["question"]
docs = retriever.invoke(query)
return {"documents": docs}
def grade_documents(state: AgentState) -> str:
"""Evaluate if retrieved documents are relevant."""
question = state["question"]
docs = state["documents"]
# Use LLM to evaluate relevance
relevant = evaluate_relevance(question, docs)
if relevant:
return "generate"
elif state.get("retry_count", 0) >= 2:
return "generate" # Give up retrying, use what we have
else:
return "rewrite"
def rewrite_query(state: AgentState) -> dict:
"""Rewrite the question for better retrieval."""
question = state["question"]
retry = state.get("retry_count", 0) + 1
rewritten = llm.invoke(
f"Rewrite this question to be more specific for document search: {question}"
).content
return {"rewritten_question": rewritten, "retry_count": retry}
def generate(state: AgentState) -> dict:
"""Generate answer from retrieved documents."""
docs_text = "\n\n".join(d.page_content for d in state["documents"])
answer = llm.invoke(
f"Context: {docs_text}\n\nQuestion: {state['question']}\n\nAnswer:"
).content
return {"generation": answer}
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("rewrite", rewrite_query)
workflow.add_node("generate", generate)
workflow.set_entry_point("retrieve")
workflow.add_conditional_edges(
"retrieve",
grade_documents,
{"generate": "generate", "rewrite": "rewrite"}
)
workflow.add_edge("rewrite", "retrieve") # Retry after rewriting
workflow.add_edge("generate", END)
app = workflow.compile()
result = app.invoke({"question": "How does Django connect to databases?"})