Flujo de trabajo del agente LangChain

En la sección anterior conocimoscreate_agent()los parámetros de.

Esta sección profundiza en el interior del agente para comprender cómo funciona: cómo colaboran el modelo y las herramientas, y cuándo se detiene el agente.

Bucle de ejecución del agente

El núcleo del agente es un bucle simple:Llamar al modelo → verificar si se necesitan herramientas → ejecutar herramientas → repetirHasta que el modelo deja de solicitar llamadas a herramientas, el agente se detiene y devuelve el resultado final.

A continuación, entendemos este proceso rastreando cada paso del agente.

Ejemplo

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:
    """Consultar el clima de la ciudad especificada.

    Args:
city: nombre de la ciudad
    """
    weather_data = {
        "Hangzhou": "Soleado, 25°C",
        "Pekín": "Nublado, 18°C",
    }
    return weather_data.get(city, f"No se encontraron datos del clima para {city}")


@tool
def get_time(city: str) -> str:
    """Consultar la hora actual de la ciudad especificada.

    Args:
city: nombre de la ciudad
    """
    time_data = {
        "Hangzhou": "14:30",
        "Pekín": "14:30",
        "Nueva York": "02:30",
    }
    return time_data.get(city, fNo se encontraron datos de hora para {city})


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[get_weather, get_time],
    system_prompt=Eres un asistente servicial.,
)

# Usar stream_mode="updates" para ver cada paso
print(=== Seguimiento del proceso de ejecución del agente ===\n")
step = 0
for chunk in agent.stream(
    {"messages": [HumanMessage(content=¿Cómo está el clima en Hangzhou ahora? ¿Qué hora es?)]},
    stream_mode="updates",
):
    step += 1
    print(f--- Paso {step} ---)
    for node_name, update in chunk.items():
        print(fNodo: {node_name})
        if "messages" in update:
            for msg in update["messages"]:
                if hasattr(msg, 'tool_calls') and msg.tool_calls:
                    # Los mensajes de AI contienen llamadas a herramientas
                    for tc in msg.tool_calls:
                        print(f→ Solicitar llamada a herramienta: {tc['name']}({tc['args']}))
                elif msg.type == "tool":
                    print(f→ Resultado de la herramienta [{msg.name}]: {msg.content})
                elif msg.type == "ai" and msg.content:
                    print(f→ Respuesta de AI: {msg.content[:100]})

Resultado de la ejecución:

=== Agent 执行过程追踪 ===

--- 步骤 1 ---
Nodo: model
  → 请求调用工具: get_weather({'city': '杭州'})
  → 请求调用工具: get_time({'city': '杭州'})

--- 步骤 2 ---
Nodo: tools
  → 工具结果 [get_weather]: 晴，25°C
  → 工具结果 [get_time]: 14:30

--- 步骤 3 ---
Nodo: model
  → AI 回复: 杭州现在天气晴朗，气温25°C，当前时间是14:30.

De este rastreo se puede ver que el agente ejecutó3 pasos：

  1. Paso 1 (nodo model): el modelo recibe la pregunta, determina que necesita llamar a las herramientas get_weather y get_time, y devuelve dos tool_call
  2. Paso 2 (nodo tools): ejecuta las dos herramientas y obtiene los resultados del clima y la hora
  3. Paso 3 (nodo model): el modelo recibe los resultados de las herramientas, determina que la información es suficiente y genera la respuesta final

Explicación detallada de stream_mode

stream() admite varios modos de stream_mode, cada uno proporciona información con diferente granularidad:

ModoContenido devueltoEscenario de aplicación
updatesActualización de estado después de la ejecución de cada nodoRastrear los pasos de ejecución del agente, mostrar resultados intermedios
valuesEstado completo después de la ejecución de cada nodoNecesita ver el historial de mensajes completo en cada paso
messagesFlujo de mensajes token por tokenEl frontend muestra en streaming el efecto de escritura de la IA
customEventos personalizadosEl middleware envía eventos personalizados mediante stream_writer

stream_mode="values": ver los cambios completos de estado

Ejemplo

print("=== stream_mode='values' ===\n")
for i, chunk in enumerate(agent.stream(
    {"messages": [HumanMessage(content=¿Cómo está el clima en Hangzhou?)]},
    stream_mode="values",
)):
    messages = chunk.get("messages", [])
    print(fEstado {i}: {len(messages)} mensajes)
    for msg in messages:
        print(f"  [{msg.type}] {str(msg.content)[:80]}")
    if i >= 3:  # Solo ver los primeros estados
        break

Resultado de la ejecución:

=== stream_mode='values' ===

Estado 0: 1 条消息
  [human] 杭州天气怎么样？

Estado 1: 2 条消息
  [human] 杭州天气怎么样？
  [ai]

Estado 2: 3 条消息
  [human] 杭州天气怎么样？
  [ai]
  [tool] 晴，25°C

Estado 3: 4 条消息
  [human] 杭州天气怎么样？
  [ai]
  [tool] 晴，25°C
  [ai] 杭州今天天气晴朗，气温25°C，适合出门活动。

stream_mode="messages": salida en streaming token por token

Ejemplo

print(=== stream_mode='messages' (efecto de escritura en streaming) ===\n")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content=Presenta EJEMPLO en una frase.)]},
    stream_mode="messages",
):
    # msg_chunk es un AIMessageChunk, cada uno contiene solo un Token
    if hasattr(msg_chunk, 'content') and msg_chunk.content:
        print(msg_chunk.content, end="", flush=True)
print()

Condiciones de salida del agente

¿Cuándo se detiene el agente? Principalmente en los siguientes casos:

Condición de salidaDescripciónEjemplo
Sin llamadas a herramientasEl AIMessage devuelto por el modelo tiene tool_calls vacíoEl modelo considera que la tarea está completa y responde directamente
return_direct=TrueLa herramienta está marcada como de retorno directo, finaliza inmediatamente después de ejecutarseHerramientas de consulta, el resultado es la respuesta final
structured_responseEl modelo ha producido una salida estructuradaLa salida estructurada especificada por response_format está completada
jump_to="end"El middleware finaliza activamente mediante el control de estadoSe detecta un problema de exceso de permisos y se termina antes de tiempo

Ejemplo

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


# Caso 1: el modelo responde directamente (no requiere herramientas)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[],  # Sin herramientas
)

result = agent.invoke({
    "messages": [HumanMessage(content="Presenta el Tutorial para novatos en una frase")]
})
# El modelo responde directamente, el bucle se ejecuta solo una vez
print(f"Escenario sin herramientas, número de mensajes: {len(result['messages'])}")
print(f"Respuesta: {result['messages'][-1].content[:80]}...")


# Caso 2: se requiere llamada a herramientas (bucle de múltiples rondas)
@tool
def search_course(keyword: str) -> str:
    """Buscar cursos del Tutorial para novatos"""
    return f"Se encontraron 3 cursos relacionados con {keyword}"


agent_with_tools = create_agent(
    model=model,
    tools=[search_course],
)

result = agent_with_tools.invoke({
    "messages": [HumanMessage(content="Buscar cursos de Python")]
})
print(f"\n"Escenario con herramientas, número de mensajes: {len(result['messages'])}")
# Normalmente habrá: 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]}")

Resultado de la ejecución:

无工具场景，消息数: 2
Respuesta: Ejemplo es una plataforma de aprendizaje de programación para principiantes, con tutoriales y ejemplos completos...

有工具场景，消息数: 4
  [human] 搜索 Python 课程
  [ai] tool_calls: ['search_course']
  [tool] 找到 python 相关课程 3 门
  [ai] Buscando en Ejemplo "cursos de Python" se encontraron 3 cursos relacionados...

Comparación entre invoke y stream

MétodoMomento de retornoEscenario de aplicaciónExperiencia de usuario
invoke()Retorna de una sola vez después de completar todoScripts, interfaces de API, procesamiento por lotesVer el resultado completo después de esperar
stream()Retorna estados intermedios paso a pasoInterfaz de chat, procesos que necesitan mostrarseVer el progreso en tiempo real
ainvoke()Retorna asincrónicamente después de completar todoServicios web, marcos asincrónicosNo bloquea el bucle de eventos
astream()Retorna asincrónicamente paso a pasoPush WebSocket, SSEPush en tiempo real desde el servidor

Pasar el ID de hilo en with_config

Si usas un checkpointer (se detallará en capítulos posteriores), debes pasar thread_id a través de config para gestionar los hilos de conversación:

Ejemplo

# config se usa para pasar la configuración en tiempo de ejecución
# thread_id se usa para diferenciar distintos hilos de conversación
config = {"configurable": {"thread_id": "conversation-001"}}

# método invoke
result = agent.invoke(
    {"messages": [HumanMessage(content="Hola")]},
    config=config,
)

# el modo stream también admite config
for chunk in agent.stream(
    {"messages": [HumanMessage(content="Hola")]},
    config=config,
    stream_mode="updates",
):
    print(chunk)
Otras extensiones