Dela via

Microsoft Agent Framework-arbetsflöden – Arbeta med agenter

Den här sidan innehåller en översikt över hur du använder agenter i Microsoft Agent Framework-arbetsflöden.

Översikt

Om du vill lägga till information i dina arbetsflöden kan du använda AI-agenter som en del av arbetsflödets körning. AI-agenter kan enkelt integreras i arbetsflöden, så att du kan skapa komplexa, intelligenta lösningar som tidigare var svåra att uppnå.

Lägga till en agent direkt i ett arbetsflöde

Du kan lägga till agenter i arbetsflödet via kanter:

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

// Create the agents first
AIAgent agentA = new ChatClientAgent(chatClient, instructions);
AIAgent agentB = new ChatClientAgent(chatClient, instructions);

// Build a workflow with the agents
WorkflowBuilder builder = new(agentA);
builder.AddEdge(agentA, agentB);
Workflow<ChatMessage> workflow = builder.Build<ChatMessage>();

Köra arbetsflödet

I arbetsflödet som skapades ovan omsluts agenterna i en exekutor som hanterar kommunikationen mellan agenten och andra delar av arbetsflödet. Exekveraren kan hantera tre meddelandetyper:

  • ChatMessage: Ett enda chattmeddelande
  • List<ChatMessage>: En lista över chattmeddelanden
  • TurnToken: En svängtoken som signalerar början på en ny sväng

Exekutorn aktiverar inte agenten för att svara förrän det mottar en TurnToken. Alla meddelanden som tas emot innan TurnToken buffras och skickas till agenten när TurnToken tas emot.

StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!"));
// Must send the turn token to trigger the agents. The agents are wrapped as executors.
// When they receive messages, they will cache the messages and only start processing
// when they receive a TurnToken. The turn token will be passed from one agent to the next.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    // The agents will run in streaming mode and an AgentRunUpdateEvent
    // will be emitted as new chunks are generated.
    if (evt is AgentRunUpdateEvent agentRunUpdate)
    {
        Console.WriteLine($"{agentRunUpdate.ExecutorId}: {agentRunUpdate.Data}");
    }
}

Använda den inbyggda agentexekutor

Du kan lägga till agenter i arbetsflödet via kanter:

from agent_framework import WorkflowBuilder
from agent_framework.azure import AzureChatClient
from azure.identity import AzureCliCredential

# Create the agents first
chat_client = AzureChatClient(credential=AzureCliCredential())
writer_agent: ChatAgent = chat_client.create_agent(
    instructions=(
        "You are an excellent content writer. You create new content and edit contents based on the feedback."
    ),
    name="writer_agent",
)
reviewer_agent = chat_client.create_agent(
    instructions=(
        "You are an excellent content reviewer."
        "Provide actionable feedback to the writer about the provided content."
        "Provide the feedback in the most concise manner possible."
    ),
    name="reviewer_agent",
)

# Build a workflow with the agents
builder = WorkflowBuilder()
builder.set_start_executor(writer_agent)
builder.add_edge(writer_agent, reviewer_agent)
workflow = builder.build()

Köra arbetsflödet

I arbetsflödet som skapades ovan omsluts agenterna i en exekutor som hanterar kommunikationen mellan agenten och andra delar av arbetsflödet. Exekveraren kan hantera tre meddelandetyper:

  • str: Ett enda chattmeddelande i strängformat
  • ChatMessage: Ett enda chattmeddelande
  • List<ChatMessage>: En lista över chattmeddelanden

När exekutorn tar emot ett meddelande av någon av dessa typer, kommer det att aktivera agenten att svara, och svarstypen kommer att vara ett AgentExecutorResponse-objekt. Den här klassen innehåller användbar information om agentens svar, inklusive:

  • executor_id: ID för exekutorn som skapade det här svaret
  • agent_run_response: Det fullständiga svaret från agenten
  • full_conversation: Hela konversationshistoriken hittills

Två möjliga händelsetyper relaterade till agenternas svar kan genereras när arbetsflödet körs:

  • AgentRunUpdateEvent innehåller delar av agentens svar när de genereras i streamingläge.
  • AgentRunEvent innehåller det fullständiga svaret från agenten i icke-direktuppspelningsläge.

Som standard omsluts agenter i exekverare som körs i strömningsläge. Du kan anpassa det här beteendet genom att skapa en anpassad exekverare. Mer information finns i nästa avsnitt.

last_executor_id = None
async for event in workflow.run_streaming("Write a short blog post about AI agents."):
    if isinstance(event, AgentRunUpdateEvent):
        if event.executor_id != last_executor_id:
            if last_executor_id is not None:
                print()
            print(f"{event.executor_id}:", end=" ", flush=True)
            last_executor_id = event.executor_id
        print(event.data, end="", flush=True)

Använda en anpassad agentexekutor

Ibland kanske du vill anpassa hur AI-agenter integreras i ett arbetsflöde. Du kan uppnå detta genom att skapa en anpassad exekverare. På så sätt kan du styra:

  • Agentens anrop: direktuppspelning eller icke-direktuppspelning
  • Meddelandetyperna som agenten hanterar, inklusive anpassade meddelandetyper
  • Agentens livscykel, inklusive initiering och rensning
  • Användning av agenttrådar och andra resurser
  • Ytterligare händelser som genereras under agentens körning, inklusive anpassade händelser
  • Integrering med andra arbetsflödesfunktioner, till exempel delade tillstånd och begäranden/svar
internal sealed class CustomAgentExecutor : Executor<CustomInput, CustomOutput>("CustomAgentExecutor")
{
    private readonly AIAgent _agent;

    /// <summary>
    /// Creates a new instance of the <see cref="CustomAgentExecutor"/> class.
    /// </summary>
    /// <param name="agent">The AI agent used for custom processing</param>
    public CustomAgentExecutor(AIAgent agent) : base("CustomAgentExecutor")
    {
        this._agent = agent;
    }

    public async ValueTask<CustomOutput> HandleAsync(CustomInput message, IWorkflowContext context)
    {
        // Retrieve any shared states if needed
        var sharedState = await context.ReadStateAsync<SharedStateType>("sharedStateId", scopeName: "SharedStateScope");

        // Render the input for the agent
        var agentInput = RenderInput(message, sharedState);

        // Invoke the agent
        // Assume the agent is configured with structured outputs with type `CustomOutput`
        var response = await this._agent.RunAsync(agentInput);
        var customOutput = JsonSerializer.Deserialize<CustomOutput>(response.Text);

        return customOutput;
    }
}

from agent_framework import (
    ChatAgent,
    ChatMessage,
    Executor,
    WorkflowContext,
    handler
)

class Writer(Executor):

    agent: ChatAgent

    def __init__(self, chat_client: AzureChatClient, id: str = "writer"):
        # Create a domain specific agent using your configured AzureChatClient.
        agent = chat_client.create_agent(
            instructions=(
                "You are an excellent content writer. You create new content and edit contents based on the feedback."
            ),
        )
        # Associate the agent with this executor node. The base Executor stores it on self.agent.
        super().__init__(agent=agent, id=id)

    @handler
    async def handle(self, message: ChatMessage, ctx: WorkflowContext[list[ChatMessage]]) -> None:
        """Handles a single chat message and forwards the accumulated messages to the next executor in the workflow."""
        # Invoke the agent with the incoming message and get the response
        messages: list[ChatMessage] = [message]
        response = await self.agent.run(messages)
        # Accumulate messages and send them to the next executor in the workflow.
        messages.extend(response.messages)
        await ctx.send_message(messages)

Nästa steg

  • Last updated on