Рабочий процесс агента LangChain
В предыдущем разделе мы узнали о параметрахcreate_agent().
В этом разделе мы глубоко погрузимся во внутреннее устройство агента, чтобы понять, как оно работает — как модель и инструменты взаимодействуют и когда агент останавливается.
Цикл выполнения агента
В основе агента лежит простой цикл:Вызвать модель → Проверить, нужен ли инструмент → Выполнить инструмент → Повторить. Пока модель не перестанет запрашивать вызовы инструментов, агент останавливается и возвращает окончательный результат.
Ниже мы прослеживаем каждый шаг Агента, чтобы понять этот процесс.
Пример
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:
"""Запросить погоду для указанного города.
Аргументы:
city: Название города
"""
weather_data = {
"Ханчжоу": "Солнечно, 25°C",
"Пекин": "Облачно, 18°C",
}
return weather_data.get(city, f"Данные о погоде для {city} не найдены")
@tool
def get_time(city: str) -> str:
"""Запросить текущее время для указанного города.
Аргументы:
city: Название города
"""
time_data = {
"Ханчжоу": "14:30",
"Пекин": "14:30",
"Нью-Йорк": "02:30",
}
return time_data.get(city, f"Данные о времени для {city} не найдены")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="Вы — полезный помощник.",
)
# Использование stream_mode="updates" позволяет видеть каждый шаг
print("=== Трассировка выполнения агента ==="\n")
шаг = 0
for chunk in agent.stream(
{": [Human(content="Какая сейчас погода в Ханчжоу? Который сейчас час?")]},
stream_mode="updates",
):
step += 1
print(f"--- Шаг {step} ---")
for node_name, update in chunk.items():
print(f"Узел: {node_name}")
if "messages" in update:
for msg in update[messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# AI-сообщение содержит вызовы инструментов
for tc in msg.tool_calls:
print(f" → Запрос вызова инструмента: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → Результат инструмента [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → Ответ AI: {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:
"""Запросить погоду для указанного города.
Аргументы:
city: Название города
"""
weather_data = {
"Ханчжоу": "Солнечно, 25°C",
"Пекин": "Облачно, 18°C",
}
return weather_data.get(city, f"Данные о погоде для {city} не найдены")
@tool
def get_time(city: str) -> str:
"""Запросить текущее время для указанного города.
Аргументы:
city: Название города
"""
time_data = {
"Ханчжоу": "14:30",
"Пекин": "14:30",
"Нью-Йорк": "02:30",
}
return time_data.get(city, f"Данные о времени для {city} не найдены")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="Вы — полезный помощник.",
)
# Использование stream_mode="updates" позволяет видеть каждый шаг
print("=== Трассировка выполнения агента ==="\n")
шаг = 0
for chunk in agent.stream(
{": [Human(content="Какая сейчас погода в Ханчжоу? Который сейчас час?")]},
stream_mode="updates",
):
step += 1
print(f"--- Шаг {step} ---")
for node_name, update in chunk.items():
print(f"Узел: {node_name}")
if "messages" in update:
for msg in update[messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# AI-сообщение содержит вызовы инструментов
for tc in msg.tool_calls:
print(f" → Запрос вызова инструмента: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → Результат инструмента [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → Ответ AI: {msg.content[:100]}")
Результат выполнения:
=== 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.
Из этой трассировки видно, что Агент выполнил3 шага：
- Шаг 1 (узел модели): Модель получает вопрос, определяет, что ей нужно вызвать два инструмента get_weather и get_time, и возвращает два tool_calls
- Шаг 2 (узел инструментов): Выполняет два инструмента, чтобы получить результаты погоды и времени
- Шаг 3 (узел модели): Модель получает результаты инструментов, определяет, что информации достаточно, и генерирует окончательный ответ
Подробное объяснение stream_mode
stream() поддерживает несколько stream_modes, каждый предоставляет информацию с разной степенью детализации:
| Mode | Возвращаемое содержимое | Применимые сценарии |
|---|---|---|
| updates | Обновления состояния после выполнения каждого узла | Отслеживание шагов выполнения агента, отображение промежуточных результатов |
| values | Полное состояние после выполнения каждого узла | Нужно видеть полную историю сообщений на каждом шаге |
| messages | Поток сообщений по токенам | Потоковое отображение на фронтенде эффекта набора текста ИИ |
| custom | Пользовательские события | Middleware отправляет пользовательские события через stream_writer |
stream_mode="values" — Просмотр полных изменений состояния
Пример
print("=== stream_mode='values' ===\n")
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="Какая погода в Ханчжоу?")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"Состояние {i}: {len(messages)} сообщений")
for msg in messages:
print(f" [{msg.type}] {str(msg.content)[:80]}")
if i >= 3: # Смотрим только первые несколько состояний
break
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="Какая погода в Ханчжоу?")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"Состояние {i}: {len(messages)} сообщений")
for msg in messages:
print(f" [{msg.type}] {str(msg.content)[:80]}")
if i >= 3: # Смотрим только первые несколько состояний
break
Результат выполнения:
=== 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" — Потоковый вывод токен за токеном
Пример
print("=== stream_mode='messages' (эффект потоковой печати) ==="\n")
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="Опишите Python одним предложением")]},
stream_mode="messages",
):
# msg_chunk — это AIMessageChunk, каждый содержит только один токен
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="Опишите Python одним предложением")]},
stream_mode="messages",
):
# msg_chunk — это AIMessageChunk, каждый содержит только один токен
if hasattr(msg_chunk, 'content') and msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print()
Условия выхода агента
Когда агент останавливается? В основном в следующих случаях:
| Условие выхода | Описание | Пример |
|---|---|---|
| Нет вызовов инструментов | tool_calls пуст в AIMessage, возвращённом моделью | Модель считает задачу выполненной и отвечает напрямую |
| return_direct=True | Инструмент помечен для прямого возврата, завершается сразу после выполнения | Инструмент типа запроса, результат — окончательный ответ |
| structured_response | Модель сформировала структурированный вывод | Структурированный вывод, заданный response_format, завершён |
| jump_to="end" | Middleware активно завершает работу через управление состоянием | Обнаружено нарушение разрешений, работа завершена досрочно |
Пример
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# Случай 1: модель отвечает напрямую (инструмент не нужен)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # Нет инструментов
)
result = agent.invoke({
"messages": [HumanMessage(content="Опишите Python одним предложением")]
})
# Модель отвечает напрямую, цикл выполняется только один раз
print(f"Сценарий без инструментов, количество сообщений: {len(result['messages'])}")
print(f"Ответ: {result['messages'][-1].content[:80]}...")
# Случай 2: Требуется вызов инструмента (многораундовый цикл)
@tool
def search_course(keyword: str) -> str:
"""Поиск примеров курсов"""
return f"Найдено 3 курса, связанных с {keyword}"
agent_with_tools = create_agent(
model=model,
tools=[search_course],
)
result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Поиск курсов Python")]
})
print(f"\nСценарий с инструментом, количество сообщений: {len(result['messages'])}")
# Обычно будут: 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
# Случай 1: модель отвечает напрямую (инструмент не нужен)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # Нет инструментов
)
result = agent.invoke({
"messages": [HumanMessage(content="Опишите Python одним предложением")]
})
# Модель отвечает напрямую, цикл выполняется только один раз
print(f"Сценарий без инструментов, количество сообщений: {len(result['messages'])}")
print(f"Ответ: {result['messages'][-1].content[:80]}...")
# Случай 2: Требуется вызов инструмента (многораундовый цикл)
@tool
def search_course(keyword: str) -> str:
"""Поиск примеров курсов"""
return f"Найдено 3 курса, связанных с {keyword}"
agent_with_tools = create_agent(
model=model,
tools=[search_course],
)
result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Поиск курсов Python")]
})
print(f"\nСценарий с инструментом, количество сообщений: {len(result['messages'])}")
# Обычно будут: 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]}")
Результат выполнения:
无工具场景，消息数: 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 и stream
| Метод | Момент возврата | Применимые сценарии | Пользовательский опыт |
|---|---|---|---|
| invoke() | Возвращает всё сразу после завершения | Скрипты, API-интерфейсы, пакетная обработка | Посмотреть полный результат после ожидания |
| stream() | Пошагово возвращает промежуточные состояния | Чат-интерфейсы, когда нужно отображать процесс | Отслеживать прогресс в реальном времени |
| ainvoke() | Возвращает результат после завершения асинхронной операции | Веб-сервисы, асинхронные фреймворки | Не блокирует цикл событий |
| astream() | Асинхронный пошаговый возврат | WebSocket, SSE push | Серверная отправка в реальном времени |
Передача Thread ID в with_config
Если вы используете checkpointer (подробнее в следующих главах), вам нужно передать thread_id через config для управления потоками диалога:
Пример
# config используется для передачи конфигурации времени выполнения
# thread_id используется для различения разных потоков диалога
config = {"configurable": {"thread_id": "conversation-001"}}
# метод invoke
result = agent.invoke(
{"messages": [HumanMessage(content="Hello")]},
config=config,
)
# метод stream также поддерживает config
for chunk in agent.stream(
{"messages": [HumanMessage(content="Hello")]},
config=config,
stream_mode="updates",
):
print(chunk)
# thread_id используется для различения разных потоков диалога
config = {"configurable": {"thread_id": "conversation-001"}}
# метод invoke
result = agent.invoke(
{"messages": [HumanMessage(content="Hello")]},
config=config,
)
# метод stream также поддерживает config
for chunk in agent.stream(
{"messages": [HumanMessage(content="Hello")]},
config=config,
stream_mode="updates",
):
print(chunk)