deep-agents-orchestration

Par langchain-ai · langchain-skills

Invoquez cette compétence lors de l'utilisation de sous-agents, de la planification de tâches ou de l'approbation humaine dans les Deep Agents. Couvre SubAgentMiddleware, TodoList pour la planification et les interruptions HITL.

npx skills add https://github.com/langchain-ai/langchain-skills --skill deep-agents-orchestration

<overview> Les Deep Agents incluent trois capacités d'orchestration :

  1. SubAgentMiddleware : Déléguer du travail via l'outil task à des agents spécialisés
  2. TodoListMiddleware : Planifier et suivre les tâches via l'outil write_todos
  3. HumanInTheLoopMiddleware : Exiger une approbation avant les opérations sensibles

Les trois sont automatiquement inclus dans create_deep_agent(). </overview>

Subagents (Délégation de tâches)

Utiliser les Subagents Quand Utiliser l'Agent Principal Quand
La tâche nécessite des outils spécialisés Les outils à usage général suffisent
Vous voulez isoler un travail complexe Opération en une seule étape
Vous avez besoin d'un contexte propre pour l'agent principal L'encombrement du contexte est acceptable

L'agent principal dispose de l'outil task -> crée un subagent nouveau -> le subagent s'exécute de manière autonome -> retourne un rapport final.

Subagent par défaut : « general-purpose » - automatiquement disponible avec les mêmes outils/configuration que l'agent principal.

Créer un subagent « researcher » personnalisé avec des outils spécialisés pour la recherche de publications académiques.

from deepagents import create_deep_agent
from langchain.tools import tool

@tool
def search_papers(query: str) -> str:
    """Search academic papers."""
    return f"Found 10 papers about {query}"

agent = create_deep_agent(
    subagents=[
        {
            "name": "researcher",
            "description": "Conduct web research and compile findings",
            "system_prompt": "Search thoroughly, return concise summary",
            "tools": [search_papers],
        }
    ]
)

# Main agent delegates: task(agent="researcher", instruction="Research AI trends")

Créer un subagent « researcher » personnalisé avec des outils spécialisés pour la recherche de publications académiques.

import { createDeepAgent } from "deepagents";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const searchPapers = tool(
  async ({ query }) => `Found 10 papers about ${query}`,
  { name: "search_papers", description: "Search papers", schema: z.object({ query: z.string() }) }
);

const agent = await createDeepAgent({
  subagents: [
    {
      name: "researcher",
      description: "Conduct web research and compile findings",
      systemPrompt: "Search thoroughly, return concise summary",
      tools: [searchPapers],
    }
  ]
});

// Main agent delegates: task(agent="researcher", instruction="Research AI trends")

Configurer un subagent avec approbation HITL pour les opérations sensibles.

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(
    subagents=[
        {
            "name": "code-deployer",
            "description": "Deploy code to production",
            "system_prompt": "You deploy code after tests pass.",
            "tools": [run_tests, deploy_to_prod],
            "interrupt_on": {"deploy_to_prod": True},  # Require approval
        }
    ],
    checkpointer=MemorySaver()  # Required for interrupts
)

Les subagents sont sans état - fournissez des instructions complètes en un seul appel.

# WRONG: Subagents don't remember previous calls
# task(agent='research', instruction='Find data')
# task(agent='research', instruction='What did you find?')  # Starts fresh!

# CORRECT: Complete instructions upfront
# task(agent='research', instruction='Find data on AI, save to /research/, return summary')

Les subagents sont sans état - fournissez des instructions complètes en un seul appel.

// WRONG: Subagents don't remember previous calls
// task research: Find data
// task research: What did you find?  // Starts fresh!

// CORRECT: Complete instructions upfront
// task research: Find data on AI, save to /research/, return summary

Les subagents personnalisés n'héritent pas des compétences de l'agent principal.

# WRONG: Custom subagent won't have main agent's skills
agent = create_deep_agent(
    skills=["/main-skills/"],
    subagents=[{"name": "helper", ...}]  # No skills inherited
)

# CORRECT: Provide skills explicitly (general-purpose subagent DOES inherit)
agent = create_deep_agent(
    skills=["/main-skills/"],
    subagents=[{"name": "helper", "skills": ["/helper-skills/"], ...}]
)

TodoList (Planification des tâches)

Utiliser TodoList Quand Ignorer TodoList Quand
Tâches multi-étapes complexes Tâches simples en une seule action
Opérations longues Opérations rapides (< 3 étapes) 
write_todos(todos: list[dict]) -> None

Chaque élément de liste de tâches a :

  • content : Description de la tâche
  • status : L'une de "pending", "in_progress", "completed"

Invoquer un agent qui crée automatiquement une liste de tâches pour une tâche multi-étapes.

from deepagents import create_deep_agent

agent = create_deep_agent()  # TodoListMiddleware included by default

result = agent.invoke({
    "messages": [{"role": "user", "content": "Create a REST API: design models, implement CRUD, add auth, write tests"}]
}, config={"configurable": {"thread_id": "session-1"}})

# Agent's planning via write_todos:
# [
#   {"content": "Design data models", "status": "in_progress"},
#   {"content": "Implement CRUD endpoints", "status": "pending"},
#   {"content": "Add authentication", "status": "pending"},
#   {"content": "Write tests", "status": "pending"}
# ]

Invoquer un agent qui crée automatiquement une liste de tâches pour une tâche multi-étapes.

import { createDeepAgent } from "deepagents";

const agent = await createDeepAgent();  // TodoListMiddleware included

const result = await agent.invoke({
  messages: [{ role: "user", content: "Create a REST API: design models, implement CRUD, add auth, write tests" }]
}, { configurable: { thread_id: "session-1" } });

Accéder à la liste de tâches à partir de l'état final de l'agent après son invocation.

result = agent.invoke({...}, config={"configurable": {"thread_id": "session-1"}})

# Access todo list from final state
todos = result.get("todos", [])
for todo in todos:
    print(f"[{todo['status']}] {todo['content']}")

L'état de la liste de tâches nécessite un thread_id pour la persistance entre les invocations.

# WRONG: Fresh state each time without thread_id
agent.invoke({"messages": [...]})

# CORRECT: Use thread_id
config = {"configurable": {"thread_id": "user-session"}}
agent.invoke({"messages": [...]}, config=config)  # Todos preserved

Human-in-the-Loop (Flux d'approbation)

Utiliser HITL Quand Ignorer HITL Quand
Opérations à enjeux élevés (écritures BD, déploiements) Opérations en lecture seule
La conformité exige une surveillance humaine Flux de travail entièrement automatisés

Configurer les outils qui nécessitent une approbation humaine avant l'exécution.

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(
    interrupt_on={
        "write_file": True,  # All decisions allowed
        "execute_sql": {"allowed_decisions": ["approve", "reject"]},
        "read_file": False,  # No interrupts
    },
    checkpointer=MemorySaver()  # REQUIRED for interrupts
)

Configurer les outils qui nécessitent une approbation humaine avant l'exécution.

import { createDeepAgent } from "deepagents";
import { MemorySaver } from "@langchain/langgraph";

const agent = await createDeepAgent({
  interruptOn: {
    write_file: true,
    execute_sql: { allowedDecisions: ["approve", "reject"] },
    read_file: false,
  },
  checkpointer: new MemorySaver()  // REQUIRED
});

Flux de travail complet : déclencher une interruption, vérifier l'état, approuver l'action et reprendre l'exécution.

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

agent = create_deep_agent(
    interrupt_on={"write_file": True},
    checkpointer=MemorySaver()
)

config = {"configurable": {"thread_id": "session-1"}}

# Step 1: Agent proposes write_file - execution pauses
result = agent.invoke({
    "messages": [{"role": "user", "content": "Write config to /prod.yaml"}]
}, config=config)

# Step 2: Check for interrupts
state = agent.get_state(config)
if state.next:
    print(f"Pending action")

# Step 3: Approve and resume
result = agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)

Flux de travail complet : déclencher une interruption, vérifier l'état, approuver l'action et reprendre l'exécution.

import { createDeepAgent } from "deepagents";
import { MemorySaver, Command } from "@langchain/langgraph";

const agent = await createDeepAgent({
  interruptOn: { write_file: true },
  checkpointer: new MemorySaver()
});

const config = { configurable: { thread_id: "session-1" } };

// Step 1: Agent proposes write_file - execution pauses
let result = await agent.invoke({
  messages: [{ role: "user", content: "Write config to /prod.yaml" }]
}, config);

// Step 2: Check for interrupts
const state = await agent.getState(config);
if (state.next) {
  console.log("Pending action");
}

// Step 3: Approve and resume
result = await agent.invoke(
  new Command({ resume: { decisions: [{ type: "approve" }] } }), config
);

Rejeter une action en attente avec commentaires, incitant l'agent à essayer une approche différente.

result = agent.invoke(
    Command(resume={"decisions": [{"type": "reject", "message": "Run tests first"}]}),
    config=config,
)

Rejeter une action en attente avec commentaires, incitant l'agent à essayer une approche différente.

const result = await agent.invoke(
  new Command({ resume: { decisions: [{ type: "reject", message: "Run tests first" }] } }),
  config,
);

Modifier les arguments de l'action proposée avant d'autoriser l'exécution.

result = agent.invoke(
    Command(resume={"decisions": [{
        "type": "edit",
        "edited_action": {
            "name": "execute_sql",
            "args": {"query": "DELETE FROM users WHERE last_login < '2020-01-01' LIMIT 100"},
        },
    }]}),
    config=config,
)

Ce que les Agents PEUVENT configurer

  • Noms des subagents, outils, modèles, system prompts
  • Quels outils nécessitent une approbation
  • Types de décisions autorisés par outil
  • Contenu et structure de la TodoList

Ce que les Agents NE PEUVENT PAS configurer

  • Noms des outils (task, write_todos)
  • Protocole HITL (structure approve/edit/reject)
  • Ignorer le besoin de checkpointer pour les interruptions
  • Rendre les subagents avec état (ils sont éphémères)

Le Checkpointer est requis lors de l'utilisation de interrupt_on pour les flux de travail HITL.

# WRONG
agent = create_deep_agent(interrupt_on={"write_file": True})

# CORRECT
agent = create_deep_agent(interrupt_on={"write_file": True}, checkpointer=MemorySaver())

Le Checkpointer est requis lors de l'utilisation de interruptOn pour les flux de travail HITL.

// WRONG
const agent = await createDeepAgent({ interruptOn: { write_file: true } });

// CORRECT
const agent = await createDeepAgent({ interruptOn: { write_file: true }, checkpointer: new MemorySaver() });

Un thread_id cohérent est requis pour reprendre les flux de travail interrompus.

# WRONG: Can't resume without thread_id
agent.invoke({"messages": [...]})

# CORRECT
config = {"configurable": {"thread_id": "session-1"}}
agent.invoke({...}, config=config)
# Resume with Command using same config
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)

Un thread_id cohérent est requis pour reprendre les flux de travail interrompus.

// WRONG: Can't resume without thread_id
await agent.invoke({ messages: [...] });

// CORRECT
const config = { configurable: { thread_id: "session-1" } };
await agent.invoke({ messages: [...] }, config);
// Resume with Command using same config
await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);

Les interruptions se produisent ENTRE les appels invoke(), pas en cours d'exécution.

result = agent.invoke({...}, config=config)       # Step 1: triggers interrupt
if "__interrupt__" in result:                      # Step 2: check for interrupt
    result = agent.invoke(                         # Step 3: resume
        Command(resume={"decisions": [{"type": "approve"}]}),
        config=config,
    )

Skills similaires

deep-agents-core
langchain-ai / langchain-skills

Créer des agents IA multi-étapes avec planification, mémoire persistante et délégation de tâches.

689
video-summarization
nvidia / skills

Résumer automatiquement des vidéos courtes ou longues via VLM NIM ou LVS.

87
eval-bootstrap
datadog-labs / agent-skills

Générer du code d'évaluation Python à partir de traces de production LLM Datadog.

110
geofeed-tuner
github / awesome-copilot

Créer et optimiser des feeds de géolocalisation IP au format CSV selon RFC 8805.

33 040
ad-model-onboard
nvidia / skills

Automatiser l'intégration de modèles HuggingFace dans AutoDeploy avec tests et rapport.

87