How to restore the history / thread in a workflow for either an agent or the whole workflow?

[Cotspheer]

I'm currently having a hard time to understand when and how threads / message histories are loaded / restored and what I have to do manually. Sadly, non of the samples explicitly shows that hence the question.

I try to build a workflow that takes in a visitors question, tries to figure out to which area it belongs to and then provides the facts necessary to a final agent - VoiceAgent - that will produce a script which in turn is returned back as Text-To-Speech. To jump start things I thought I just create a simple workflow that takes in the users questions and forwards it to the VoiceAgent. The agent is configured with a simple sample prompt that later on should be dynamic enhanced / enriched during the workflow. For the steps in-between the history is not necessary but I planned to store the output messages in the history / thread of the VoiceAgent - all have the VisitorId as the ThreadId / Key - and then make that available to the VoiceAgent. Additionally the VoiceAgent must know the previous / recent messages to get the conversation context.

So for now it should be: UserInputExecutor => VoiceAgent => Result

Notes:

• The VisitorId is generated once per tour and sticks until the journey is restarted.
• Messages are getting stored using my Repository. Just restoring the history is not working.

I saw that I can do workflow.Build().AsAgent() and then either call GetNewThread() or DeserializeThread() but it looks like that is not the thing I'm looking for.

So to summarize: How and when is ChatMessageStore.GetMessagesAsync getting called? What steps do I have to implement to make sure it is getting called and made available to the agent?

Any suggestion on what I've missed? Thanks in advance for any advice! If I missed something in the documentation just drop me an URL and I will be happy to read it.

First Attempt - Workflow directly
I assumed that something will trigger the ChatMessageStore.GetMessagesAsync.

        var workflow = new WorkflowBuilder(userInput)
            .AddEdge(userInput, agent)
            .WithOutputFrom(agent)
            .Build();

        var message = new ChatMessage
        {
            Role = ChatRole.User,
            AuthorName = "JohnDoe",
            Contents = new[] {
                    new TextContent(dto.Text)
                }
        };

        await using StreamingRun run = await InProcessExecution.StreamAsync(workflow.Build(), input: message);
        var response = await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); // Kickstart the Workflow
        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
        {
                    // [...] some code
        }

Second Attempt - Workflow as Agent and passing by the existing thread

        // Trying to restore the thread on the agent but this does not result in a call to `ChatMessageStore.GetMessagesAsync`
        var thread = agent.DeserializeThread(JsonSerializer.SerializeToElement(ChatStore.StoreState.From(this.VisitorId)));

        var workflow = new WorkflowBuilder(userInput)
            .AddEdge(userInput, agent)
            .WithOutputFrom(agent)
            .Build()
            .AsAgent();

        var message = new ChatMessage
        {
            Role = ChatRole.User,
            AuthorName = "JohnDoe",
            Contents = new[] {
                    new TextContent(dto.Text)
                }
        };

        // This does not work as it gives the error: Workflow does not support ChatProtocol: At least List<ChatMessage> and TurnToken must be supported as input.
        await foreach (var agentResponse in workflow.RunStreamingAsync(message, thread, cancellationToken: cancellationToken))
        {
                    // [...] some code
        }

Agent

        AIAgent agent = this.serviceProvider.GetRequiredService<IChatClient>().CreateAIAgent
        (
           new ChatClientAgentOptions
           {
               Name = "VoiceGuide",
               Instructions = prompt,
               ChatMessageStoreFactory = ctx =>
               {
                   var repository = this.serviceProvider.GetRequiredService<IChatRepository>();

                   if (ctx.SerializedState.ValueKind == JsonValueKind.Object)
                   {
                       return new ChatStore(repository, ctx.SerializedState, ctx.JsonSerializerOptions);
                   }

                   return new ChatStore(repository, ChatStore.StoreState.From(this.VisitorId));
               },
           },
           services: serviceProvider
        );

ChatStore

public class ChatStore : ChatMessageStore
{
    private readonly IChatRepository chatRepository;

    private readonly StoreState state;

    public sealed class StoreState
    {
        public VisitorId VisitorId { get; set; }

        public static StoreState From(VisitorId visitorId)
        {
            return new StoreState
            {
                VisitorId = visitorId,
            };
        }
    }

    public ChatStore
    (
        IChatRepository chatRepository,
        StoreState state
    )
    {
        this.chatRepository = chatRepository;
        this.state = state;
    }

    public ChatStore
    (
        IChatRepository chatRepository,
        JsonElement serializedStoreState,
        JsonSerializerOptions? jsonSerializerOptions = null
    )
    {
        this.chatRepository = chatRepository;
        this.state = serializedStoreState.Deserialize<StoreState>(jsonSerializerOptions)!;
    }


    public override async Task AddMessagesAsync(IEnumerable<ChatMessage> messages, CancellationToken cancellationToken = default)
    {
        await this.chatRepository.AppendMessages(this.state.VisitorId, messages, cancellationToken);
    }

    public override async Task<IEnumerable<ChatMessage>> GetMessagesAsync(CancellationToken cancellationToken = default)
    {
        return await this.chatRepository.GetMessages(this.state.VisitorId, cancellationToken);
    }

    public override JsonElement Serialize(JsonSerializerOptions? jsonSerializerOptions = null)
    {
        return JsonSerializer.SerializeToElement(this.state, jsonSerializerOptions);
    }
}

UserInputExecutor - As Start-Node

    internal sealed class UserInputExecutor() : Executor<ChatMessage, ChatMessage>("UserInput")
    {
        public override async ValueTask<ChatMessage> HandleAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
        {
            await context.SendMessageAsync(message, cancellationToken);
            await context.SendMessageAsync(new TurnToken(emitEvents: true), cancellationToken: cancellationToken);

            return message;
        }
    }

[Cotspheer]

PS: If I just call the agent directly without having it inside a workflow the ChatMessageStore.GetMessagesAsync is actually executed when calling var thread = agent.DeserializeThread(JsonSerializer.SerializeToElement(ChatStore.StoreState.From(this.VisitorId))); first.

[lokitoth]

When agents are hosted in workflows, in general the suggested path is to leave the management of the agent threads to the workflow and rely on Checkpointing to resume in the right place. This has the advantage of working even in cases of server-side threads for, e.g. persisted agents, because the checkpointing mechanism will either externalize the actual chat store state (and restore it during checkpoint resume), or just the conversation id.

The way this is managed for WorkflowHostAgent, which is what backs the .AsAgent() call, is twofold: You can provide a CheckpointManager directly on the AsAgent call, which will then manage the thread state for you.

Alternatively, if you do not provide one, an InMemory CheckpointManager will be created, which is used to drive the Serialize call on the workflow agent's thread, and then to deserialize it back using DeserializeThread.

If you just wire up the agent together sequentially, each subsequent agent will get the outgoing messages from the source agent, but not the ones coming into the source agent. If you need something more akin to a handoff-with-full-context, you might want to try to use the HandoffsWorkflowBuilder to use a higher-level "orchestration" pattern, but that does increase the complexity of the underlying workflow somewhat.

As an addendum - right now providing your own ChatMessageStore for the WorkflowHostAgent's AgentThread is not supported, because the management of new vs. already processed messages is not cleanly factored from the storage. It is something we are open to implementing, but do not have current plans to do so.

[Cotspheer]

Thank you so much for clarifying things! I'll check CheckpointManager for sure and will tinker around that!

[for-shariq]

@lokitoth What is the best way to Resume / Rehyderate a workflow using Checkpoint, if it being used as an Agent. Because i only see the samples to resume using a Checkpoint Run. In my case i am using a thread agent.RunStreamingAsync(messages, thread).