RunnableLambda & Custom Logic

import html
import unicodedata
from langchain_core.runnables import RunnableLambda

def sanitize_text(text: str) -> str:
  # 1. Remove HTML artifacts (e.g., &)
  text = html.unescape(text)
  # 2. Normalize unicode to avoid 'invisible' chars or confusing glyphs
  text = unicodedata.normalize("NFKC", text)
  # 3. Hard limit on length to prevent context overflow attacks
  return text[:10000].strip()

# The Sentinel Layer
sanitization_layer = RunnableLambda(sanitize_text)

def canonicalize_query(query: str) -> str:
  mappings = {
      "mbp": "MacBook Pro",
      "ai": "Artificial Intelligence",
      "js": "JavaScript"
  }
  # Deterministic dictionary lookup is O(1) and 100% accurate.
  return mappings.get(query.lower(), query)

resolver_layer = RunnableLambda(canonicalize_query)

def robust_int_cast(text: str) -> int:
  # Handle verbal numbers if needed, or strip 'stars' suffix
  text = text.lower().replace("stars", "").strip()
  try:
      return int(text)
  except ValueError:
      # Fallback logic defined in CODE, not hallucinated.
      # Maybe return a default, or raise a specific error for retry.
      return 0

casting_layer = RunnableLambda(robust_int_cast)

import json
import re

def extract_json(text: str) -> dict:
  # Regex to find the first { and the last }
  match = re.search(r"{.*}", text, re.DOTALL)
  if not match:
      raise ValueError("No JSON found")
  return json.loads(match.group())

json_cleaner = RunnableLambda(extract_json)

class OutOfScopeError(Exception):
  pass

def circuit_breaker(query: str):
  forbidden_terms = ["medical", "doctor", "prescription"]
  if any(term in query.lower() for term in forbidden_terms):
      # Raise an error to stop execution immediately
      raise OutOfScopeError("This bot handles legal queries only.")
  return query

# Usage: Chain stops here if invalid. No GPU tokens wasted.
guard_layer = RunnableLambda(circuit_breaker)

# Pass "summary" if text is long, else pass raw "text"
# This "Micro-Branching" handles edge cases linearly.
smart_router = RunnableLambda(
  lambda x: x['summary'] if len(x['text']) > 1000 else x['text']
)

# Adapter: Maps specific PDF output to generic prompt inputs
adapter = RunnableLambda(lambda x: {
  "context": x["pdf_content"],
  "question": x["user_message_v2"]
})

# seamless composition
chain = pdf_loader | adapter | summarization_chain

# Strip the object wrapper, extract the string payload
unwrap = RunnableLambda(lambda msg: msg.content)

# Now the output is compatible with simple string tools
chain = model | unwrap | simple_tool

def reliable_math(inputs):
  # Deterministic math. Cannot hallucinate.
  salary = inputs["salary"]
  rate = inputs["tax_rate"]
  return salary * (1 - rate)

math_layer = RunnableLambda(reliable_math)

def deterministic_sort(items):
  # Sort by price ascending, then rating descending
  return sorted(items, key=lambda x: (x['price'], -x['rating']))

sorter = RunnableLambda(deterministic_sort)

def calculate_shipping(user_profile):
  # Complex proprietary logic
  if user_profile['state'] == 'CA' and user_profile['spend'] > 100:
      return 0
  return 15.99

# The prompt receives: "Shipping Cost: $0"
# It doesn't need to know WHY. It just reports it.
shipping_calculator = RunnableLambda(calculate_shipping)

from datetime import datetime

# The model now knows exactly what "Today" means
time_injector = RunnableLambda(lambda _: datetime.now().isoformat())

chain = (
  RunnablePassthrough.assign(current_time=time_injector)
  | prompt # Prompt can now use {current_time} inside the template
  | model
)

import aiohttp

async def fetch_weather(city):
  async with aiohttp.ClientSession() as session:
      async with session.get(f"https://api.weather.com/{city}") as resp:
          return await resp.json()

# LangChain automatically handles the async execution
weather_fetcher = RunnableLambda(fetch_weather)

# Prints state to console and returns it unchanged
debug_probe = RunnableLambda(lambda x: print(f"DEBUG: {x}") or x)

chain = step1 | debug_probe | step2