Module 5 Hands-On Coding Lab & Practical Scenarios
Master the practical mechanics of stateful agent systems: implementing function calling dispatchers, cyclic LangGraph state machines, and autonomous self-correction loops.
🛠️ Challenge 1: The Function Calling Dispatcher
Scenario
An LLM emits tool call requests in JSON format. Build a safe dispatcher that validates argument types against a schema and executes the corresponding Python function.
🔍 Click to view Solution & Architecture Explanation
import json
from typing import Dict, Any, Callable
class SafeToolDispatcher:
def __init__(self):
self.registry: Dict[str, Callable] = {}
def register(self, name: str, func: Callable):
self.registry[name] = func
def execute_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
if tool_name not in self.registry:
return {"status": "ERROR", "error": f"Tool '{tool_name}' is not registered."}
try:
result = self.registry[tool_name](**arguments)
return {"status": "SUCCESS", "output": result}
except TypeError as e:
return {"status": "ERROR", "error": f"Invalid arguments: {e}"}
# Test functions
dispatcher = SafeToolDispatcher()
dispatcher.register("inspect_s3_bucket", lambda bucket, prefix="": f"Objects in s3://{bucket}/{prefix}: ['data_01.parquet']")
res = dispatcher.execute_tool_call("inspect_s3_bucket", {"bucket": "analytics-prod", "prefix": "raw/"})
print("Tool Execution Result:", res)
🛠️ Challenge 2: A Cyclic LangGraph-Style State Machine
Scenario
Model an iterative loop where an agent attempts a task, evaluates if the result meets a quality threshold, and loops back if needed (up to 3 iterations).
🔍 Click to view Solution & Architecture Explanation
from typing import Dict, Any
class StateMachineAgent:
def __init__(self, max_iterations: int = 3):
self.max_iterations = max_iterations
def run(self, initial_query: str) -> Dict[str, Any]:
state = {
"query": initial_query,
"iteration": 0,
"quality_score": 0.0,
"status": "IN_PROGRESS"
}
while state["status"] == "IN_PROGRESS":
state["iteration"] += 1
print(f"[Iteration {state['iteration']}] Generating & evaluating response...")
# Simulated quality improvement over iterations
state["quality_score"] += 0.35
if state["quality_score"] >= 0.8:
state["status"] = "COMPLETED"
state["final_answer"] = f"High-confidence verified answer (Score: {state['quality_score']:.2f})"
elif state["iteration"] >= self.max_iterations:
state["status"] = "MAX_RETRIES_REACHED"
state["final_answer"] = "Best-effort answer returned."
return state
agent = StateMachineAgent(max_iterations=3)
result = agent.run("Optimize multi-table join")
print("
Final State Machine Result:", result)
Find this content helpful?
☕ Buy me a coffee