Dans cet exemple, nous utilisons des DBO pour construire un agent de recherche en profondeur d'IA qui recherche de manière autonome les nouvelles de pirate d'informations sur n'importe quel sujet.
Cet exemple montre comment construire des agents d'IA fiables et durables avec des DBO. L'agent commence par un sujet de recherche, recherche de manière autonome des informations connexes, prend des décisions sur le moment de poursuivre la recherche et synthétise les résultats dans un rapport complet. Étant donné que l'agent est mis en œuvre en tant que flux de travail durable DBOS, il peut se remettre automatiquement de toute défaillance et poursuivre les recherches d'où elle s'était arrêtée, garantissant qu'aucun travail n'est perdu.
Cet exemple montre également à quel point il est facile d'ajouter des DBO à une application agentique existante. L'ajout de DBO à cet agent pour le rendre fiable et observable changeant <20 lignes de code. Tout ce que vous avez à faire est d'annoter des workflows et des étapes.
Tous les code source sont disponibles sur GitHub.
Le cœur de l'agent est le principal flux de travail de recherche. Il commence par un sujet et explore de manière autonome les requêtes connexes jusqu'à ce qu'elle ait suffisamment d'informations, puis synthétise un rapport final.
@ Dbos.workflow () def agentic_research_workflow (Sujet: STR, max_iterations: int) -> dict [str, any]: "" "Le flux de travail agentique principal qui recherche de manière autonome un sujet. Cela démontre un flux de travail agentique complet à l'aide de DBOS. L'agent commence par un sujet de recherche sur les quères liés. Décisions sur le moment de continuer 4. Synthétise les résultats dans un rapport final. Current_Query, Current_Iteration) Research_History.APPEND (itération_result) all_findings.append (itération_result ["Évaluation"]) # Cas où aucun résultat n'est trouvé historique_found = itération_result ["Stories_found"] if stories_found == 0: # génération Alternative Queries (Triting se termine alternative_quey = génération_ all_findings, current_iteration) si alternative_query: current_query = alternative_query continue # évaluer s'il faut poursuivre la recherche devrait_continue = devrait_continue_step (thème, all_finding Generate_follow_ups_step (Topic "Max_iterations": max_iterations, "Research_history": Research_History, "final_report": final_report, "résumé": {"total_stories": sum (r ["stories_found"] pour r dans Research_History), "total_comements": sum ...
[Courte citation de 8% de l'article original]