[Python] How to solve GRAPH_RECURSION_LIMIT error in langgraph and streamlit?

I am new to langgraph and streamlit. I am working on a graph which will takes interview of condidate.

Logic is it will ask user information one by one in self loop.Get all data. Then generate user interview question and ask.

My interview_app.py:

st.set_page_config(page_title="Interview Bot", layout="wide")
st.title("AI Interview Assistant")

# Initialize execution manager
if "execution_manager" not in st.session_state:
st.session_state.execution_manager = GraphExecutionManager()

if "candidates" not in st.session_state:
st.session_state.candidates = {}

if "current_candidate" not in st.session_state:
st.session_state.current_candidate = None


# Initialize LLM and templates
if "llm" not in st.session_state:
st.session_state.llm = ChatGroq(model="openai/gpt-oss-20b", temperature=0.0, streaming=True)

if "candidate_information_need" not in st.session_state:
st.session_state.candidate_information_need = ["name of candidate", "year of passout", "technical_skills"]

# Candidate ID input
candidate_id = st.text_input("Enter your Candidate ID:")

# Start new interview button
if st.button("Start New Interview") and candidate_id:
st.session_state.current_candidate = candidate_id

if candidate_id not in st.session_state.candidates:
# Start async interview
async def start_interview():
thread_id, initial_state = await st.session_state.execution_manager.start_interview(candidate_id)
return thread_id, initial_state

# Run async function
thread_id, initial_state = asyncio.run(start_interview())

st.session_state.candidates[candidate_id] = {
"thread_id": thread_id,
"state": initial_state,
"interview_active": True
}

st.success(f"✅ New Candidate Session Started! Your ID: {candidate_id}")
st.rerun()
else:
st.warning(f"⚠ Candidate ID {candidate_id} already exists, resuming session.")

# Main interview UI
if candidate_id and candidate_id in st.session_state.candidates:
candidate_data = st.session_state.candidates[candidate_id]
current_state = candidate_data["state"]

# Display chat history
for msg in current_state.get("messages", []):
with st.chat_message(msg["role"]):
st.markdown(msg["content"])

# Show current question if waiting for input
if current_state.get("waiting_for_input") and current_state.get("current_question"):
with st.chat_message("assistant"):
st.markdown(current_state["current_question"])

# Handle user input
if current_state.get("waiting_for_input"):
user_key = f"user_input_{candidate_id}_{current_state.get('current_index', 0)}"

if user_input := st.chat_input("Your answer...", key=user_key):
# Process input asynchronously
async def process_input():
return await st.session_state.execution_manager.process_input(
candidate_data["thread_id"],
user_input
)

# Update state
updated_state = asyncio.run(process_input())
st.session_state.candidates[candidate_id]["state"] = updated_state

st.rerun()

# Show completion status
if not current_state.get("waiting_for_input") and current_state.get("current_index", 0) >= len(st.session_state.candidate_information_need):
st.success("✅ Information collection complete!")

else:
st.warning("⚠ Please enter a candidate ID and start the interview.")


Graph Code:

def create_graph():
memory = MemorySaver()

graph = StateGraph(InterviewState)

graph.add_node("ask_information", ask_information)
graph.add_node("generate_interview_question", generate_interview_question)
graph.add_node("ask_questions", ask_questions)

graph.add_edge(START, "ask_information")

graph.add_conditional_edges(
"ask_information",
lambda state: "continue" if state["current_index"] < len(state['candidate_information_need']) else "done",
{
"continue": "ask_information",
"done": "generate_interview_question",
}
)


graph.add_conditional_edges(
"generate_interview_question",
lambda state: "continue" if state["topic_index"] < len(state['candidate_skill_list']) else "done",
{
"continue": "generate_interview_question",
"done": "ask_questions",
}
)

graph.add_conditional_edges(
"ask_questions",
lambda state: "continue" if state.get("question_index", 0) < len(
state["list_of_quetion_per_skill"].get(
state["candidate_skill_list"][state.get("topic_index", 0)-1], []
)
) else "done",
{
"continue": "ask_questions",
"done": END
}
)

return graph.compile(checkpointer=memory)


class GraphExecutionManager:
def __init__(self):
self.graph = create_graph()

async def start_interview(self, candidate_id: str) -> str:
"""Start a new interview session"""
thread_id = str(uuid.uuid4())

initial_state = {
"candidate_information": [],
"candidate_skill_list": [],
"list_of_quetion_per_skill": {},
"candidate_information_need": CANDIDATE_INFORMATION_NEED,
"current_index": 0,
"topic_index": 0,
"question_index": 0,
"messages": [],
"last_bot_message": None,
"last_asked_index": None,
"waiting_for_input": False,
"current_question": "",
"thread_id": thread_id
}

# Execute first step
config = {"configurable": {"thread_id": thread_id}}
result = await self.graph.ainvoke(initial_state, config)

return thread_id, result

async def process_input(self, thread_id: str, user_input: str) -> InterviewState:
"""Process user input and continue graph execution"""
config = {"configurable": {"thread_id": thread_id}}

# Get current state
current_state = self.graph.get_state(config).values

# Process user input
updated_state = await process_user_input(current_state, user_input)

# Continue graph execution
result = await self.graph.ainvoke(updated_state, config)

return result


ask_information() node:

async def ask_information(state: InterviewState) -> InterviewState:
"""Async version that handles user input collection"""
idx = state.get("current_index", 0)
llm = st.session_state.llm
# candidate_information_need = st.session_state.candidate_information_need
candidate_information_need = state['candidate_information_need']

if idx >= len(candidate_information_need):
state["waiting_for_input"] = False
return state

# Generate question if not already asked
if state.get("last_asked_index") != idx:
chat_template = ChatPromptTemplate.from_messages([
HumanMessagePromptTemplate.from_template(
"""
You are an interview assistant named 'deep'.
Politely ask the candidate for <data_to_ask>.
If <information_index> == 2 (technical_skills), tell them to list their skills.

<information_index>: {information_index}
<data_to_ask>: {data_to_ask}
"""
)
])

bot_question = (chat_template | llm).invoke({
"data_to_ask": candidate_information_need[idx],
"information_index": idx
}).content

state["messages"].append({"role": "assistant", "content": bot_question})
state["current_question"] = bot_question # name
state["last_asked_index"] = idx
state["waiting_for_input"] = True

return state

async def process_user_input(state: InterviewState, user_input: str) -> InterviewState:
"""Process user input and update state accordingly"""
idx = state.get("current_index", 0)
llm = st.session_state.llm

state["messages"].append({"role": "user", "content": user_input})

if idx == 2: # Special handling for technical skills
clarify_chain = clarify_template | llm | clarify_parser
try:
parsed = clarify_chain.invoke({"answer": user_input})
state["candidate_skill_list"].extend(parsed.get("skills", []))

if not parsed.get("has_experience_info", True):
# Need follow-up for experience
followup_msg = parsed.get("followup", "Can you provide more details about your experience?")
state["messages"].append({"role": "assistant", "content": followup_msg})
state["current_question"] = followup_msg
state["waiting_for_input"] = True
return state
else:
state["candidate_information"].append(user_input)
except Exception as e:
st.error(f"Error processing skills: {e}")
state["candidate_information"].append(user_input)
else:
state["candidate_information"].append(user_input)

# Move to next question
state["current_index"] = idx + 1
state["last_asked_index"] = None
state["waiting_for_input"] = False

return state


When ever trying to run this code it is asking for id then bellow error:

error:

$ streamlit run interview_app.py

You can now view your Streamlit app in your browser.

Local URL: http://localhost:8501
Network URL: http://192.168.1.14:8501

2025-09-15 14:37:35.031 Uncaught app execution
Traceback (most recent call last):
File "U:\generative_ai_projects\interview_creation\venv\lib\site-packages\streamlit\runtime\scriptrunner\exec_code.py", line 128, in exec_func_with_error_handling
result = func()
File "U:\generative_ai_projects\interview_creation\venv\lib\site-packages\streamlit\runtime\scriptrunner\script_runner.py", line 669, in code_to_exec
exec(code, module.__dict__) # noqa: S102
File "U:\generative_ai_projects\interview_creation\interview_app.py", line 55, in <module>
thread_id, initial_state = asyncio.run(start_interview())
File "U:\generative_ai_projects\interview_creation\venv\lib\asyncio\runners.py", line 44, in run
return loop.run_until_complete(main)
File "U:\generative_ai_projects\interview_creation\venv\lib\asyncio\base_events.py", line 641, in run_until_complete
return future.result()
File "U:\generative_ai_projects\interview_creation\interview_app.py", line 51, in start_interview
thread_id, initial_state = await st.session_state.execution_manager.start_interview(candidate_id)
File "U:\generative_ai_projects\interview_creation\src\graph.py", line 84, in start_interview
result = await self.graph.ainvoke(initial_state, config)
File "U:\generative_ai_projects\interview_creation\venv\lib\site-packages\langgraph\pregel\main.py", line 3112, in ainvoke
async for chunk in self.astream(
File "U:\generative_ai_projects\interview_creation\venv\lib\site-packages\langgraph\pregel\main.py", line 2977, in astream
raise GraphRecursionError(msg)
langgraph.errors.GraphRecursionError: Recursion limit of 25 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/GRAPH_RECURSION_LIMIT


Thanks in advance, I hope to hear from you soon

Continue reading...