slhaf/Partner
1
0
Fork 0
mirror of https://github.com/slhaf/Partner.git synced 2026-05-12 16:53:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(context): support assign response channel while running flow context creating

Browse Source
This commit is contained in:
slhafzjw
2026-05-02 16:32:25 +08:00
parent 2b575df3f9
commit afb896e6db
6 changed files with 34 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Partner-Core/src/main/java/work/slhaf/partner/module/communication/CommunicationProducer.java
View File

@@ -94,7 +94,7 @@ public class CommunicationProducer extends AbstractAgentModule.Running<PartnerRu
}
private void executeChat(PartnerRunningFlowContext runningFlowContext) {
StreamChatMessageConsumer consumer = ReplyDispatcher.INSTANCE.createConsumer(runningFlowContext.getTarget());
StreamChatMessageConsumer consumer = ReplyDispatcher.INSTANCE.createConsumer(runningFlowContext.getTarget(), runningFlowContext.getResopnseChannel());
this.streamChat(buildChatMessages(runningFlowContext), consumer)
.onFailure(exception -> consumer.onDelta(INTERRUPTED_MARKER));
updateChatMessages(runningFlowContext, consumer.collectResponse());

19
Partner-Core/src/main/java/work/slhaf/partner/module/communication/ReplyDispatcher.kt
View File

@@ -30,7 +30,7 @@ object ReplyDispatcher {
} ?: break
if (nextChunk.isClosed) {
flush(builder.toString(), firstChunk.target)
flush(builder.toString(), firstChunk.target, firstChunk.responseChannel)
return@launch
}
@@ -43,7 +43,7 @@ object ReplyDispatcher {
}
}
flush(builder.toString(), firstChunk.target)
flush(builder.toString(), firstChunk.target, firstChunk.responseChannel)
}
}
}
@@ -51,7 +51,7 @@ object ReplyDispatcher {
/**
* flush 将推送至 AgentRuntime 的默认通道。
*/
private fun flush(content: String, target: String) {
private fun flush(content: String, target: String, responseChannel: String?) {
if (content.isEmpty()) {
return
}
@@ -62,22 +62,29 @@ object ReplyDispatcher {
mode = ReplyEvent.ContentMode.APPEND,
done = false
)
AgentRuntime.response(event)
if (responseChannel == null) {
AgentRuntime.response(event)
}else{
AgentRuntime.response(event, responseChannel)
}
}
fun createConsumer(target: String): StreamChatMessageConsumer = ReplyConsumer(
fun createConsumer(target: String, responseChannel: String?): StreamChatMessageConsumer = ReplyConsumer(
collectorChannel = collectorChannel,
target = target,
responseChannel = responseChannel
)
private data class ReplyChunk(
val delta: String,
val target: String,
val responseChannel: String?,
)
private class ReplyConsumer(
private val collectorChannel: Channel<ReplyChunk>,
private val target: String,
private val responseChannel: String?
) : StreamChatMessageConsumer() {
private enum class VisibilityState {
@@ -156,7 +163,7 @@ object ReplyDispatcher {
}
private fun flushVisible(delta: String) {
collectorChannel.trySend(ReplyChunk(delta, target)).isSuccess
collectorChannel.trySend(ReplyChunk(delta, target, responseChannel)).isSuccess
}
override fun consumeDelta(delta: String?) {

23
Partner-Core/src/main/java/work/slhaf/partner/runtime/PartnerRunningFlowContext.kt
View File

@@ -8,8 +8,9 @@ import work.slhaf.partner.framework.agent.interaction.flow.RunningFlowContext
class PartnerRunningFlowContext private constructor(
override val source: String,
inputs: List<InputEntry>,
firstInputEpochMillis: Long
) : RunningFlowContext(inputs, firstInputEpochMillis) {
firstInputEpochMillis: Long,
responseChannel: String? = null
) : RunningFlowContext(inputs, firstInputEpochMillis, responseChannel) {
companion object {
@@ -32,12 +33,18 @@ class PartnerRunningFlowContext private constructor(
@JvmStatic
@JvmOverloads
fun fromUser(userId: String, input: String, receivedAtMillis: Long = System.currentTimeMillis()) =
fun fromUser(
userId: String,
input: String,
receivedAtMillis: Long = System.currentTimeMillis(),
responseChannel: String? = null
) =
PartnerRunningFlowContext(
SourceTag.buildUserSource(userId),
listOf(InputEntry(0L, input)),
receivedAtMillis
)
receivedAtMillis,
responseChannel
).apply { this.target = userId }
@JvmStatic
@JvmOverloads
@@ -45,7 +52,8 @@ class PartnerRunningFlowContext private constructor(
PartnerRunningFlowContext(
SourceTag.buildAgentSource(),
listOf(InputEntry(0L, input)),
receivedAtMillis
receivedAtMillis,
null
).apply {
putUserInfo(InfoKeys.PLATFORM, SOURCE_SELF_PLATFORM)
putUserInfo(InfoKeys.NICKNAME, SOURCE_SELF_NICKNAME)
@@ -56,7 +64,8 @@ class PartnerRunningFlowContext private constructor(
return PartnerRunningFlowContext(
source = source,
inputs = inputs,
firstInputEpochMillis = System.currentTimeMillis()
firstInputEpochMillis = System.currentTimeMillis(),
resopnseChannel
)
}

2
Partner-Core/src/main/java/work/slhaf/partner/runtime/gateway/WebSocketGateway.java
View File

@@ -52,7 +52,7 @@ public class WebSocketGateway extends WebSocketServer implements AgentGateway<In
@Override
public PartnerRunningFlowContext parseRunningFlowContext(InputData inputData) {
PartnerRunningFlowContext context = PartnerRunningFlowContext.fromUser(inputData.getSource(), inputData.getContent());
PartnerRunningFlowContext context = PartnerRunningFlowContext.fromUser(inputData.getSource(), inputData.getContent(),System.currentTimeMillis(), getChannelName());
inputData.getMeta().forEach(context::putUserInfo);
return context;
}