Langchain Agent Workflow

In the previous section, we learned about the parameters of create_agent().

\\n

This section delves into the internals of the Agent to understand how it works—how the model and tools collaborate, and when the Agent stops.

\\n

Agent Execution Loop

\\n

The core of an Agent is a simple loop: call the model → check if tools are needed → execute tools → repeat. Until the model no longer requests tool calls, the Agent stops and returns the final result.

\\n

Below, we trace each step of the Agent to understand this process.

\\n

Example

\\n

from dotenv import load_dotenv\\n\\n load_dotenv()\\n\\nfrom langchain.tools import tool\\n\\nfrom langchain.agents import create_agent\\n\\nfrom langchain.chat_models import init_chat_model\\n\\nfrom langchain.messages import HumanMessage\\n\\n@tool\\n\\ndef get_weather(city: str) ->str:\\n\\n"""Query the weather for a specified city.\\nArgs:\\n\\n city: City name\\n\\n """\\n\\n weather_data ={\\n\\n"Hangzhou": "Sunny，25°C",\\n\\n"Beijing": "Cloudy, 18°C",\\n\\n}\\n\\nreturn weather_data.get(city, f"NotFound {city} 's weather data")\\n\\n@tool\\n\\ndef get_time(city: str) ->str:\\n\\n"""Query the current time for a specified city.\\nArgs:\\n\\n city: City name\\n\\n """\\n\\n time_data ={\\n\\n"Hangzhou": "14:30",\\n\\n"Beijing": "14:30",\\n\\n"New York": "02:30",\\n\\n}\\n\\nreturn time_data.get(city, f"NotFound {city} 's time data")\\n\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\n\\n agent = create_agent(\\n\\n model=model,\\n\\n tools=[get_weather, get_time],\\n\\n system_prompt="You are a helpful assistant.",\\n\\n)\\n\\n# Using stream_mode="updates" You can see each step\\n\\nprint("=== Agent Execution process tracing ===n")\\n\\n step =0\\n\\nfor chunk in agent.stream(\\n\\n{"messages": [HumanMessage(content="HangzhouNow How's the weather?What time is it?")]},\\n\\n stream_mode="updates",\\n\\n):\\n\\n step +=1\\n\\nprint(f"--- Step {step} ---")\\n\\nfor node_name, update in chunk.items():\\n\\nprint(f"Node: {node_name}")\\n\\nif"messages"in update:\\n\\nfor msg in update:\\n\\nif hasattr(msg,'tool_calls')and msg.tool_calls:\\n\\n# AI Message contains tool calls\\n\\nfor tc in msg.tool_calls:\\n\\nprint(f" → Tool call request: {tc['name']}({tc['args']})")\\n\\nelif msg.type=="tool":\\n\\nprint(f" → Tool result [{msg.name}]: {msg.content}")\\n\\nelif msg.type=="ai"and msg.content:\\n\\nprint(f" → AI Response: {msg.content[:100]}")

\\n

Execution result:

\\n

=== Agent Execution process tracing ===--- Step 1 ---Node: model → Tool call request: get_weather({'city': 'Hangzhou'}) → Tool call request: get_time({'city': 'Hangzhou'})--- Step 2 ---Node: tools → Tool result : Sunny，25°C → Tool result : 14:30--- Step 3 ---Node: model → AI Response: HangzhouCurrent weather is Sunny, temperature 25°C，Current time is 14:30。

\\n

From this trace, we can see that the Agent executed 3 steps:

\\n

Step 1 (model node): The model receives the question, determines that get_weather and get_time tools need to be called, and returns two tool_calls
Step 2 (tools node): Execute both tools to get weather and time results
Step 3 (model node): The model receives the tool results, determines that information is sufficient, and generates the final response

\\n

stream_mode Details

\\n

stream() supports multiple stream_mode options, each providing different levels of granularity:

\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n

Mode	Return Content	Applicable Scenario
updates	State updates after each node execution	Tracing Agent execution steps, displaying intermediate results
values	Complete state after each node execution	Need to see complete message history at each step
messages	Token-by-token message stream	Frontend streaming display of AI typing effect
custom	Custom events	Middleware sending custom events via stream_writer

\\n

stream_mode="values" — Viewing Complete State Changes

\\n

Example

\\n

print("=== stream_mode='values' ===n")\\n\\nfor i,chunk in enumerate(agent.stream(\\n\\n{"messages": [HumanMessage(content="HangzhouHow's the weather?")]},\\n\\n stream_mode="values",\\n\\n)):\\n\\n messages =chunk.get("messages",[])\\n\\nprint(f"State {i}: {len(messages)} messages")\\n\\nfor msg in messages:\\n\\nprint(f" [{msg.type}] {str(msg.content)[:80]}")\\n\\nif i >=3: # Only look at the first few States\\n\\nbreak

\\n

Execution result:

\\n

=== stream_mode='values' ===State 0: 1 messages  HangzhouHow's the weather?State 1: 2 messages  HangzhouHow's the weather? State 2: 3 messages  HangzhouHow's the weather?   Sunny，25°C State 3: 4 messages  HangzhouHow's the weather?   Sunny，25°C  HangzhouToday's weather is Sunny, temperature 25°C，Suitable for outdoor activities.

\\n

stream_mode="messages" — Token-by-Token Streaming Output

\\n

Example

\\n

print("=== stream_mode='messages'（streaming typing effect)===n")\\n\\nfor msg_chunk, metadata in agent.stream(\\n\\n{"messages": [HumanMessage(content="Introduce TUTORIAL in one sentence")]},\\n\\n stream_mode="messages",\\n\\n):\\n\\n# msg_chunk is AIMessageChunk，Each contains only one Token\\n\\nif hasattr(msg_chunk,'content')and msg_chunk.content:\\n\\nprint(msg_chunk.content, end="", flush=True)\\n\\nprint()

\\n

Agent Exit Conditions

\\n

When does the Agent stop? There are mainly the following situations:

\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n

Exit Condition	Description	Example
No tool calls	The AIMessage returned by the model has empty tool_calls	Model considers task complete, replies directly
return_direct=True	Tool marked for direct return, ends immediately after execution	Query tools where result is the final answer
structured_response	Model produces structured output	Structured output specified by response_format is complete
jump_to="end"	Middleware actively ends through state control	Detecting out-of-bounds questions, terminating early

\\n

Example

\\n

from langchain.tools import tool\\n\\nfrom langchain.agents import create_agent\\n\\nfrom langchain.chat_models import init_chat_model\\n\\nfrom langchain.messages import HumanMessage\\n\\n# Case 1: Model responds directly (no tools needed)\\n\\n model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\n\\n agent = create_agent(\\n\\n model=model,\\n\\n tools=[],# No tool\\n\\n)\\n\\nresult = agent.invoke({\\n\\n"messages": [HumanMessage(content="Introduce in one sentence")]\\n\\n})\\n\\n# Model responds directly, loop executes only once\\n\\nprint(f"No-tool scenario, message count: {len(result['messages'])}")\\n\\nprint(f"Response: {result['messages'].content[:80]}...")\\n\\n# Case 2: Tool call required (multi-turn loop)\\n\\n@tool\\n\\ndef search_course(keyword: str) ->str:\\n\\n"""Search Course"""\\n\\nreturn f"Found {keyword} 3 related courses"\\n\\nagent_with_tools = create_agent(\\n\\n model=model,\\n\\n tools=,\\n\\n)\\n\\nresult = agent_with_tools.invoke({\\n\\n"messages": [HumanMessage(content="Search Python courses")]\\n\\n})\\n\\nprint(f"n Tool scenario, message count: {len(result['messages'])}")\\n\\n# Usually includes: human, ai(tool_call), tool(result), ai(final)\\n\\nfor msg in result:\\n\\nprint(f" [{msg.type}]", end="")\\n\\nif hasattr(msg,'tool_calls')and msg.tool_calls:\\n\\nprint(f" tool_calls: {[tc['name'] for tc in msg.tool_calls]}")\\n\\nelse:\\n\\nprint(f" {str(msg.content)[:50]}")

\\n

Execution result:

\\n

No-tool scenario, message count: 2Response: isA programming learning platform for beginners, offering rich tutorials and examples...Tool scenario, message count: 4  Search Python courses  tool_calls: ['search_course']  Found python 3 related courses  Search in "Python Course" Found 3 related courses...

\\n

invoke vs stream Comparison

\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n\\n

Method	Return Timing	Applicable Scenario	User Experience
invoke()	Returns all at once after completion	Scripts, API interfaces, batch processing	See complete result after waiting
stream()	Returns intermediate states progressively	Chat interfaces, need to show process	See progress in real-time
ainvoke()	Returns asynchronously after completion	Web services, async frameworks	Does not block event loop
astream()	Returns asynchronously and progressively	WebSocket, SSE push	Server real-time push

\\n

Passing Thread ID in with_config

\\n

If you use a checkpointer (detailed in subsequent chapters), you need to pass thread_id through config to manage conversation threads:

\\n

Example

\\n

# config Used to pass runtime configuration\\n\\n# thread_id Used to distinguish different conversation threads\\n\\n config ={"configurable": {"thread_id": "conversation-001"}}\\n\\n# invoke Method\\n\\n result = agent.invoke(\\n\\n{"messages": [HumanMessage(content="Hello")]},\\n\\n config=config,\\n\\n)\\n\\n# stream MethodAlso supports config\\n\\nfor chunk in agent.stream(\\n\\n{"messages": [HumanMessage(content="Hello")]},\\n\\n config=config,\\n\\n stream_mode="updates",\\n\\n):\\n\\nprint(chunk)

YouTip