AskAiml
Sign in Get started

The agent loop and LangChain LCEL

What an agent loop actually is, and how LangChain expression language composes runnables around it.

🤖 Module 3 7 min read Not started

Why this matters

Every agent — no matter how fancy the framework — runs a variation of the same loop: think → act → observe → repeat. If you can write that loop yourself in 50 lines, you understand more than 80% of the candidates applying for the same job. LangChain LCEL is the building-block layer the rest of the ecosystem composes.

Learning objectives

  1. Write a manual agent loop with no framework.
  2. Use LangChain LCEL (the | pipe) to compose chains.
  3. Build a clean RAG chain with LCEL.
  4. Use RunnableLambda, RunnableParallel, RunnableBranch.

1. The 50-line agent loop (you must internalise this)

python
import json
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Callable

client = OpenAI()

class Tool(BaseModel):
    name: str
    description: str
    args_schema: type[BaseModel]
    fn: Callable

def to_openai_tool(t: Tool):
    return {"type": "function", "function": {
        "name": t.name, "description": t.description,
        "parameters": t.args_schema.model_json_schema(), "strict": True
    }}

def run_agent(user_msg: str, tools: list[Tool], system: str, max_steps: int = 8):
    registry = {t.name: t for t in tools}
    history = [
        {"role": "system", "content": system},
        {"role": "user", "content": user_msg},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=history,
            tools=[to_openai_tool(t) for t in tools],
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        history.append(msg.model_dump(exclude_unset=True))
        if not msg.tool_calls:
            return msg.content
        for call in msg.tool_calls:
            tool = registry[call.function.name]
            args = tool.args_schema(**json.loads(call.function.arguments))
            try:
                result = tool.fn(args)
            except Exception as e:
                result = {"error": str(e)[:200]}
            history.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result, default=str),
            })
    return "ran out of steps"

That is it. Frameworks add: state typing, multi-agent, durability, observability, evals, sub-graphs. The core idea above does not change.

2. LangChain LCEL — the pipe that composes everything

LCEL stands for LangChain Expression Language. Components implement a Runnable interface: they have .invoke(), .stream(), .batch(), .ainvoke(). You compose them with |.

python
# uv add langchain langchain-openai
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise assistant."),
    ("user", "{question}"),
])
llm = ChatOpenAI(model="gpt-4.1-mini")
chain = prompt | llm | StrOutputParser()

print(chain.invoke({"question": "Why is the sky blue?"}))

Why |? It is just __or__ overloaded. prompt | llm returns a new Runnable that does both. Composable. Streamable. Async-ready out of the box.

Streaming any chain

python
for chunk in chain.stream({"question": "Explain RAG."}):
    print(chunk, end="", flush=True)

Batching for evals (free parallelism)

python
results = chain.batch([
    {"question": "What is RAG?"},
    {"question": "What is MCP?"},
    {"question": "What is LangGraph?"},
], config={"max_concurrency": 5})

batch runs in parallel. abatch is async. Use it everywhere in evals.

Async

python
import asyncio
async def go():
    return await chain.ainvoke({"question": "Hi"})
asyncio.run(go())

3. The LCEL building blocks you actually use

RunnableLambda — wrap any function as a Runnable

python
from langchain_core.runnables import RunnableLambda

def upper(text: str) -> str: return text.upper()
chain = RunnableLambda(upper) | (lambda x: x[::-1])  # second one auto-wrapped
chain.invoke("hello")  # -> "OLLEH"

RunnableParallel — fan out to multiple chains, merge results

python
from langchain_core.runnables import RunnableParallel

joke_chain = ChatPromptTemplate.from_template("Tell me a joke about {topic}") | llm | StrOutputParser()
fact_chain = ChatPromptTemplate.from_template("State a fact about {topic}") | llm | StrOutputParser()

ensemble = RunnableParallel(joke=joke_chain, fact=fact_chain)
print(ensemble.invoke({"topic": "the moon"}))
# {"joke": "...", "fact": "..."}

This is "do many things at once," which the next chain step can read as {"joke": ..., "fact": ...}.

RunnableBranch — conditional routing

python
from langchain_core.runnables import RunnableBranch

router = RunnableBranch(
    (lambda x: "billing" in x["q"].lower(), billing_chain),
    (lambda x: "tech" in x["q"].lower(), tech_chain),
    default_chain,
)
router.invoke({"q": "my bill is wrong"})

Equivalent of an if/elif/else over chains.

RunnablePassthrough — pass input through unchanged

Useful for shape-shifting input dicts:

python
from langchain_core.runnables import RunnablePassthrough

retriever_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | rag_prompt | llm | StrOutputParser()
)

Reads "make a dict where context is the retriever's output for the input, and question is the input itself."

4. A production-quality RAG chain in LCEL

python
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. Vector store
emb = OpenAIEmbeddings(model="text-embedding-3-small")
vectordb = Chroma(persist_directory="./chroma_db", collection_name="docs", embedding_function=emb)
retriever = vectordb.as_retriever(search_type="similarity", search_kwargs={"k": 5})

# 2. Prompt
rag_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "Answer ONLY using the provided context. If insufficient, say 'I do not know based on the provided documents.'"
     "\nCITE sources by their index in [n]."),
    ("user", "Context:\n{context}\n\nQuestion: {question}"),
])

# 3. Helper to format docs
def format_docs(docs):
    return "\n\n".join(f"[{i+1}] {d.page_content}" for i, d in enumerate(docs))

# 4. The chain
llm = ChatOpenAI(model="gpt-4.1-mini")
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

print(rag_chain.invoke("What is contextual retrieval?"))

Notice: the chain reads top-to-bottom like a sentence. Senior LangChain code looks like this — short, declarative, composable.

5. When you outgrow LCEL

LCEL is great for:

  • Linear pipelines (RAG, summarisation, classification).
  • Light branching (RunnableBranch).
  • Parallel fan-out (RunnableParallel).

It is awkward for:

  • Loops (e.g., reflexion, self-correction).
  • Persisting state across user turns.
  • Human-in-the-loop pause/resume.
  • Complex multi-agent topologies.

That is when you switch to LangGraph (next lesson). Rule of thumb: if your "chain" needs an if-then-loop, draw it as a graph, not a chain.

6. Output parsers (Pydantic over strings)

LCEL has multiple parsers:

  • StrOutputParser — plain text.
  • JsonOutputParser — text → dict.
  • PydanticOutputParser — text → typed model. (Less needed now; modern LLMs support structured outputs natively.)
  • PydanticToolsParser — for tool-using chains.

Modern pattern: use with_structured_output on ChatOpenAI / ChatAnthropic:

python
from pydantic import BaseModel
class Sentiment(BaseModel):
    label: str
    confidence: float

parser_llm = ChatOpenAI(model="gpt-4.1-mini").with_structured_output(Sentiment)
chain = ChatPromptTemplate.from_template("Classify: {x}") | parser_llm
chain.invoke({"x": "I love it"})  # Sentiment(label='positive', confidence=0.97)

That is the cleanest way to get typed outputs out of any chain.

Hands-on lab (3 hours)

Build rag_lcel.py:

  1. Index 50 markdown notes with Chroma + OpenAI embeddings.
  2. Build the RAG LCEL chain shown above.
  3. Add a RunnableBranch so questions starting with code: go through a code-specific prompt with code-only documents (filter retriever by metadata={"type":"code"}).
  4. Add RunnableParallel to also produce a 1-line summary alongside the answer:
    python
    parallel = RunnableParallel(answer=rag_chain, summary=summary_chain)
  5. Stream output to the terminal.
  6. Build a pytest test using LCEL's .batch() to evaluate 10 questions in parallel.

Acceptance criteria:

  • One file, < 200 lines, all LCEL.
  • Streaming works in the terminal.
  • Tests run in < 3 seconds with batching.
  • README explains why LCEL not a manual loop.

Common pitfalls

  1. Forgetting StrOutputParser — chains return AIMessage objects unless you parse.
  2. Mixing async and sync — chain.invoke cannot be await-ed. Use .ainvoke.
  3. Serialising raw Documents — call .page_content to get text.
  4. Passing the wrong shape into a chain — use RunnablePassthrough.assign to shape inputs.
  5. Hand-writing what with_structured_output already does.

Self-check

  1. What does the | operator return?
  2. Why is batch() faster than a Python for loop calling .invoke?
  3. When do you reach for LangGraph instead of LCEL?
  4. What is the difference between RunnableLambda and RunnablePassthrough?
  5. Why is with_structured_output better than PydanticOutputParser in 2026?

References

Sign in to save your progress and earn badges.