"Building Intelligent Agents with LangChain: Runnables, Tools, and Human-in-the-Loop"

Building a simple LLM query is a weekend project. Building a production-ready AI agent that autonomously solves tasks, uses external tools safely, and integrates seamlessly with human verification is a software engineering challenge.

In this article, we'll dive into the mechanics of the LangChain Expression Language (LCEL), writing robust custom tools, and building human-in-the-loop (HITL) workflows.

1. LCEL & Runnables: The Standard Interface

At the core of modern LangChain lies the Runnable interface. By implementing a uniform interface across components (prompts, models, parsers, and custom functions), LangChain makes it trivial to chain them together.

Every Runnable implements standard methods like:

• invoke(): Sync invocation.
• ainvoke(): Async invocation.
• stream() / astream(): Streaming outputs token-by-token.
• batch() / abatch(): Parallel batch processing.

Using the pipe (|) operator, you can build declarative chains:

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

prompt = ChatPromptTemplate.from_template("Explain {topic} in one sentence.")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# Under the hood, this compiles into a RunnableSequence
chain = prompt | model | parser

response = await chain.ainvoke({"topic": "LangChain Runnables"})

2. Sequence vs. Parallel Chains

When designing complex agents, you often need to run tasks sequentially or in parallel.

Sequential execution (RunnableSequence)

A sequence passes the output of one runnable directly as the input of the next. This is useful for multi-step reasoning:

# planning_chain -> execution_chain -> parsing_chain
agent_pipeline = planning_prompt | llm | tool_calling_parser

Parallel execution (RunnableParallel)

Sometimes you need to kick off multiple operations concurrently—for example, retrieving context from a vector database and a web search at the same time. RunnableParallel executes runnables in parallel, returning a unified dictionary:

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

map_chain = RunnableParallel(
    context=retriever,
    question=RunnablePassthrough()
)
# Both inputs are fetched concurrently and fed into the next step
rag_chain = map_chain | prompt | llm | parser

3. Designing Robust Agent Tools

An agent is only as good as the tools it can use. To make tools reliable, we must define clear input schemas and handle exceptions gracefully.

In LangChain, we define tools using the @tool decorator, providing rich docstrings that the LLM uses to understand when and how to call the tool:

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="The search query to look up in the database")

@tool("db_search", args_schema=SearchInput)
def db_search(query: str) -> str:
    """Searches the internal knowledge database for relevant documentation."""
    # Implementation here
    return "Result found..."

Rule of thumb: Always write explicit validation schemas (args_schema) and handle tool execution errors within the tool so the agent can learn from the traceback and retry rather than crashing.

4. Human-in-the-Loop (HITL)

Fully autonomous agents are powerful, but for critical actions—like committing code, sending emails, or executing financial transactions—you need a human checkpoint.

In a stateful agent system (like LangGraph), we implement this using a breakpoint. Before executing a sensitive node, the event loop pauses and saves the state.

# Conceptual LangGraph setup with checkpoints
builder = StateGraph(AgentState)

# Define nodes...
builder.add_node("agent", call_model)
builder.add_node("execute_sensitive_action", run_transaction)

# Define edges with a breakpoint before execution
graph = builder.compile(interrupt_before=["execute_sensitive_action"])

When the state hits the breakpoint, it stops. The frontend or backend service sends the state details to the user (e.g. via SSE or WebSockets), gets human approval (or feedback), and resumes the run with the approved payload.

5. Practical Takeaways

• Prefer Async (ainvoke/astream) in production to maximize throughput under heavy loads.
• Rerank and Parallelize: Use RunnableParallel to execute searches concurrently, saving precious milliseconds on token latency.
• Enforce Structure: Use structured output models (with_structured_output) to guarantee the agent returns valid schemas.

Intelligent agents aren't magic; they are structured, composable state machines. By leveraging LCEL and HITL design patterns, we build systems that are both highly capable and completely safe.