Course Content
LangChain Mastery
7 sections · 109 lessons
Write a function to implement a custom LangChain agent.
What you need to know
Three levels of "custom"
| Need | Tool |
|---|---|
| Change what happens around model or tool calls | create_agent + middleware (before_model, wrap_model_call, wrap_tool_call) |
| Change the control flow itself | LangGraph StateGraph |
| Understand or teach the loop | A plain Python loop |
Pick the lowest level that works. Middleware keeps you on the maintained runtime; a custom graph means you own more code.
The loop by hand
1from langchain.messages import HumanMessage, ToolMessage23def run_agent(model, tools, question: str, max_steps: int = 5) -> str:4 bound = model.bind_tools(tools)5 by_name = {t.name: t for t in tools}6 messages = [HumanMessage(question)]7 for _ in range(max_steps):8 ai = bound.invoke(messages)9 messages.append(ai)10 if not ai.tool_calls: # no tool call = final answer11 return ai.content12 for call in ai.tool_calls:13 result = by_name[call["name"]].invoke(call["args"])14 messages.append(ToolMessage(str(result), tool_call_id=call["id"]))15 return "Stopped: step limit reached."Every ToolMessage must carry the tool_call_id of the call it answers; providers reject a history where a tool call has no matching result. This loop has no persistence, streaming, error handling or parallel execution, which is why it is for learning, not production.
The same agent as a LangGraph graph
1from langgraph.graph import StateGraph, MessagesState, START, END2from langgraph.prebuilt import ToolNode, tools_condition34def build_custom_agent(model, tools, checkpointer=None):5 bound = model.bind_tools(tools)67 def call_model(state: MessagesState):8 return {"messages": [bound.invoke(state["messages"])]}910 g = StateGraph(MessagesState)11 g.add_node("model", call_model)12 g.add_node("tools", ToolNode(tools))13 g.add_edge(START, "model")14 g.add_conditional_edges("model", tools_condition) # "tools" or END15 g.add_edge("tools", "model")16 return g.compile(checkpointer=checkpointer)MessagesState is a state with one messages list whose reducer appends new messages. ToolNode runs all tool calls from the last message, in parallel, and returns ToolMessages. tools_condition routes to "tools" if there are tool calls, otherwise to END. Now you can add your own nodes: for example a check_answer node between the model and END that verifies every price in the answer came from a tool result.
Customising with middleware instead
1from langchain.agents.middleware import dynamic_prompt, ModelRequest23@dynamic_prompt4def store_prompt(request: ModelRequest) -> str:5 return f"You are the store assistant. Store hours today: {hours()}."Pass it as create_agent(..., middleware=[store_prompt]). The system prompt is rebuilt on every model call without adding anything to the saved message history. This is how most teams "customise" an agent in LangChain 1.x.
A real-life example
An electronics store's product Q&A assistant sometimes quoted prices from memory instead of the get_price tool, and a few answers showed last month's prices.
A system-prompt rule reduced this but did not stop it. The team rebuilt the agent as a StateGraph with one extra node, verify_prices, that runs after the model's final answer. It extracts every rupee amount with a regex and checks it appears in a get_price tool result in the state. If not, it adds a message — "Price Rs 54,999 was not from a tool; call get_price" — and routes back to the model. Wrong-price answers in the weekly audit went from 11 in 2,000 to zero, at the cost of one extra model call on about 3% of conversations.
Follow-up questions to expect
- "When would you not use
create_agent?" — When the flow is not a simple loop: a fixed plan-then-execute, a mandatory validation step, or several agents with shared state. - "What does
tools_conditiondo?" — It checks the last AI message for tool calls and returns"tools"orEND. - "What is a reducer?" — The function that merges a node's update into state; for
messagesit appends (and replaces by ID) instead of overwriting.