Création de flux de travail agentiques Human-In-The-Loop | Vers la science des données

comme GPT-5.4 d’OpenAI et Opus 4.6 d’Anthropic ont démontré des capacités exceptionnelles dans l’exécution de tâches agentiques de longue durée.

En conséquence, nous constatons une utilisation accrue de Agents LLM dans les environnements individuels et professionnels pour accomplir des tâches complexes, telles que l’exécution d’analyses financières, la création d’applications et la réalisation de recherches approfondies.

Ces agents, qu’ils fassent partie d’une configuration hautement autonome ou d’un flux de travail prédéfini, peuvent exécuter des tâches en plusieurs étapes à l’aide d’outils pour atteindre des objectifs avec surveillance humaine minimale.

Toutefois, « minimal » ne signifie pas qu’il n’y a aucune surveillance humaine.

Au contraire, l’examen humain reste important en raison de la nature probabiliste inhérente aux LLM et du risque d’erreurs.

Ces erreurs peuvent se propager et s’aggraver tout au long du flux de travail, en particulier lorsque nous enchaînons de nombreux composants agentiques.

Vous auriez remarqué les progrès impressionnants réalisés par les agents dans le domaine du codage. La raison en est que le code est relativement facile à vérifier (c’est-à-dire qu’il s’exécute ou échoue, et les commentaires sont visibles immédiatement).

Mais dans des domaines tels que la création de contenu, la recherche ou la prise de décision, l’exactitude est souvent subjective et plus difficile à évaluer automatiquement.

C’est pourquoi humain dans la boucle (HITL) la conception reste essentielle.

Dans cet article, nous expliquerons comment utiliser LangGraph pour mettre en place un humain dans la boucle workflow agentique pour la génération et la publication de contenu sur Ciel bleu.

Contenu

(1) Introduction à LangGraph
(2) Exemple de flux de travail
(3) Concepts clés
(4) Procédure pas à pas du code
(5) Meilleures pratiques en matière d’interruptions

Vous trouverez l’accompagnement Dépôt GitHub ici.

---

(1) Introduction à LangGraph

LangGraph (une partie du LangChaîne écosystème) est un cadre d’orchestration d’agents de bas niveau et un environnement d’exécution permettant de créer des flux de travail agentiques.

Il s’agit de mon framework de prédilection étant donné son degré élevé de contrôle et de personnalisation, ce qui est vital pour les solutions de production.

Alors que LangChain propose un objet middleware (HumanInTheLoopMiddleware) pour démarrer facilement la surveillance humaine dans les appels d’agents, cela se fait à un niveau d’abstraction élevé qui masque les mécanismes sous-jacents.

LangGraph, en revanche, ne fait pas abstraction des invites ou de l’architecture, nous donnant ainsi le degré de contrôle plus fin dont nous avons besoin. Il nous permet explicitement de définir :

• Comment les données circulent entre les étapes
• Où se produisent les décisions et les exécutions de code
• Là où une intervention humaine est requise

Nous utiliserons donc LangGraph démontrer le concept HITL dans un flux de travail agent.

Il est également utile de distinguer flux de travail agents et agents d’IA autonomes.

Flux de travail agents ont des chemins prédéterminés et sont conçus pour s’exécuter dans un ordre défini, avec des LLM et/ou des agents intégrés dans un ou plusieurs composants. D’autre part, Agents IA planifier, exécuter et itérer de manière autonome vers un objectif.

Dans cet article, nous nous concentrons sur flux de travail agentsdans lequel nous insérons délibérément des points de contrôle humains dans un flux prédéfini.

---

(2) Exemple de flux de travail

Pour notre exemple, nous allons créer un workflow de génération de contenu sur les réseaux sociaux comme suit :

1. L’utilisateur saisit un sujet d’intérêt (par exemple, « dernières nouvelles sur Anthropic»).
2. Le nœud de recherche Web utilise le Tavily outil pour rechercher en ligne des articles correspondant au top.
3. Le premier résultat de recherche est sélectionné et introduit dans un LLM dans le nœud de création de contenu pour générer une publication sur les réseaux sociaux.
4. Dans le noeud de révisionil existe deux points de contrôle d’examen humain :
   (je) Présenter le contenu généré pour que les humains puissent l’approuver, le rejeter ou le modifier ;
   (ii) Après approbation, le workflow déclenche le Ciel bleu API outil et demande une confirmation finale avant de la mettre en ligne.

Voici à quoi cela ressemble une fois exécuté depuis le terminal :

Et voici le post en direct sur mon profil Bluesky :

Bluesky est une plateforme sociale similaire à Twitter (X), et elle est choisie dans cette démo car son API est beaucoup plus facile d’accès et d’utilisation.

---

(3) Concepts clés

Le mécanisme de base derrière la configuration HITL dans LangGraph est le concept de interrompt.

Interruptions (en utilisant interrupt() et Command dans LangGraph) nous permettent de suspendre l’exécution du graphique à des moments précis, afficher certaines informations à l’humain, et attendre leur contribution avant de reprendre le flux de travail.

Commande est un objet polyvalent qui nous permet de mettre à jour l’état du graphique (update), spécifiez le prochain nœud à exécuter (goto), ou capturez la valeur pour reprendre l’exécution du graphique avec (resume).

Voici à quoi ressemble le flux :

(1) En arrivant au interrupt() fonction, les pauses d’exécution et la charge utile qui y est transmise sont affichées à l’utilisateur. La charge utile est passée interrupt doit généralement être au format JSON ou chaîne, par exemple :

decision = interrupt("Should we get KFC for lunch?") # String shown to user

(2) Une fois que l’utilisateur a répondu, nous transmettons les valeurs de réponse au graphique pour reprendre l’exécution. Il s’agit d’utiliser Command et son resume paramètre dans le cadre de la réinvocation du graphique :

if human_response == "yes":
    return graph.invoke(Command(resume="KFC"))
else:
    return graph.invoke(Command(resume="McDonalds"))

(3) La valeur de réponse dans resume est renvoyé dans le decision variable, que le nœud utilisera pour le reste de l’exécution du nœud et le flux graphique ultérieur :

if decision == "KFC":
    return Command(goto="kfc_order_node", update={"lunch_choice": "KFC")
else:
    return Command(goto="mcd_order_node", update={"lunch_choice": "McDonalds")

Les interruptions sont dynamiques et peuvent être placées n’importe où dans le code, contrairement aux points d’arrêt statiques, qui sont fixés avant ou après des nœuds spécifiques.

Cela dit, nous plaçons généralement des interruptions soit dans les nœuds, soit dans les outils appelés lors de l’exécution du graphe.

---

Enfin, parlons de points de contrôle. Lorsqu’un flux de travail s’arrête suite à une interruption, nous avons besoin d’un moyen de sauvegarder son état actuel pour pouvoir reprendre plus tard.

Nous avons donc besoin d’un point de contrôle pour persister l’état afin que l’état ne soit pas perdu pendant la pause d’interruption. Considérez un point de contrôle comme un instantané de l’état du graphique à un instant donné.

Pour le développement, il est acceptable de sauvegarder l’état en mémoire avec le InMemorySaver pointeur de contrôle.

Pour la production, il vaut mieux utiliser des magasins comme PostgreSQL ou Rédis. Dans cette optique, nous utiliserons le SQLite point de contrôle dans cet exemple au lieu d’un magasin en mémoire.

Pour assurer le graphique reprend exactement au point là où l’interruption s’est produite, nous devons passer et utiliser le même identifiant du fil de discussion.

Considérez un thread comme une session d’exécution unique (comme une conversation individuelle distincte) où chacun possède un identifiant unique et conserve son propre état et son propre historique.

L’ID du fil est transmis dans config à chaque invocation de graphique afin que LangGraph sache à partir de quel état reprendre après l’interruption.

Maintenant que nous avons abordé les concepts d’interruptions, Commandles points de contrôle et les threads, passons à la présentation du code.

---

Comme l’accent sera mis sur la mécanique de l’humain dans la boucle, nous ne couvrirons pas la configuration complète du code. Visitez le Dépôt GitHub pour la mise en œuvre complète.

(4) Procédure pas à pas du code

(4.1) Configuration initiale

Nous commençons par installer les dépendances requises et générer des clés API pour Bluesky, OpenAI, LangChain, LangGraph et Tavily.

# requirements.txt
langchain-openai>=1.1.9
langgraph>=1.0.8
langgraph-checkpoint-sqlite>=3.0.3
openai>=2.20.0
tavily-python>=0.7.21

# env.example
export OPENAI_API_KEY=your_openai_api_key
export TAVILY_API_KEY=your_tavily_api_key
export BLUESKY_HANDLE=yourname.bsky.social
export BLUESKY_APP_PASSWORD=your_bluesky_app_password

---

(4.2) Définir l’État

Nous avons mis en place le Statequi est l’objet de données partagé et structuré servant de mémoire centrale du graphique. Il comprend des champs qui capturent des informations clés, telles que le contenu de la publication et le statut d’approbation.

Le post_data La clé est l’endroit où le contenu de la publication généré sera stocké.

---

(4.3) Interruption au niveau du nœud

Nous avons mentionné précédemment que des interruptions peuvent se produire au niveau du nœud ou lors des appels d’outils. Voyons comment le premier fonctionne en configurant le nœud d’examen humain.

Le but du nœud de révision est de suspendre l’exécution et de présenter le brouillon de contenu à l’utilisateur pour révision.

Ici, nous voyons le interrupt() en action (lignes 8 à 13), où l’exécution du graphique s’arrête à la première section de la fonction de nœud.

Le details clé passée dans interrupt() contient le contenu généré, tandis que le action la touche déclenche une fonction de gestionnaire (handle_content_interrupt()) pour appuyer l’examen :

Le contenu généré est imprimé dans le terminal pour que l’utilisateur puisse le visualiser, et celui-ci peut l’approuver tel quel, le rejeter catégoriquement ou le modifier directement dans le terminal avant de l’approuver.

En fonction de la décision, la fonction de gestionnaire renvoie l’une des trois valeurs suivantes :

• True (approuvé),
• False (rejeté), ou
• Valeur de chaîne correspondant au contenu modifié par l’utilisateur (edited).

Cette valeur de retour est renvoyée au nœud de révision à l’aide de graph.invoke(Command=resume…)qui reprend l’exécution d’où interrupt() a été appelé (ligne 15) et détermine à quel nœud accéder ensuite : approuver, rejeter ou modifier le contenu et procéder à l’approbation.

---

(4.4) Interruption au niveau de l’outil

Les interruptions peuvent également être définies au niveau de l’appel de l’outil. Ceci est démontré lors du prochain point de contrôle de l’examen humain dans le approuver le nœud avant que le contenu ne soit publié en ligne sur Bluesky.

Au lieu de placer interrupt() à l’intérieur d’un nœud, nous le plaçons dans le publish_post outil qui crée des publications via l’API Bluesky :

Tout comme ce que nous avons vu au niveau du nœud, nous appelons une fonction de gestionnaire (handle_publish_interrupt) pour capturer la décision humaine :

La valeur de retour de cette étape de révision est soit :

• {"action": "confirm"}ou
• {"action": "cancel} ,

La dernière partie du code (c’est-à-dire à partir de la ligne 19) dans le publish_post L’outil utilise cette valeur de retour pour déterminer s’il faut procéder ou non à la publication sur Bluesky.

---

(4.5) Graphique de configuration avec point de contrôle

Ensuite, nous connectons les nœuds dans un graphique pour la compilation et introduisons un point de contrôle SQLite pour capturer des instantanés de l’état à chaque interruption.

SQLite par défaut autorise uniquement le thread qui a créé la connexion à la base de données à l’utiliser. Puisque LangGraph utilise un pool de threads pour les écritures de points de contrôle, nous devons définir check_same_thread=False pour permettre à ces threads d’accéder également à la connexion.

---

(4.6) Configurer le flux de travail complet avec Config

Une fois le graphique prêt, nous le plaçons maintenant dans un flux de travail qui lance le pipeline de génération de contenu.

Ce flux de travail comprend la configuration d’un ID de thread, qui est transmis à chaquegraph.invoke(). Cet ID est le lien qui relie les invocations entre elles, de sorte que le graphique s’arrête lors d’une interruption et reprend là où il s’était arrêté.

Vous avez peut-être remarqué le __interrupt__ saisissez le code ci-dessus. C’est simplement une clé spéciale que LangGraph ajoute au résultat chaque fois qu’un interrupt() est touché.

En d’autres termes, c’est le signal principal indiquant que l’exécution du graphique a été interrompue et attend l’intervention humaine avant de continuer.

En plaçant __interrupt__ dans le cadre d’un while boucle, cela signifie que la boucle continue de vérifier si une interruption est toujours en cours. Une fois l’interruption résolue, la clé disparaît et la boucle while se termine.

Une fois le workflow terminé, nous pouvons l’exécuter comme ceci :

run_hitl_workflow(query="latest news about Anthropic")

---

(5) Meilleures pratiques en matière d’interruptions

Bien que les interruptions soient puissantes pour activer les flux de travail HITL, elles peuvent être perturbatrices si elles ne sont pas utilisées correctement.

En tant que tel, je recommande de lire ceci Documentation LangGraph. Voici quelques règles pratiques à garder à l’esprit :

• N’encapsulez pas les appels d’interruption dans des blocs try/sauf, sinon ils ne suspendront pas correctement l’exécution.
• Gardez les appels d’interruption dans le même ordre à chaque fois et ne les sautez pas et ne les réorganisez pas.
• Transmettez uniquement les valeurs sécurisées JSON dans les interruptions et évitez les objets complexes
• Assurez-vous que tout code avant une interruption peut être exécuté plusieurs fois en toute sécurité (c’est-à-dire idempotence) ou déplacez-le après l’interruption.

Par exemple, j’ai rencontré un problème dans le nœud de recherche Web où j’ai placé une interruption juste après la recherche Tavily. L’intention était de faire une pause et de permettre aux utilisateurs de consulter les résultats de la recherche pour générer du contenu.

Mais comme les interruptions fonctionnent en réexécutant les nœuds à partir desquels elles ont été appelées, le nœud a simplement réexécuté la recherche sur le Web et transmis un ensemble de résultats de recherche différent de ceux que j’avais approuvés précédemment.

Par conséquent, les interruptions fonctionnent mieux comme une porte avant une action, mais si nous les utilisons après une étape non déterministe (comme la recherche), nous devons conserver le résultat ou risquer d’obtenir quelque chose de différent à la reprise.

---

Envelopper le tout

L’évaluation humaine peut sembler être un goulot d’étranglement dans les tâches agentiques, mais elle reste essentielle, en particulier dans les domaines où les résultats sont subjectifs ou difficiles à vérifier.

LangGraph facilite la création de flux de travail HITL avec des interruptions et des points de contrôle.

Le défi consiste donc à décider où placer ces points de décision humains afin de trouver un bon équilibre entre surveillance et efficacité.