Aracılığıyla paylaş

Microsoft Agent Framework İş Akışları - İş Akışlarını Aracı Olarak Kullanma

Bu belge, Microsoft Agent Framework'te İş Akışlarını Aracı olarak kullanma hakkında genel bir bakış sağlar.

Genel Bakış

Bazen birden çok aracı, özel yürütücü ve karmaşık mantık ile gelişmiş bir iş akışı oluşturdunuz, ancak bunu diğer aracılar gibi kullanmak istiyorsunuz. İş akışı aracıları tam olarak bunu yapmanıza izin verir. İş akışınızı bir Agent olarak sarmalayıp, basit bir sohbet aracısı için kullandığınız tanıdık API aracılığıyla bu iş akışıyla etkileşim kurabilirsiniz.

Önemli Avantajlar

  • Birleşik Arabirim: Basit aracılarla aynı API'yi kullanarak karmaşık iş akışlarıyla etkileşim kurma
  • API Uyumluluğu: İş akışlarını Aracı arabirimini destekleyen mevcut sistemlerle tümleştirme
  • Oluşturulabilirlik: İş akışı aracılarını daha büyük aracı sistemlerinde veya diğer iş akışlarında yapı taşları olarak kullanma
  • İş Parçacığı Yönetimi: Konuşma durumu, denetim noktası oluşturma ve yeniden başlatma için aracı iş parçacıklarından yararlanma
  • Akış Desteği: İş akışı yürütülürken gerçek zamanlı güncelleştirmeleri alma

Nasıl Çalışır?

Bir iş akışını aracıya dönüştürdüğünüzde:

  1. İş akışı, başlat yürütücüsünü sohbet iletilerini kabul etmesini sağlamak için doğrulanır
  2. Konuşma durumunu ve denetim noktalarını yönetmek için bir iş parçacığı oluşturulur
  3. Giriş iletileri iş akışının başlat yürütücüsine yönlendirilir
  4. İş akışı olayları aracı yanıt güncelleştirmelerine dönüştürülür
  5. Dışarıdan gelen giriş istekleri (RequestInfoExecutor), işlev çağrıları olarak sunulur.

Gereksinimler

Bir iş akışını aracı olarak kullanmak için iş akışının başlangıç yürütücüsunun giriş olarak işleyebilmesi IEnumerable<ChatMessage> gerekir. Bu, ChatClientAgent veya diğer aracı tabanlı yürütücüler kullanıldığında otomatik olarak karşılanır.

İş Akışı Aracısı Oluşturma

Uyumlu iş akışlarını AsAgent() aracıya dönüştürmek için uzantı yöntemini kullanın:

using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

// First, build your workflow
var workflow = AgentWorkflowBuilder
    .CreateSequentialPipeline(researchAgent, writerAgent, reviewerAgent)
    .Build();

// Convert the workflow to an agent
AIAgent workflowAgent = workflow.AsAgent(
    id: "content-pipeline",
    name: "Content Pipeline Agent",
    description: "A multi-agent workflow that researches, writes, and reviews content"
);

AsAgent Parametreleri

Parametre Türü Description
id string? Aracı için isteğe bağlı özgün tanımlayıcı. Sağlanmadıysa otomatik olarak oluşturulur.
name string? Aracı için isteğe bağlı görünen ad.
description string? Ajansın amacının isteğe bağlı açıklaması.
checkpointManager CheckpointManager? Oturumlar arasında kalıcılık için isteğe bağlı bir kontrol noktası yöneticisi.
executionEnvironment IWorkflowExecutionEnvironment? İsteğe bağlı yürütme ortamı. varsayılan olarak InProcessExecution.OffThread veya InProcessExecution.Concurrent iş akışı yapılandırmasına göre kullanılır.

İş Akışı Aracılarını Kullanma

Thread Oluşturma

İş akışı aracısı ile yapılan her konuşma, durumu yönetmek için bir iş parçacığı gerektirir:

// Create a new thread for the conversation
AgentThread thread = workflowAgent.GetNewThread();

Akışsız Yürütme

Yanıtın tamamını istediğiniz basit kullanım örnekleri için:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

AgentRunResponse response = await workflowAgent.RunAsync(messages, thread);

foreach (ChatMessage message in response.Messages)
{
    Console.WriteLine($"{message.AuthorName}: {message.Text}");
}

Akışta Çalıştırma

İş akışı yürütülürken gerçek zamanlı güncelleştirmeler için:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

await foreach (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // Process streaming updates from each agent in the workflow
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

Dış Giriş İsteklerini İşleme

Bir iş akışı dış giriş isteyen yürütücüler içerdiğinde (kullanarak RequestInfoExecutor), bu istekler aracı yanıtında işlev çağrıları olarak ortaya çıkar:

await foreach (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // Check for function call requests
    foreach (AIContent content in update.Contents)
    {
        if (content is FunctionCallContent functionCall)
        {
            // Handle the external input request
            Console.WriteLine($"Workflow requests input: {functionCall.Name}");
            Console.WriteLine($"Request data: {functionCall.Arguments}");

            // Provide the response in the next message
        }
    }
}

İş Parçacığı Serileştirme ve Yeniden Başlatma

İş akışı aracısı iş parçacıkları, kalıcılık için seri hale getirilebilir ve daha sonra devam ettirilebilir.

// Serialize the thread state
JsonElement serializedThread = thread.Serialize();

// Store serializedThread to your persistence layer...

// Later, resume the thread
AgentThread resumedThread = workflowAgent.DeserializeThread(serializedThread);

// Continue the conversation
await foreach (var update in workflowAgent.RunStreamingAsync(newMessages, resumedThread))
{
    Console.Write(update.Text);
}

İş Akışı Aracılarıyla Denetim Noktası Oluşturma

İşlem yeniden başlatmalarında iş akışı durumunu kalıcı hale getirmek için denetim noktası oluşturmayı etkinleştirin:

// Create a checkpoint manager with your storage backend
var checkpointManager = new CheckpointManager(new FileCheckpointStorage("./checkpoints"));

// Create workflow agent with checkpointing enabled
AIAgent workflowAgent = workflow.AsAgent(
    id: "persistent-workflow",
    name: "Persistent Workflow Agent",
    checkpointManager: checkpointManager
);

Gereksinimler

Bir iş akışını aracı olarak kullanmak için iş akışının başlangıç yürütücüsunun giriş olarak işleyebilmesi list[ChatMessage] gerekir. Bu, ChatAgent veya AgentExecutor kullanılırken otomatik olarak karşılanır.

İş Akışı Aracısı Oluşturma

Uyumlu herhangi bir iş akışını aracıya dönüştürmek için as_agent() çağırın.

from agent_framework import WorkflowBuilder, ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Create your chat client and agents
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

researcher = ChatAgent(
    name="Researcher",
    instructions="Research and gather information on the given topic.",
    chat_client=chat_client,
)

writer = ChatAgent(
    name="Writer", 
    instructions="Write clear, engaging content based on research.",
    chat_client=chat_client,
)

# Build your workflow
workflow = (
    WorkflowBuilder()
    .set_start_executor(researcher)
    .add_edge(researcher, writer)
    .build()
)

# Convert the workflow to an agent
workflow_agent = workflow.as_agent(name="Content Pipeline Agent")

as_agent Parametreleri

Parametre Türü Description
name str | None Aracı için isteğe bağlı görünen ad. Sağlanmadıysa otomatik olarak oluşturulur.

İş Akışı Aracılarını Kullanma

Thread Oluşturma

İş akışı aracısı ile yapılan her konuşma, durumu yönetmek için bir iş parçacığı gerektirir:

# Create a new thread for the conversation
thread = workflow_agent.get_new_thread()

Akışsız Yürütme

Yanıtın tamamını istediğiniz basit kullanım örnekleri için:

from agent_framework import ChatMessage, Role

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

response = await workflow_agent.run(messages, thread=thread)

for message in response.messages:
    print(f"{message.author_name}: {message.text}")

Akışta Çalıştırma

İş akışı yürütülürken gerçek zamanlı güncelleştirmeler için:

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

async for update in workflow_agent.run_stream(messages, thread=thread):
    # Process streaming updates from each agent in the workflow
    if update.text:
        print(update.text, end="", flush=True)

Dış Giriş İsteklerini İşleme

bir iş akışı dış giriş isteyen yürütücüler içerdiğinde (kullanarak RequestInfoExecutor), bu istekler işlev çağrıları olarak ortaya çıkar. İş akışı aracısı bekleyen istekleri izler ve devam etmeden önce yanıt bekler:

from agent_framework import (
    FunctionCallContent,
    FunctionApprovalRequestContent,
    FunctionApprovalResponseContent,
)

async for update in workflow_agent.run_stream(messages, thread=thread):
    for content in update.contents:
        if isinstance(content, FunctionApprovalRequestContent):
            # The workflow is requesting external input
            request_id = content.id
            function_call = content.function_call

            print(f"Workflow requests input: {function_call.name}")
            print(f"Request data: {function_call.arguments}")

            # Store the request_id to provide a response later

# Check for pending requests
if workflow_agent.pending_requests:
    print(f"Pending requests: {list(workflow_agent.pending_requests.keys())}")

Bekleyen İsteklere Yanıt Sağlama

Dış giriş isteğinden sonra iş akışı yürütmeye devam etmek için:

# Create a response for the pending request
response_content = FunctionApprovalResponseContent(
    id=request_id,
    function_call=function_call,
    approved=True,
)

response_message = ChatMessage(
    role=Role.USER,
    contents=[response_content],
)

# Continue the workflow with the response
async for update in workflow_agent.run_stream([response_message], thread=thread):
    if update.text:
        print(update.text, end="", flush=True)

Tam Örnek

Akış çıkışına sahip bir iş akışı aracısını gösteren tam bir örnek aşağıda verilmiştir:

import asyncio
from agent_framework import (
    ChatAgent,
    ChatMessage,
    Role,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework._workflows import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

    # Create specialized agents
    researcher = ChatAgent(
        name="Researcher",
        instructions="Research the given topic and provide key facts.",
        chat_client=chat_client,
    )

    writer = ChatAgent(
        name="Writer",
        instructions="Write engaging content based on the research provided.",
        chat_client=chat_client,
    )

    reviewer = ChatAgent(
        name="Reviewer",
        instructions="Review the content and provide a final polished version.",
        chat_client=chat_client,
    )

    # Build a sequential workflow
    workflow = (
        SequentialBuilder()
        .add_agents([researcher, writer, reviewer])
        .build()
    )

    # Convert to a workflow agent
    workflow_agent = workflow.as_agent(name="Content Creation Pipeline")

    # Create a thread and run the workflow
    thread = workflow_agent.get_new_thread()
    messages = [ChatMessage(role=Role.USER, content="Write about quantum computing")]

    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run_stream(messages, thread=thread):
        # Show when different agents are responding
        if update.author_name and update.author_name != current_author:
            if current_author:
                print("\n" + "-" * 40)
            print(f"\n[{update.author_name}]:")
            current_author = update.author_name

        if update.text:
            print(update.text, end="", flush=True)

    print("\n" + "=" * 60)
    print("Workflow completed!")


if __name__ == "__main__":
    asyncio.run(main())

Etkinlik Dönüşümünü Anlama

Bir iş akışı aracı olarak çalıştırıldığında, iş akışı olayları aracı yanıtlarına dönüştürülür. Yanıt türü, hangi yöntemi kullandığınıza bağlıdır:

  • run(): İş akışı tamamlandıktan sonra sonucun tamamını içeren bir AgentRunResponse döndürür
  • run_stream(): İş AgentRunResponseUpdate akışı yürütülürken nesneleri verir ve gerçek zamanlı güncelleştirmeler sağlar

Yürütme sırasında iç iş akışı olayları aracı yanıtlarına aşağıdaki gibi eşlenir:

İş Akışı Olayı Temsilci Yanıtı
AgentRunUpdateEvent AgentRunResponseUpdate olarak geçirildi (akış) veya AgentRunResponse içine toplandı (akış dışı)
RequestInfoEvent FunctionCallContent ve FunctionApprovalRequestContent olarak dönüştürüldü
Diğer olaylar Gözlemlenebilirlik için raw_representation'ye dahil edildi

Bu dönüştürme, gerektiğinde ayrıntılı iş akışı bilgilerine erişmeye devam ederken standart aracı arabirimini kullanmanıza olanak tanır.

Kullanım Örnekleri

1. Karmaşık Aracı İşlem Hatları

Çok aracılı iş akışını uygulamalarda kullanılmak üzere tek bir aracı olarak sarmalama:

User Request --> [Workflow Agent] --> Final Response
                      |
                      +-- Researcher Agent
                      +-- Writer Agent  
                      +-- Reviewer Agent

2. Ajan Bileşimi

İş akışı aracılarını daha büyük sistemlerde bileşen olarak kullanın:

  • İş akışı aracısı başka bir aracı tarafından araç olarak kullanılabilir
  • Birden çok iş akışı aracısı birlikte düzenlenebilir
  • İş akışı aracıları diğer iş akışlarının içine yerleştirilebilir

3. API Tümleştirmesi

Standart Aracı arabirimini bekleyen API'ler aracılığıyla karmaşık iş akışlarını kullanıma sunarak şunları etkinleştirebilirsiniz:

  • Gelişmiş arka uç iş akışları kullanan sohbet arabirimleri
  • Mevcut aracı tabanlı sistemlerle tümleştirme
  • Basit aracılardan karmaşık iş akışlarına aşamalı geçiş

Sonraki Adımlar

  • Last updated on