LangChain Agent Workflow
In the previous section, we learned aboutcreate_agent()the parameters of.
This section dives deep into the Agent's internals to understand how it works—how the model and tools collaborate, and when the Agent stops.
Agent Execution Loop
The core of an Agent is a simple loop:Call model → Check whether a tool is needed → Execute tool → Repeat. Until the model no longer requests tool calls, the Agent stops and returns the final result.
Below, we trace each step of the Agent to understand this process.
Example
from dotenv import load_dotenv
load_dotenv()
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def get_weather(city: str) -> str:
"""Query the weather for a specified city.
Args:
city: City name
"""
weather_data = {
"Hangzhou": "Sunny, 25°C",
"Beijing": "Cloudy, 18°C",
}
return weather_data.get(city, f"Weather data for {city} not found")
@tool
def get_time(city: str) -> str:
"""Query the current time for a specified city.
Args:
city: City name
"""
time_data = {
"Hangzhou": "14:30",
"Beijing": "14:30",
"New York": "02:30",
}
return time_data.get(city, f"Time data for {city} not found")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="You are a helpful assistant.",
)
# Using stream_mode="updates" lets you see every step
print("=== Agent Execution Trace ===\n")
step = 0
for chunk in agent.stream(
{"messages": [HumanMessage(content="What's the weather in Hangzhou now? What time is it?")]},
stream_mode="updates",
):
step += 1
print(f"--- Step {step} ---")
for node_name, update in chunk.items():
print(f"Node: {node_name}")
if "messages" in update:
for msg in update["messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# AI message contains tool calls
for tc in msg.tool_calls:
print(f" → Requesting tool call: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → Tool result [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → AI reply: {msg.content[:100]}")
load_dotenv()
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@tool
def get_weather(city: str) -> str:
"""Query the weather for a specified city.
Args:
city: City name
"""
weather_data = {
"Hangzhou": "Sunny, 25°C",
"Beijing": "Cloudy, 18°C",
}
return weather_data.get(city, f"Weather data for {city} not found")
@tool
def get_time(city: str) -> str:
"""Query the current time for a specified city.
Args:
city: City name
"""
time_data = {
"Hangzhou": "14:30",
"Beijing": "14:30",
"New York": "02:30",
}
return time_data.get(city, f"Time data for {city} not found")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="You are a helpful assistant.",
)
# Using stream_mode="updates" lets you see every step
print("=== Agent Execution Trace ===\n")
step = 0
for chunk in agent.stream(
{"messages": [HumanMessage(content="What's the weather in Hangzhou now? What time is it?")]},
stream_mode="updates",
):
step += 1
print(f"--- Step {step} ---")
for node_name, update in chunk.items():
print(f"Node: {node_name}")
if "messages" in update:
for msg in update["messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# AI message contains tool calls
for tc in msg.tool_calls:
print(f" → Requesting tool call: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → Tool result [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → AI reply: {msg.content[:100]}")
Running result:
=== Agent 执行过程追踪 ===
--- 步骤 1 ---
Node: model
→ 请求调用工具: get_weather({'city': '杭州'})
→ 请求调用工具: get_time({'city': '杭州'})
--- 步骤 2 ---
Node: tools
→ 工具结果 [get_weather]: 晴，25°C
→ 工具结果 [get_time]: 14:30
--- 步骤 3 ---
Node: model
→ AI 回复: 杭州现在天气晴朗，气温25°C，当前时间是14:30。
From this trace, you can see that the Agent executed3 steps：
- Step 1 (model node): The model receives the question, determines it needs to call the two tools get_weather and get_time, and returns two tool_calls
- Step 2 (tools node): Executes the two tools to obtain the weather and time results
- Step 3 (model node): The model receives the tool results, determines the information is sufficient, and generates the final reply
stream_mode Detailed Explanation
stream() supports multiple stream_modes, each providing information at a different granularity:
| Mode | Returned Content | Applicable Scenarios |
|---|---|---|
| updates | State updates after each node executes | Track Agent execution steps, display intermediate results |
| values | Complete state after each node executes | Need to see the complete message history at every step |
| messages | Token-by-token message stream | Frontend streaming display of AI typing effect |
| custom | Custom events | Middleware sends custom events via stream_writer |
stream_mode="values" — View Complete State Changes
Example
print("=== stream_mode='values' ===\n")
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="How's the weather in Hangzhou?")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"State {i}: {len(messages)} messages")
for msg in messages:
print(f" [{msg.type}] {str(msg.content)[:80]}")
if i >= 3: # Only look at the first few states
break
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="How's the weather in Hangzhou?")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"State {i}: {len(messages)} messages")
for msg in messages:
print(f" [{msg.type}] {str(msg.content)[:80]}")
if i >= 3: # Only look at the first few states
break
Running result:
=== stream_mode='values' === State 0: 1 条消息 [human] 杭州天气怎么样？ State 1: 2 条消息 [human] 杭州天气怎么样？ [ai] State 2: 3 条消息 [human] 杭州天气怎么样？ [ai] [tool] 晴，25°C State 3: 4 条消息 [human] 杭州天气怎么样？ [ai] [tool] 晴，25°C [ai] 杭州今天天气晴朗，气温25°C，适合出门活动。
stream_mode="messages" — Token-by-Token Streaming Output
Example
print("=== stream_mode='messages' (streaming typing effect) ===\n")
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="Introduce Python in one sentence")]},
stream_mode="messages",
):
# msg_chunk is an AIMessageChunk, each containing only one Token
if hasattr(msg_chunk, 'content') and msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print()
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="Introduce Python in one sentence")]},
stream_mode="messages",
):
# msg_chunk is an AIMessageChunk, each containing only one Token
if hasattr(msg_chunk, 'content') and msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print()
Agent Exit Conditions
When does the Agent stop? Mainly in the following cases:
| Exit Condition | Description | Example |
|---|---|---|
| No tool calls | tool_calls is empty in the AIMessage returned by the model | The model considers the task complete and replies directly |
| return_direct=True | Tool marked to return directly, ends immediately after execution | Query-type tool, the result is the final answer |
| structured_response | The model produced structured output | Structured output specified by response_format is complete |
| jump_to="end" | Middleware actively ends via state control | Detected permission violation, terminated early |
Example
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Case 1: Model replies directly (no tool needed)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # No tools
)
result = agent.invoke({
"messages": [HumanMessage(content="Introduce Python in one sentence")]
})
# The model replies directly, the loop executes only once
print(f"No-tool scenario, message count: {len(result['messages'])}")
print(f"Reply: {result['messages'][-1].content[:80]}...")
# Case 2: Tool call required (multi-round loop)
@tool
def search_course(keyword: str) -> str:
"""Search Example courses"""
return f"Found 3 courses related to {keyword}"
agent_with_tools = create_agent(
model=model,
tools=[search_course],
)
result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Search Python courses")]
})
print(f"\nTool scenario, message count: {len(result['messages'])}")
# Usually there will be: human, ai(tool_call), tool(result), ai(final)
for msg in result["messages"]:
print(f" [{msg.type}]", end="")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f" tool_calls: {[tc['name'] for tc in msg.tool_calls]}")
else:
print(f" {str(msg.content)[:50]}")
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Case 1: Model replies directly (no tool needed)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # No tools
)
result = agent.invoke({
"messages": [HumanMessage(content="Introduce Python in one sentence")]
})
# The model replies directly, the loop executes only once
print(f"No-tool scenario, message count: {len(result['messages'])}")
print(f"Reply: {result['messages'][-1].content[:80]}...")
# Case 2: Tool call required (multi-round loop)
@tool
def search_course(keyword: str) -> str:
"""Search Example courses"""
return f"Found 3 courses related to {keyword}"
agent_with_tools = create_agent(
model=model,
tools=[search_course],
)
result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Search Python courses")]
})
print(f"\nTool scenario, message count: {len(result['messages'])}")
# Usually there will be: human, ai(tool_call), tool(result), ai(final)
for msg in result["messages"]:
print(f" [{msg.type}]", end="")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f" tool_calls: {[tc['name'] for tc in msg.tool_calls]}")
else:
print(f" {str(msg.content)[:50]}")
Running result:
无工具场景，消息数: 2 Reply: Example is a programming learning platform for beginners, offering rich tutorials and examples... 有工具场景，消息数: 4 [human] 搜索 Python 课程 [ai] tool_calls: ['search_course'] [tool] 找到 python 相关课程 3 门 [ai] Searching Example for "Python courses" found 3 related courses...
invoke vs stream Comparison
| Method | Return Timing | Applicable Scenarios | User Experience |
|---|---|---|---|
| invoke() | Returns all at once after completion | Scripts, API interfaces, batch processing | See the complete result after waiting |
| stream() | Returns intermediate states step by step | Chat interfaces, when the process needs to be displayed | See progress in real time |
| ainvoke() | Returns after async completion | Web services, async frameworks | Does not block the event loop |
| astream() | Async step-by-step return | WebSocket, SSE push | Server-side real-time push |
Passing Thread ID in with_config
If you use a checkpointer (detailed in later chapters), you need to pass thread_id through config to manage conversation threads:
Example
# config is used to pass runtime configuration
# thread_id is used to distinguish different conversation threads
config = {"configurable": {"thread_id": "conversation-001"}}
# invoke method
result = agent.invoke(
{"messages": [HumanMessage(content="Hello")]},
config=config,
)
# stream method also supports config
for chunk in agent.stream(
{"messages": [HumanMessage(content="Hello")]},
config=config,
stream_mode="updates",
):
print(chunk)
# thread_id is used to distinguish different conversation threads
config = {"configurable": {"thread_id": "conversation-001"}}
# invoke method
result = agent.invoke(
{"messages": [HumanMessage(content="Hello")]},
config=config,
)
# stream method also supports config
for chunk in agent.stream(
{"messages": [HumanMessage(content="Hello")]},
config=config,
stream_mode="updates",
):
print(chunk)