Microsoft Agent Framework-Workflows – Arbeiten mit Agents

Direktes Hinzufügen eines Agents zu einem Workflow

Sie können Ihrem Workflow Agents über Ränder hinzufügen:

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>();

Ausführen des Workflows

Innerhalb des oben erstellten Workflows werden die Agents tatsächlich in einen Executor eingeschlossen, der die Kommunikation des Agents mit anderen Teilen des Workflows verarbeitet. Der Executor kann drei Nachrichtentypen verarbeiten:

• ChatMessage: Eine einzelne Chatnachricht
• List<ChatMessage>: Eine Liste der Chatnachrichten
• TurnToken: Ein Turntoken, das den Beginn einer neuen Drehung signalisiert

Der Executor veranlasst den Agenten erst zu reagieren, wenn er eine TurnToken erhält. Alle Nachrichten, die vor dem TurnToken empfangen werden, werden zwischengespeichert und beim Empfang des TurnToken an den Agenten gesendet.

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

Verwenden des integrierten Agent-Executors

Sie können Ihrem Workflow Agents über Ränder hinzufügen:

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()

Ausführen des Workflows

Innerhalb des oben erstellten Workflows werden die Agents tatsächlich in einen Executor eingeschlossen, der die Kommunikation des Agents mit anderen Teilen des Workflows verarbeitet. Der Executor kann drei Nachrichtentypen verarbeiten:

• str: Eine einzelne Chatnachricht im Zeichenfolgenformat
• ChatMessage: Eine einzelne Chatnachricht
• List<ChatMessage>: Eine Liste der Chatnachrichten

Wenn der Executor eine Nachricht eines dieser Typen empfängt, löst er den Agent aus, um zu antworten, und der Antworttyp ist ein AgentExecutorResponse Objekt. Diese Klasse enthält nützliche Informationen zur Antwort des Agents, einschließlich:

• executor_id: Die ID des Ausführenden, der diese Antwort erzeugt hat
• agent_run_response: Die vollständige Antwort des Agents
• full_conversation: Der vollständige Unterhaltungsverlauf bis zu diesem Punkt

Zwei mögliche Ereignistypen im Zusammenhang mit den Antworten der Agents können beim Ausführen des Workflows ausgegeben werden:

• AgentRunUpdateEvent enthält Abschnitte der Antwort des Agents, die im Streamingmodus generiert werden.
• AgentRunEvent enthält die vollständige Antwort des Agents im Nicht-Streaming-Modus.

Standardmäßig werden Agents in Executoren umschlossen, die im Streamingmodus ausgeführt werden. Sie können dieses Verhalten anpassen, indem Sie einen benutzerdefinierten Executor erstellen. Weitere Informationen finden Sie im nächsten Abschnitt.

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)

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)