確定的ワークフロー

エージェントを使用した確定的マルチステップ・ワークフローの作成方法について学習します。

マルチステップ・ワークフローの例

この例では、Pythonのネイティブ制御フローと「Reasoning and Acting」(ReAct)エージェント・フレームワークを組み合せて、エージェント・ステップを含む確定的ワークフローを作成する方法を示します。

完全自律型のエージェント・ワークフローとは異なり、決定的なワークフローでは、特定のタスクにエージェントの能力を活用しながら、一連のステップを制御できます。

マルチステップworkflow.py

from oci.addons.adk import Agent, AgentClient
from custom_functon_tools import ResearcherToolkit, WriterToolkit

"""
This examples shows how you can build "deterministically orchestrated workflows with agentic steps".
"""

# Your (existing) vanilla python code to be integrated into this agentic workflow
def get_user_preferences():
    # Simulate result you fetched from a DB
    return {
        "email": "your@email.com",
        "style": ["casual", "humorous"],
        "topics": ["ai"]
    }

def main():

    client = AgentClient(
        auth_type="api_key",
        profile="DEFAULT",
        region="us-chicago-1"
    )

    researcher = Agent(
        client=client,
        agent_endpoint_id="ocid1.genaiagentendpoint...",
        name="Researcher",
        instructions="You are a researcher. You research trending keywords based on the user preferences.",
        tools=[ResearcherToolkit()]
    )

    writer = Agent(
        client=client,
        agent_endpoint_id="ocid1.genaiagentendpoint...",
        name="Writer",
        instructions="You are a writer. You write a blog post based on the trending keywords and the user preferences.",
        tools=[WriterToolkit()]
    )

    researcher.setup()
    writer.setup()

    # Step 1: Fetch user preferences or any pre-processing information. (non-agentic step)
    user_preferences = get_user_preferences()

    # Step 2: Research trending keywords using outputs from the previous steps as input. (agentic step)
    topics = user_preferences['topics']
    researcher_prompt = f"Research trending keywords for the following topics: {topics}"
    last_run_response = researcher.run(researcher_prompt)

    # Step 3: Write a blog post using outputs from last two steps as input. (agentic step) 
    keywords = last_run_response.output
    style = user_preferences['style']
    email = user_preferences['email']
    writer_prompt = f"Write a 5 sentences blog post and email it to {email}. Use style: {style}. Blog post should be based on: {keywords}."
    last_run_response = writer.run(writer_prompt)

    # Step 4: Do whatever you want with the last step output. Here we just print it.
    last_run_response.pretty_print()

if __name__ == "__main__":
    main()

カスタム関数ツール

この例では、研究者およびライター・エージェントにカスタム・ツールキットを使用します。次に、次のような内容の簡略化されたバージョンを示します。

custom_functon_tools.py

from typing import Dict, List
from oci.addons.adk import tool, Toolkit

@tool
def get_trending_keywords(topic: str) -> Dict[str, List[str]]:
    """Get trending keywords for a given topic"""
    # In a real implementation, this might call an API or database.
    if topic == "ai":
        return {"topic": topic, "keywords": ["generative AI", "multi-agent systems", "LLM agents"]}
    return {"topic": topic, "keywords": ["unknown"]}

@tool
def send_email(recipient: str, subject: str, body: str) -> str:
    """Send an email with the given subject and body to the recipient"""
    # In a real implementation, this would send an actual email.
    print(f"Sending email to {recipient}")
    print(f"Subject: {subject}")
    print(f"Body: {body}")
    return "Email sent successfully"

class ResearcherToolkit(Toolkit):
    """Toolkit for researching trending topics"""

    def __init__(self):
        super().__init__(name="ResearcherToolkit", description="Tools for researching trending topics")
        self.add_tool(get_trending_keywords)

class WriterToolkit(Toolkit):
    """Toolkit for writing content and sending emails"""

    def __init__(self):
        super().__init__(name="WriterToolkit", description="Tools for writing content and sending emails")
        self.add_tool(send_email)

このパターンを使用するタイミング

このパターンは、次の場合に特に役立ちます。

• 予測可能で反復可能な実行が必要です。
• ワークフローの一部は、既存のシステムまたはデータベースによって異なります。
• エージェントを起動するタイミングと方法を正確に制御する必要があります。
• ビジネス・ロジックには、エージェントの意思決定に任せてはならない特定の順序が必要です。