Module 4 Hands-On Coding Lab & Practical Scenarios
Master the practical engineering of LangChain LCEL and LlamaIndex with runnable Python implementations.
🛠️ Challenge 1: Implementing an LCEL Parallel Pipeline
Scenario
When a user asks for query optimization advice, you must concurrently:
- Extract referenced table names from the SQL query.
- Estimate estimated row count from table statistics.
- Combine the outputs into an execution plan recommendation.
Task
Implement an LCEL-style chain using the pipe operator (|) and parallel execution:
🔍 Click to view Solution & Architecture Explanation
from typing import Dict, Any
class Runnable:
def __init__(self, fn):
self.fn = fn
def __or__(self, other):
return Runnable(lambda x: other.invoke(self.invoke(x)))
def invoke(self, x: Any) -> Any:
return self.fn(x)
class ParallelRunnable(Runnable):
def __init__(self, **kwargs):
self.branches = kwargs
def invoke(self, x: Any) -> Dict[str, Any]:
# Concurrently evaluate branches
return {key: branch.invoke(x) for key, branch in self.branches.items()}
# Define pipeline components
extract_tables = Runnable(lambda q: ["orders", "customers"] if "join" in q.lower() else ["orders"])
estimate_rows = Runnable(lambda q: 50_000_000 if "orders" in q.lower() else 10_000)
combine_analysis = ParallelRunnable(
tables=extract_tables,
estimated_rows=estimate_rows
)
format_advice = Runnable(lambda res: (
f"Query touches tables {res['tables']} with estimated {res['estimated_rows']:,} rows. "
f"Recommendation: {'Use Broadcast Join' if res['estimated_rows'] < 100_000 else 'Use Sort-Merge Join'}."
))
# Compose the chain
analysis_chain = combine_analysis | format_advice
sql = "SELECT * FROM orders JOIN customers ON orders.c_id = customers.id"
print("=== Pipeline Analysis ===")
print(analysis_chain.invoke(sql))
🛠️ Challenge 2: Custom Runnable with Exponential Backoff Retry
Scenario
Cloud LLM APIs periodically throw HTTP 429 (Rate Limit) or HTTP 503 (Service Unavailable) errors. Build a custom wrapper that catches transient errors and retries with exponential backoff.
Task
Implement ResilientRunnable:
🔍 Click to view Solution & Architecture Explanation
import time
from typing import Callable, Any
class ResilientRunnable:
def __init__(self, func: Callable[[Any], Any], max_retries: int = 3, base_delay: float = 0.5):
self.func = func
self.max_retries = max_retries
self.base_delay = base_delay
def invoke(self, input_val: Any) -> Any:
attempt = 0
while True:
try:
return self.func(input_val)
except Exception as e:
attempt += 1
if attempt > self.max_retries:
raise RuntimeError(f"Operation failed after {self.max_retries} retries. Error: {e}")
delay = self.base_delay * (2 ** (attempt - 1))
print(f"[Warning] Transient failure: {e}. Retrying in {delay:.2f}s (Attempt {attempt}/{self.max_retries})...")
time.sleep(delay)
# Example demonstration with transient failure
counter = 0
def flaky_api_call(prompt: str):
global counter
counter += 1
if counter < 3:
raise ConnectionResetError("HTTP 503: Service Unavailable")
return f"Successfully processed prompt: '{prompt}'"
resilient_call = ResilientRunnable(flaky_api_call, max_retries=3)
result = resilient_call.invoke("SELECT count(*) FROM parquet_data")
print("Result:", result)
Find this content helpful?
☕ Buy me a coffee