Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -27,4 +27,10 @@ public class AgentTurnCallbacks
/// and the tool may take a while before anything else is said
/// </summary>
public Func<Task>? OnStreamEndedWithToolCall { get; init; }

/// <summary>
/// The last say over what the conversation history keeps of the reply: called with the reply the agent wrote,
/// returns what is stored in its place. Lets a flow record a reply the caller cut off as the part they heard.
/// </summary>
public Func<string, string>? RewriteReply { get; init; }
}
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public class ConversationRelaySession
private bool _utteranceOpen;
private bool _streamedThisTurn;
private bool _dropStream;
private bool _cutInMidReply;

// The current turn's reply is in; the response watchdog says nothing more once it is set
private bool _turnReplied;
Expand Down Expand Up @@ -235,6 +236,7 @@ private async Task HandleMessageAsync(string json)
if (_isProcessing && _streamedThisTurn)
{
_dropStream = true;
_cutInMidReply = true;
}
}
var interruptedAfter = GetString(root, "durationUntilInterruptMs");
Expand Down Expand Up @@ -316,6 +318,7 @@ private async Task ProcessPendingAsync()
_utteranceOpen = false;
_streamedThisTurn = false;
_dropStream = false;
_cutInMidReply = false;
_turnReplied = false;
_turnStartedUtc = DateTime.UtcNow;
}
Expand All @@ -333,7 +336,8 @@ private async Task ProcessPendingAsync()
OnStreamStarted = stream ? CloseUtteranceAsync : null,
// A line said before a slow tool ("Okay, that's everything I need") is spoken now, and the watchdog
// counts the silence after it instead of waiting for the next response to close it
OnStreamEndedWithToolCall = stream ? CloseUtteranceAsync : null
OnStreamEndedWithToolCall = stream ? CloseUtteranceAsync : null,
RewriteReply = stream ? MarkWhatTheCallerCutOff : null
};

AssistantMessage? reply;
Expand Down Expand Up @@ -587,6 +591,40 @@ await SendIfAsync(() =>
}, new { type = "text", token = string.Empty, last = true });
}

/// <summary>
/// What the conversation history keeps of a reply the caller cut in on: the part that went out, and after it a
/// mark naming the rest, so the next turn knows the caller never heard it. Every other reply is kept as it is.
/// </summary>
/// <remarks>
/// Only a reply cut off while it was still being streamed can be marked: once the whole reply is out, Twilio
/// holds the text it has yet to speak and a cut-in there leaves nothing of ours to go by.
/// </remarks>
private string MarkWhatTheCallerCutOff(string content)
{
var final = content?.Trim() ?? string.Empty;
string sent;
lock (_sync)
{
if (!_cutInMidReply || final.Length == 0)
{
return content;
}
sent = (_utteranceOpen ? _streamSegment.ToString() : _lastStreamSegment).Trim();
}

// The reply is not what was streamed (text set by a tool, or a reply rewritten after the model wrote it),
// so where the caller cut in is not known
if (sent.Length == 0 || !final.StartsWith(sent, StringComparison.Ordinal))
{
return content;
}

var unheard = final.Substring(sent.Length).Trim();
return unheard.Length == 0
? content
: $"{sent} [the caller cut in here; they did not hear: \"{unheard}\"]";
}

/// <summary>
/// The part of the final reply that was not streamed. Normally nothing; the whole reply when it is not
/// what the model streamed last, e.g. text set by a tool, or a reply rewritten after the model wrote it.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -99,11 +99,14 @@ await conv.SendMessage(message.AgentId,
replyMessage: await BuildPostbackMessageModel(conv, message),
async msg =>
{
// What the caller is told, which a rewrite below must not change
var spoken = msg.Content;
RewriteForHistory(msg, callbacks);
reply = new AssistantMessage()
{
ConversationEnd = msg.Instruction?.ConversationEnd ?? false,
HumanIntervationNeeded = string.Equals("human_intervention_needed", msg.FunctionName),
Content = msg.Content,
Content = spoken,
MessageId = msg.MessageId
};
}
Expand All @@ -112,6 +115,31 @@ await conv.SendMessage(message.AgentId,
return reply;
}

/// <summary>
/// Lets the flow change what the history keeps of the reply. The dialog is appended after this callback runs,
/// so the message changed here is the one that is stored.
/// </summary>
private static void RewriteForHistory(RoleDialogModel msg, AgentTurnCallbacks? callbacks)
{
if (callbacks?.RewriteReply == null || string.IsNullOrEmpty(msg.Content))
{
return;
}

var stored = callbacks.RewriteReply(msg.Content);
if (string.IsNullOrEmpty(stored) || stored == msg.Content)
{
return;
}

// Kept in step only where it is the reply itself; a rich message the model refined is left alone
if (msg.RichContent?.Message != null && msg.RichContent.Message.Text == msg.Content)
{
msg.RichContent.Message.Text = stored;
}
msg.Content = stored;
}

private static async Task<PostbackMessageModel?> BuildPostbackMessageModel(
IConversationService conv, CallerMessage message)
{
Expand Down
Loading