Task 5: Dynamic Difficulty with Agents

A good interviewer adapts. If a candidate struggles, they dial back. If they excel, they push harder.

Agent vs Chain

Agent Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        ReAct Agent                               │
│                                                                  │
│  Input -> Think -> Act -> Observe -> Think -> Act -> ... -> End │
│              │                           │                       │
│              v                           v                       │
│         ┌─────────┐               ┌─────────────┐               │
│         │ Tools   │               │ Observations │              │
│         │ - Easy Q│               │ from tools   │              │
│         │ - Hard Q│               │              │              │
│         │ - Score │               │              │              │
│         │ - Skip  │               │              │              │
│         └─────────┘               └─────────────┘               │
└─────────────────────────────────────────────────────────────────┘

Step 1: Define Tools

Tools are functions the agent can call:

# agents/tools.py
from langchain_core.tools import tool
from typing import Literal

# Track state (in production, use proper state management)
interview_state = {
    "difficulty": "medium",
    "scores": [],
    "questions_asked": 0,
    "topics_covered": []
}

@tool
def get_easy_question(topic: str) -> str:
    """Get an easy/foundational question about a topic.
    Use when candidate is struggling (avg score < 5)."""

    questions = {
        "python": "What are the basic data types in Python?",
        "async": "What's the difference between sync and async code?",
        "databases": "What is a primary key in a database?",
        "api": "What does REST stand for?",
    }
    return questions.get(topic.lower(), f"Explain the basics of {topic}")

@tool
def get_medium_question(topic: str) -> str:
    """Get a medium difficulty question about a topic.
    Use for standard assessment."""

    questions = {
        "python": "Explain how Python's GIL affects multi-threading.",
        "async": "How does asyncio's event loop work?",
        "databases": "When would you use an index, and what are the tradeoffs?",
        "api": "How would you design pagination for a REST API?",
    }
    return questions.get(topic.lower(), f"Explain best practices for {topic}")

@tool
def get_hard_question(topic: str) -> str:
    """Get a challenging question about a topic.
    Use when candidate is performing well (avg score > 7)."""

    questions = {
        "python": "How would you implement a custom metaclass? Give a use case.",
        "async": "Design an async rate limiter using asyncio primitives.",
        "databases": "Explain transaction isolation levels and when to use each.",
        "api": "How would you handle eventual consistency in a microservices API?",
    }
    return questions.get(topic.lower(), f"Explain advanced concepts in {topic}")

@tool
def evaluate_answer(answer: str, question: str) -> dict:
    """Evaluate a candidate's answer and return score with feedback.
    Always use this after receiving an answer."""

    # In production, this would call our evaluator chain
    # Simplified for demonstration
    word_count = len(answer.split())

    if word_count < 10:
        score = 3
        feedback = "Answer too brief, needs more detail"
    elif word_count < 30:
        score = 5
        feedback = "Decent answer, could elaborate more"
    elif word_count < 60:
        score = 7
        feedback = "Good explanation with solid understanding"
    else:
        score = 8
        feedback = "Comprehensive answer showing expertise"

    interview_state["scores"].append(score)

    return {
        "score": score,
        "feedback": feedback,
        "avg_score": sum(interview_state["scores"]) / len(interview_state["scores"])
    }

@tool
def get_interview_status() -> dict:
    """Get current interview status including average score and questions asked.
    Use this to decide difficulty level."""

    scores = interview_state["scores"]
    return {
        "questions_asked": len(scores),
        "average_score": sum(scores) / len(scores) if scores else 0,
        "current_difficulty": interview_state["difficulty"],
        "should_increase_difficulty": len(scores) >= 2 and sum(scores[-2:]) / 2 > 7,
        "should_decrease_difficulty": len(scores) >= 2 and sum(scores[-2:]) / 2 < 5
    }

@tool
def set_difficulty(level: Literal["easy", "medium", "hard"]) -> str:
    """Set the interview difficulty level.
    Use based on candidate performance."""

    interview_state["difficulty"] = level
    return f"Difficulty set to {level}"

Step 2: Create the Agent

# agents/coach.py
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain import hub

from agents.tools import (
    get_easy_question,
    get_medium_question,
    get_hard_question,
    evaluate_answer,
    get_interview_status,
    set_difficulty
)

def create_interview_agent():
    """Create an adaptive interview agent."""

    # Define available tools
    tools = [
        get_easy_question,
        get_medium_question,
        get_hard_question,
        evaluate_answer,
        get_interview_status,
        set_difficulty
    ]

    # Custom prompt for interview context
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are an adaptive technical interview coach.

Your job is to conduct a fair but thorough interview that adjusts to the candidate's level.

Strategy:
1. Start with medium difficulty
2. If avg_score > 7 for 2+ questions, increase difficulty
3. If avg_score < 5 for 2+ questions, decrease difficulty
4. Always evaluate answers before asking next question
5. Cover different topics throughout the interview

Be encouraging but honest. After each answer, briefly acknowledge it before moving on.

Current interview context:
- Interview type: {interview_type}
- Position: {position}
- Topics to cover: {topics}
"""),
        MessagesPlaceholder(variable_name="chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])

    # Create the agent
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

    agent = create_react_agent(llm, tools, prompt)

    # Wrap in executor
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,  # Set False in production
        max_iterations=5,
        handle_parsing_errors=True
    )

    return agent_executor

Step 3: Run Adaptive Interview

# main.py
from agents.coach import create_interview_agent

def run_adaptive_interview():
    agent = create_interview_agent()

    interview_config = {
        "interview_type": "technical",
        "position": "Senior Python Developer",
        "topics": "Python, async programming, databases, API design"
    }

    print("=" * 50)
    print("Adaptive AI Interview Coach")
    print("=" * 50)
    print("The interviewer will adjust difficulty based on your performance.")
    print("Type 'quit' to exit\n")

    chat_history = []

    # Start interview
    result = agent.invoke({
        **interview_config,
        "chat_history": chat_history,
        "input": "Start the interview. Ask the first question."
    })

    print(f"\nInterviewer: {result['output']}\n")
    chat_history.append(("ai", result['output']))

    while True:
        answer = input("You: ")
        if answer.lower() == 'quit':
            break

        chat_history.append(("human", answer))

        # Agent decides what to do next
        result = agent.invoke({
            **interview_config,
            "chat_history": chat_history,
            "input": f"The candidate answered: {answer}\n\nEvaluate their answer and ask the next question, adjusting difficulty if needed."
        })

        print(f"\nInterviewer: {result['output']}\n")
        chat_history.append(("ai", result['output']))

    print("\nInterview complete!")

if __name__ == "__main__":
    run_adaptive_interview()

What Happens During Execution

When verbose=True, you'll see the agent's reasoning:

> Entering new AgentExecutor chain...

Thought: I need to evaluate the candidate's answer first, then check the interview status to decide on difficulty.

Action: evaluate_answer
Action Input: {"answer": "Lists are mutable, tuples immutable...", "question": "Explain lists vs tuples"}

Observation: {"score": 6, "feedback": "Decent answer, could elaborate more", "avg_score": 6.0}

Thought: Score is 6, which is medium range. I should check the overall status.

Action: get_interview_status
Action Input: {}

Observation: {"questions_asked": 3, "average_score": 6.3, "should_increase_difficulty": false}

Thought: Average is 6.3, staying at medium difficulty. Let me ask the next question.

Action: get_medium_question
Action Input: {"topic": "async"}

Observation: "How does asyncio's event loop work?"

Final Answer: Good answer! You correctly identified the key difference between lists and tuples. For senior positions, I'd also expect discussion of when to use each, like tuples for dictionary keys. Let's move on - how does asyncio's event loop work under the hood?

Step 4: Using LangGraph for Complex Flows

For more complex agent behavior, use LangGraph:

# agents/coach_langgraph.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

class InterviewState(TypedDict):
    messages: Annotated[List[str], operator.add]
    current_difficulty: str
    scores: List[int]
    current_topic_index: int
    topics: List[str]

def should_continue(state: InterviewState) -> str:
    """Decide whether to continue or end the interview."""
    if state["current_topic_index"] >= len(state["topics"]):
        return "end"
    return "continue"

def adjust_difficulty(state: InterviewState) -> InterviewState:
    """Adjust difficulty based on recent performance."""
    scores = state["scores"]
    if len(scores) >= 2:
        recent_avg = sum(scores[-2:]) / 2
        if recent_avg > 7:
            state["current_difficulty"] = "hard"
        elif recent_avg < 5:
            state["current_difficulty"] = "easy"
        else:
            state["current_difficulty"] = "medium"
    return state

def ask_question(state: InterviewState) -> InterviewState:
    """Generate and ask the next question."""
    topic = state["topics"][state["current_topic_index"]]
    difficulty = state["current_difficulty"]

    # Generate question based on difficulty
    # ... (use appropriate tool)

    state["current_topic_index"] += 1
    return state

# Build the graph
workflow = StateGraph(InterviewState)
workflow.add_node("adjust_difficulty", adjust_difficulty)
workflow.add_node("ask_question", ask_question)
workflow.add_conditional_edges("ask_question", should_continue, {
    "continue": "adjust_difficulty",
    "end": END
})
workflow.add_edge("adjust_difficulty", "ask_question")
workflow.set_entry_point("adjust_difficulty")

app = workflow.compile()

Key Takeaways

• Tools are functions the agent can call
• ReAct agents think, act, observe in a loop
• AgentExecutor manages the agent lifecycle
• Adaptive behavior comes from tool design and prompt engineering
• LangGraph offers more control for complex state machines
• Verbose mode helps debug agent reasoning

Next Up