Build a Native Think Tool for Orchestrator Agents (LangGraph Example)

objective → think → act → think → act → final response

from langchain.tools import tool
 
 
@tool("think")
def think(thought: str) -> str:
    """Use this tool to plan, reason, and self-critique before or after
    taking action.
 
    Call this tool to:
    - List applicable constraints and policies before acting
    - Verify all required information is present
    - Evaluate candidate actions and their risks
    - Reflect on tool output before deciding the next step
 
    This tool does NOT fetch external data and does NOT change system state.
    It is a pure reasoning scratchpad.
 
    Args:
        thought: Your structured reasoning, plan, critique, or checklist.
    """
    # Pure scratchpad: return the thought as-is.
    # The model's reasoning is now logged in the message history
    # as a tool call + tool response, making the decision chain
    # observable and auditable.
    return thought

from langchain.tools import tool
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
 
 
# --- Domain tools (simulated for this example) ---
 
@tool("get_account_balance")
def get_account_balance(account_id: str) -> str:
    """Get the current balance for a bank account.
 
    Args:
        account_id: The account identifier (e.g., ACC-001).
    """
    balances = {"ACC-001": 5432.10, "ACC-002": 1280.50}
    balance = balances.get(account_id)
    if balance is None:
        return f"Account {account_id} not found."
    return f"Account {account_id} balance: ${balance:,.2f}"
 
 
@tool("transfer_funds")
def transfer_funds(
    from_account: str, to_account: str, amount: float
) -> str:
    """Transfer funds between two bank accounts.
    This action is irreversible once executed.
 
    Args:
        from_account: Source account ID.
        to_account: Destination account ID.
        amount: Dollar amount to transfer.
    """
    return (
        f"Transferred ${amount:,.2f} "
        f"from {from_account} to {to_account}. "
        f"Transaction confirmed."
    )
 
 
# --- Orchestrator agent ---
 
SYSTEM_PROMPT = """\
You are a financial operations agent with access to account and transfer tools.
 
IMPORTANT — Think-before-you-act policy:
1. Before ANY irreversible action (e.g., transfer_funds), you MUST call the
   `think` tool to:
   - list what you know so far,
   - verify all required parameters are present,
   - evaluate risk and constraints,
   - state your planned action and why it is safe.
2. After receiving tool output that requires interpretation, call `think`
   to reflect before responding to the user.
3. Never skip the think step for high-impact operations.
"""
 
model = ChatOpenAI(model="gpt-4o")
 
agent = create_agent(
    model,
    tools=[think, get_account_balance, transfer_funds],
    system_prompt=SYSTEM_PROMPT,
)
 
# --- Run the agent ---
 
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Transfer $500 from ACC-001 to ACC-002"}]}
)
 
# Expected agent behavior:
# 1. Call think() — plan: verify accounts, check amount, assess risk
# 2. Call get_account_balance("ACC-001") — verify sufficient funds
# 3. Call think() — reflect on balance, confirm transfer is safe
# 4. Call transfer_funds("ACC-001", "ACC-002", 500.0)
# 5. Respond to the user with confirmation

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
 
# Reuse the same @tool-decorated functions defined above
tools = [think, get_account_balance, transfer_funds]
 
model = ChatOpenAI(model="gpt-4o").bind_tools(tools)
 
 
def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
 
# Build the graph
builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))
 
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
 
graph = builder.compile()
 
# Run
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Transfer $500 from ACC-001 to ACC-002"}]}
)

from typing import Annotated
from langchain.tools import tool
from langgraph.types import Command
from langgraph.graph import MessagesState
import operator
 
 
class OrchestratorState(MessagesState):
    """Extended state that tracks thought history."""
    thoughts: Annotated[list[str], operator.add]
 
 
@tool("think")
def think_with_state(thought: str) -> Command:
    """Use this tool to plan, reason, and self-critique before or after
    taking action. This tool does not fetch external data or change
    system state.
 
    Args:
        thought: Your structured reasoning, plan, critique, or checklist.
    """
    return Command(update={"thoughts": [thought]})

Before any irreversible action, call `think` to verify constraints and plan.

When calling `think`, structure your reasoning as:
1. Goal — what am I trying to accomplish?
2. Constraints — what policies or limits apply?
3. Known information — what do I already have?
4. Missing information — what do I still need?
5. Candidate actions — what are my options?
6. Risk assessment — what could go wrong?
7. Decision — which action and why?

Before EVERY tool call (not just irreversible ones), call `think` to:
- verify the action complies with the relevant policy,
- confirm no PII is exposed in tool inputs,
- check rate limits have not been exceeded.