LCEL: LangChain Expression Language Overview

# Old way (v0.0.x — deprecated)
from langchain.chains import LLMChain
chain = LLMChain(llm=model, prompt=prompt)

# LCEL way (current)
chain = prompt | model | output_parser

# All four methods work on any Runnable (prompt, model, parser, chain, retriever, etc.)

# .invoke() — synchronous, single input
result = chain.invoke({"drug": "warfarin"})

# .stream() — stream output tokens as they arrive
for chunk in chain.stream({"drug": "warfarin"}):
    print(chunk, end="", flush=True)

# .batch() — process multiple inputs efficiently
results = chain.batch([
    {"drug": "warfarin"},
    {"drug": "metformin"},
    {"drug": "aspirin"},
])

# .ainvoke() / .astream() / .abatch() — async equivalents
import asyncio
result = await chain.ainvoke({"drug": "warfarin"})

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o", temperature=0)
parser = StrOutputParser()

# A simple chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a clinical pharmacist."),
    ("human", "Explain the mechanism of action of {drug}."),
])

chain = prompt | model | parser

# The | operator creates a RunnableSequence
# Data flows: {"drug": "..."} → prompt → ChatMessages → model → AIMessage → parser → str

result = chain.invoke({"drug": "warfarin"})
print(result)   # "Warfarin inhibits vitamin K epoxide reductase..."

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

# Prompt: dict → list[BaseMessage]
prompt_output = prompt.invoke({"drug": "warfarin"})
# → [SystemMessage("You are..."), HumanMessage("Explain warfarin...")]

# Model: list[BaseMessage] → AIMessage
model_output = model.invoke(prompt_output)
# → AIMessage(content="Warfarin inhibits...", response_metadata={...})

# StrOutputParser: AIMessage → str
parser_output = parser.invoke(model_output)
# → "Warfarin inhibits..."

# Full chain: dict → str
chain_output = chain.invoke({"drug": "warfarin"})
# → "Warfarin inhibits..."

from langchain_core.runnables import RunnablePassthrough

# Problem: chain transforms input to output, losing original input
# Solution: use RunnablePassthrough to carry original input forward

explain_prompt = ChatPromptTemplate.from_template(
    "Explain {drug} mechanism in one sentence."
)
monitor_prompt = ChatPromptTemplate.from_template(
    """Drug: {drug}
Mechanism: {mechanism}

What clinical monitoring is required?"""
)

# Build a two-step chain that keeps "drug" available for step 2
chain = (
    # Step 1: produce mechanism (keep original dict alongside)
    RunnablePassthrough.assign(mechanism=explain_prompt | model | parser)
    # Now we have: {"drug": "...", "mechanism": "..."}
    | monitor_prompt
    | model
    | parser
)

result = chain.invoke({"drug": "warfarin"})
# "drug" and "mechanism" are both available in step 2

from langchain_core.runnables import RunnableParallel

# Run multiple chains simultaneously, combine results
mechanism_chain = prompt | model | parser

side_effects_prompt = ChatPromptTemplate.from_template(
    "List the top 3 side effects of {drug}."
)
side_effects_chain = side_effects_prompt | model | parser

# Both chains run concurrently
parallel_chain = RunnableParallel(
    mechanism=mechanism_chain,
    side_effects=side_effects_chain,
)

result = parallel_chain.invoke({"drug": "metformin"})
# {"mechanism": "...", "side_effects": "..."}
# Latency ≈ max(chain1_time, chain2_time), not chain1_time + chain2_time

from langchain_core.runnables import RunnableLambda

def format_clinical_query(inputs: dict) -> dict:
    """Pre-process inputs before sending to model."""
    drug = inputs["drug"].strip().lower()
    patient_age = inputs.get("patient_age", "adult")
    return {
        "query": f"For a {patient_age} patient, explain {drug} dosing and monitoring.",
    }

def post_process_response(response: str) -> dict:
    """Post-process model output into structured form."""
    lines = response.split("\n")
    return {
        "summary": lines[0] if lines else "",
        "full_text": response,
        "word_count": len(response.split()),
    }

clinical_prompt = ChatPromptTemplate.from_template("{query}")

chain = (
    RunnableLambda(format_clinical_query)
    | clinical_prompt
    | model
    | parser
    | RunnableLambda(post_process_response)
)

result = chain.invoke({"drug": "warfarin", "patient_age": "elderly"})
print(result["summary"])

import sys

# Stream tokens as they arrive
for chunk in chain.stream({"drug": "warfarin"}):
    print(chunk, end="", flush=True)
print()  # newline after stream ends

# Async streaming for web servers (FastAPI, etc.)
import asyncio

async def stream_response(drug: str):
    full_response = ""
    async for chunk in chain.astream({"drug": drug}):
        full_response += chunk
        yield chunk    # Server-sent event
    return full_response

# Stream with intermediate values (debug all steps)
for step in chain.stream({"drug": "warfarin"}, stream_mode="values"):
    print("--- Step output ---")
    print(step)

from langchain_core.runnables import RunnableConfig

# Add retries with fallback
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

# Fallback to another chain on error
fallback_chain = (
    ChatPromptTemplate.from_template("{drug} overview:")
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | parser
)

chain_with_fallback = chain.with_fallbacks([fallback_chain])

# Configure at runtime (tags, metadata, callbacks)
result = chain.invoke(
    {"drug": "warfarin"},
    config=RunnableConfig(
        tags=["clinical", "drug-info"],
        metadata={"session_id": "abc123"},
    ),
)