.NET Agent Framework Fan-In Issue: AggregationExecutor Not Running

Nov 3, 2025 by Admin 67 views

Hey guys! So, you're diving into the awesome world of the .NET Agent Framework and trying to build some cool concurrent workflows, right? You probably followed that neat tutorial on creating a simple concurrent workflow, which is a fantastic way to get started. But then, BAM! You hit a snag. Your aggregationExecutor just isn't firing up, and you're left scratching your head wondering what went wrong. Don't worry, this is a super common hiccup, and we're going to untangle it together. Let's get this sorted so you can get back to building those powerful AI agents!

Understanding the Core Problem: The Fan-In Conundrum

Alright, let's chat about what's actually happening here. You've set up your WorkflowBuilder with a ConcurrentStartExecutor and a ConcurrentAggregationExecutor. You've added the AddFanOutEdge to send messages from the start to your agents (like our brilliant physicist and chemist personas) and then the AddFanInEdge to bring those agent responses back to the aggregationExecutor. The theory is sound – messages go out, come back, and get aggregated. So why isn't the aggregationExecutor seeing any messages?

The key to this puzzle often lies in how the messages and the turn token are being handled. In your ConcurrentStartExecutor, you're correctly sending the initial user message. Then, you're broadcasting a TurnToken to kick things off. This TurnToken is like a little signal saying, "Okay, agents, you can start processing now!" However, the ConcurrentAggregationExecutor is designed to receive ChatMessage objects. The problem arises because the TurnToken itself is not a ChatMessage. While the TurnToken successfully initiates the agents you've fanned out to, it doesn't provide the data that the aggregationExecutor is explicitly waiting for – which is a ChatMessage.

Think of it like this: you send out invitations (user messages) and then ring the doorbell (turn token) to get people to come to the party. The physicist and chemist agents hear the doorbell and start chatting. But the aggregationExecutor is like the host who's waiting to greet each guest personally with a handshake (a ChatMessage) before they can start mingling. If the host only hears the doorbell but never gets the actual handshake, they don't know who's arrived or that they should start the welcoming speech.

The ConcurrentAggregationExecutor in your ConcurrentAggregationExecutor.cs is set up with Executor<ChatMessage>. This means it's specifically typed to expect and handle ChatMessage objects. When the ConcurrentStartExecutor sends the TurnToken, it's not the ChatMessage type the ConcurrentAggregationExecutor is looking for. This is why, even though the agents might be processing, their responses (which are ChatMessage objects) aren't getting directed or recognized by the aggregationExecutor in the way you expect.

The tutorial you followed likely has a slightly different structure or expectation for how the TurnToken interacts with subsequent executors, especially in a fan-in scenario. We need to ensure that the output from the fanned-out agents is correctly captured and passed to the aggregationExecutor.

Let's dive deeper into how we can adjust the flow to make sure those messages get to the right place and the aggregationExecutor does its job!

Debugging the Workflow Execution Flow

Okay, so we've identified that the aggregationExecutor isn't running, and we suspect it's a communication breakdown. The best way to tackle this is to become a detective and trace the flow of messages and events. We'll add some strategic logging to see exactly what's happening at each step.

First, let's revisit the ConcurrentStartExecutor. This executor is responsible for initiating the workflow. In your current setup, it sends the initial user message and then a TurnToken. The TurnToken is crucial because it tells the downstream agents (the physicist and chemist) that they should start processing the messages they've received. Without it, they'd just be sitting there, waiting.

Now, the ConcurrentAggregationExecutor is designed to receive ChatMessage objects. The key insight here is that the TurnToken itself isn't a ChatMessage. While the TurnToken correctly prompts the physicist and chemist agents to respond, the aggregationExecutor isn't configured to directly process these TurnToken signals. It's waiting for the actual chat messages from the agents.

Let's look at the provided code snippet for Program.cs. You have this excellent await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) loop. This is your primary window into what the workflow is doing. Currently, you're printing [Event] {evt.GetType().Name}: {evt}. This is great! Keep this in! When you run your code, what do you see here? Do you see TurnToken events? Do you see ChatMessage events being generated by the physicist and chemist? Do you see any events related to the aggregationExecutor trying to handle something?

If you don't see ChatMessage events originating from the physicist or chemist agents, it means they haven't successfully responded. This could be due to the TurnToken not being interpreted correctly, or perhaps an issue with the agent setup itself (though less likely if they are basic ChatClientAgent instances).

If you do see ChatMessage events from the agents, but no events related to the aggregationExecutor being invoked or processing these messages, then it strongly suggests that the output of the physicist and chemist agents isn't being correctly routed or received by the aggregationExecutor. The AddFanInEdge(aggregationExecutor, sources: [physicist, chemist]) part is supposed to handle this routing, but maybe there's a nuance we're missing.

One crucial aspect of workflows, especially with fan-in, is ensuring that the output of the fanned-out tasks is what feeds into the fan-in step. The ConcurrentStartExecutor doesn't produce a ChatMessage output that gets sent to the aggregationExecutor. It sends a ChatMessage to the agents and a TurnToken to kick them off. The aggregationExecutor needs to receive the actual ChatMessage responses from the physicist and chemist.

Let's consider how the TurnToken is meant to work. It acts as a signal. When the physicist and chemist receive the TurnToken, they should then proceed to generate their ChatMessage responses. These responses are then what should be captured by the aggregationExecutor.

So, the core debugging strategy is: Verify that the physicist and chemist agents are indeed generating ChatMessage responses after receiving the TurnToken, and check if these ChatMessage events are reaching the aggregationExecutor. Your WatchStreamAsync loop is your best friend here for observing this flow.

The Solution: Correcting the Fan-In Logic

Okay, team, let's zero in on the fix! Based on our detective work, the main culprit is how the ConcurrentAggregationExecutor is expecting input and how the workflow is currently set up to provide it. Remember, the aggregationExecutor is typed as Executor<ChatMessage>, meaning it's specifically designed to receive ChatMessage objects. The TurnToken sent by the ConcurrentStartExecutor isn't a ChatMessage, so it doesn't directly get processed by the aggregationExecutor.

What we need is for the aggregationExecutor to receive the actual ChatMessage responses from the physicist and chemist agents. The AddFanInEdge(aggregationExecutor, sources: [physicist, chemist]) configuration is correctly telling the framework that the aggregationExecutor should listen to outputs from physicist and chemist. The issue isn't the connection itself, but rather ensuring that the right type of data flows through that connection.

Here’s the strategic adjustment: The ConcurrentStartExecutor’s job is to initiate. It sends the user's query to the fanned-out agents and signals them to proceed. The physicist and chemist agents, upon receiving the TurnToken, should then generate their ChatMessage responses. These ChatMessage responses are what should be directed to the aggregationExecutor.

Let's refine the ConcurrentStartExecutor. It's already sending the user's message. The TurnToken is necessary to activate the downstream agents. The problem isn't that the TurnToken is being sent instead of a ChatMessage to the aggregationExecutor; it's that the TurnToken is being sent to the agents, and we need to ensure their responses are captured.

The provided ConcurrentAggregationExecutor.cs correctly aggregates messages once it receives them (this._messages.Count == 2). The issue is that it's likely never reaching the if (this._messages.Count == 2) condition because it's not receiving the expected ChatMessage objects from the fanned-out agents.

The key insight: The AddFanOutEdge(startExecutor, targets: [physicist, chemist]) sends messages from startExecutor to physicist and chemist. The AddFanInEdge(aggregationExecutor, sources: [physicist, chemist]) should mean that messages from physicist and chemist are sent to aggregationExecutor. The TurnToken itself isn't the message being aggregated; it's the prompt for the agents to generate their own messages.

Let's consider the flow again:

ConcurrentStartExecutor receives `input: