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

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, AsAIAgent ile oluşturulan 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ı AsAIAgent() aracıya dönüştürmek için uzantı yöntemini kullanın:

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

// Create agents
AIAgent researchAgent = chatClient.AsAIAgent("You are a researcher. Research and gather information on the given topic.");
AIAgent writerAgent = chatClient.AsAIAgent("You are a writer. Write clear, engaging content based on research.");
AIAgent reviewerAgent = chatClient.AsAIAgent("You are a reviewer. Review the content and provide a final polished version.");

// Build a sequential workflow
var workflow = new WorkflowBuilder(researchAgent)
    .AddEdge(researchAgent, writerAgent)
    .AddEdge(writerAgent, reviewerAgent)
    .Build();

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

AsAIAgent Parametreleri

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

Oturum Oluşturma

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

// Create a new session for the conversation
AgentSession session = await workflowAgent.CreateSessionAsync();

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")
};

AgentResponse response = await workflowAgent.RunAsync(messages, session);

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 (AgentResponseUpdate update in workflowAgent.RunStreamingAsync(messages, session))
{
    // 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 (AgentResponseUpdate update in workflowAgent.RunStreamingAsync(messages, session))
{
    // 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
        }
    }
}

Oturum Serileştirme ve Yeniden Başlatma

İş akışı aracısı oturumları kalıcılık için seri hale getirilebilir ve daha sonra sürdürülebilir:

// Serialize the session state
JsonElement serializedSession = await workflowAgent.SerializeSessionAsync(session);

// Store serializedSession to your persistence layer...

// Later, resume the session
AgentSession resumedSession = await workflowAgent.DeserializeSessionAsync(serializedSession);

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

Gereksinimler

Bir iş akışını aracı olarak kullanmak için iş akışının başlangıç yürütücüsü ileti girişini işleyebilmelidir. Bu, Agent veya aracı tabanlı yürütücüler 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.foundry import FoundryChatClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential

# Create your chat client and agents
client = FoundryChatClient(
    project_endpoint="<your-endpoint>",
    model="<your-deployment>",
    credential=AzureCliCredential(),
)

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

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

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

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

as_agent Parametreleri

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

Oturum Oluşturma

İsteğe bağlı olarak, konuşma durumunu birden çok sırayla yönetmek için bir oturum oluşturabilirsiniz:

# Create a new session for the conversation
session = await workflow_agent.create_session()

Akışsız Yürütme

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

# You can pass a plain string as input
response = await workflow_agent.run("Write an article about AI trends")

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:

async for update in workflow_agent.run(
    "Write an article about AI trends",
    stream=True,
):
    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 request_info), bu istekler aracı yanıtında işlev çağrıları olarak ortaya çıkar. İşlev çağrısı şu adı WorkflowAgent.REQUEST_INFO_FUNCTION_NAMEkullanır:

from agent_framework import Content, Message, WorkflowAgent

response = await workflow_agent.run("Process my request")

# Look for function calls in the response
human_review_function_call = None
for message in response.messages:
    for content in message.contents:
        if content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
            human_review_function_call = content

Bekleyen İsteklere Yanıt Sağlama

Dış giriş isteğinden sonra iş akışı yürütmeye devam etmek için bir işlev sonucu oluşturun ve geri gönderin:

if human_review_function_call:
    # Parse the request arguments
    request = WorkflowAgent.RequestInfoFunctionArgs.from_json(
        human_review_function_call.arguments
    )

    # Create a response (your custom response type)
    result_data = MyResponseType(approved=True, feedback="Looks good")

    # Create the function call result
    function_result = Content.from_function_result(
        call_id=human_review_function_call.call_id,
        result=result_data,
    )

    # Send the response back to continue the workflow
    response = await workflow_agent.run(Message("tool", [function_result]))

Tam Örnek

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

import asyncio
import os

from agent_framework.foundry import FoundryChatClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        credential=AzureCliCredential(),
    )

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

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

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

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

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

    # Run the workflow
    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run(
        "Write about quantum computing",
        stream=True,
    ):
        # 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ü, nasıl çağırdığınıza run()bağlıdır:

• run(): İş akışı tamamlandıktan sonra sonucun tamamını içeren bir AgentResponse döndürür
• run(..., stream=True): İş akışı yürütülürken AgentResponseUpdate nesnelerinin zaman uyumsuz olarak yinelenebilir bir şeklini döndürür ve gerçek zamanlı güncellemeler sunar.

as_agent(), hem "output" (terminal) hem de "intermediate" olaylarını çağıran tarafa iletir. İletilen olay türleri kümesidir AGENT_FORWARDED_EVENT_TYPES = {"output", "intermediate"}. Diğer tüm iş akışı içindeki olaylar göz ardı edilir.

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

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. Hem .text hem de AgentResponse üzerindeki AgentResponseUpdate özelliği yalnızca nihai ("output") yanıtı döndürür; ara ilerleme bilgilerine erişmek için text_reasoning içerik öğelerini inceleyin.

Go, iş akışlarını workflow/agentworkflow ile ajanlar olarak paketler. Bu, sağlayıcı iş akışını arka planda yürütürken çağıran tarafların normal ajan çalıştırma API'lerini kullanmasına olanak tanır.

Gereksinimler

İş akışının başlangıç yürütücüsü kabul []*message.Messageetmelidir. Barındırılan aracı yürütücüleri ve messageworkflow.Configure ile yapılandırılmış yürütücüler bu gereksinimi karşılar.

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

agentworkflow.New kullanarak herhangi bir uyumlu iş akışını ajan olarak sarmalayın:

wfAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
    IncludeOutputsInResponse: true,
    Config: agent.Config{
        Name: "WorkflowAgent",
    },
})
if err != nil {
    return err
}

agentworkflow.AgentConfig Parametreleri

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

Oturum Oluşturma

İş akışı durumunun etkileşimler boyunca kalıcı olmasını istediğinizde bir ajan oturumu oluşturun:

session, err := wfAgent.CreateSession(ctx)
if err != nil {
    return err
}

Akışsız Yürütme

RunText veya Run kullanın ve akışsız yürütme için yanıtı toplayın:

response, err := wfAgent.RunText(ctx, "Analyze this", agent.WithSession(session)).Collect()
if err != nil {
    return err
}
fmt.Println(response.String())

Akışta Çalıştırma

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

for update, err := range wfAgent.RunText(ctx, "Analyze this", agent.WithSession(session), agent.Stream(true)) {
    if err != nil {
        return err
    }
    fmt.Print(update.String())
}

Dış Giriş İsteklerini İşleme

İş akışından gelen dış istekler, aracı yanıtında işlev çağrısı içeriği olarak ortaya çıkar. İstek içeriği için yanıt iletilerini inceleyin ve eşleşen yanıtı sonraki bir çalıştırmada gönderin.

var requestCall *message.FunctionCallContent
for content := range response.Contents() {
    if call, ok := content.(*message.FunctionCallContent); ok {
        requestCall = call
        break
    }
}

Bekleyen İsteklere Yanıt Sağlama

İş akışı yürütmeye devam etmek için eşleşen yanıt içeriğini iş akışı aracısına döndürin:

result := &message.FunctionResultContent{
    CallID: requestCall.CallID,
    Result: "approved",
}

response, err = wfAgent.Run(
    ctx,
    []*message.Message{{
        Role:     message.RoleTool,
        Contents: []message.Content{result},
    }},
    agent.WithSession(session),
).Collect()
if err != nil {
    return err
}

Oturum Serileştirme ve Yeniden Başlatma

İş akışı aracısı oturumları kalıcılık için seri hale getirilebilir ve daha sonra sürdürülebilir:

// Serialize the session state.
serializedSession, err := json.Marshal(session)
if err != nil {
    return err
}

// Store serializedSession to your persistence layer...

// Later, resume the session.
var resumedSession agent.Session
if err := json.Unmarshal(serializedSession, &resumedSession); err != nil {
    return err
}

for update, err := range wfAgent.RunText(ctx, "Continue the article", agent.WithSession(&resumedSession), agent.Stream(true)) {
    if err != nil {
        return err
    }
    fmt.Print(update.String())
}

Tam Örnek

Aşağıdaki örnek, bir içerik işlem hattı iş akışı oluşturur, bunu bir aracı olarak paketler ve yanıtları standart aracı API'si üzerinden akıtır:

package main

import (
    "cmp"
    "context"
    "fmt"
    "log"
    "os"

    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
)

func main() {
    ctx := context.Background()
    endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
    model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

    credential, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        log.Fatal(err)
    }

    researcher := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Research and gather information on the given topic.",
        Config:      agent.Config{Name: "Researcher"},
    })
    writer := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Write clear, engaging content based on research.",
        Config:      agent.Config{Name: "Writer"},
    })
    reviewer := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Review the content and provide a final polished version.",
        Config:      agent.Config{Name: "Reviewer"},
    })

    wf, err := agentworkflow.NewSequentialWorkflowBuilder(researcher, writer, reviewer).
        WithName("content-pipeline").
        Build()
    if err != nil {
        log.Fatal(err)
    }

    wfAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
        IncludeOutputsInResponse: true,
        Config: agent.Config{
            Name: "Content Pipeline Agent",
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    session, err := wfAgent.CreateSession(ctx)
    if err != nil {
        log.Fatal(err)
    }

    for update, err := range wfAgent.RunText(ctx, "Write about quantum computing", agent.WithSession(session), agent.Stream(true)) {
        if err != nil {
            log.Fatal(err)
        }
        if text := update.String(); text != "" {
            fmt.Print(text)
        }
    }
}