Losita/smartmate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Version: 0.9.26.dev.260417

Browse Source 
后端：
1. Prompt 层从 execute 专属骨架重构为全节点统一四段式 buildUnifiedStageMessages
  - 新增 unified_context.go：定义 StageMessagesConfig + buildUnifiedStageMessages 统一骨架，所有节点（Chat/Plan/Execute/Deliver/DeepAnswer）共用同一套 msg0~msg3 拼装逻辑
  - 新增 conversation_view.go：通用对话历史渲染 buildConversationHistoryMessage，各节点复用，不再各自维护提取逻辑
  - 新增 chat_context.go / plan_context.go / deliver_context.go：各节点自行渲染 msg1（对话视图）和 msg2（工作区），统一层只负责"怎么拼"，不再替节点决定"放什么"
  - Chat/Plan/Deliver/Execute 的 BuildXXXMessages 全部从 buildStageMessages 切到 buildUnifiedStageMessages，移除旧路径
  - 删除 execute_pinned.go：execute 记忆渲染合并到统一层 renderUnifiedMemoryContext
  - Plan prompt 不再在 user prompt 中拼装任务类 ID 列表和 renderStateSummary，改为依赖 msg2 规划工作区；Chat 粗排判断从"上下文有任务类 ID"改为"批量调度需求"
  - Deliver prompt 新增 IsAborted/IsExhaustedTerminal 区分，支持粗排收口和主动终止场景
2. Execute ReAct 上下文简化——移除归档搬运、窗口裁剪和重复工具压缩
  - 移除 splitExecuteLoopRecordsByBoundary、findLatestExecuteBoundaryMarker、tailExecuteLoops、compressExecuteLoopObservationsByTool、buildEarlyExecuteReactSummary、trimExecuteMessage1ByBudget 等六个函数
  - 移除 executeLoopWindowLimit / executeConversationTurnLimit / executeMessage1MaxRunes 等预算常量
  - msg1 不再从历史中归档上一轮 ReAct 结果，只保留真实对话流（user + assistant speak），全量注入
  - msg2 不再按 loop_closed / step_advanced 边界切分"归档/活跃"，直接全量注入全部 ReAct Loop 记录
  - token 预算由统一压缩层兜底，prompt 层不再做提前裁剪
3. 压缩层从 Execute 专属提升为全节点通用 UnifiedCompact
  - 删除 execute_compact.go（Execute 专属压缩文件）
  - 新增 unified_compact.go：UnifiedCompactInput 参数化，各节点（Plan/Chat/Deliver/Execute）构造时从自己的 NodeInput 提取公共字段，消除对 Execute 的直接依赖
  - CompactionStore 接口扩展 LoadStageCompaction / SaveStageCompaction，各节点按 stageKey 独立维护压缩状态互不覆盖
  - 非 4 段式消息时退化成按角色汇总统计，确保 context_token_stats 仍然刷新
4. Retry 重试机制全面下线
  - dao/agent.go：saveChatHistoryCore / SaveChatHistory / SaveChatHistoryInTx 移除 retry_group_id / retry_index /
  retry_from_user_message_id / retry_from_assistant_message_id 四个参数，修复乱码注释
  - dao/agent-cache.go：移除 ApplyRetrySeed 和 extractMessageHistoryID 两个方法
  - conv/agent.go：ToEinoMessages 不再回灌 retry_* 字段到运行期上下文
  - service/agentsvc/agent.go：移除 chatRetryMeta 及 resolveRetryGroupID / buildRetrySeed 等全部重试逻辑
  - service/agentsvc/agent_quick_note.go：整个文件删除（retry 快速补写路径已无用）
  - service/events/chat_history_persist.go：移除 retry 参数传递
5. 节点层瘦身 + 可见消息逐条持久化
  - agent_nodes.go 大幅简化：Chat/Plan/Execute/Deliver 节点方法移除 ToolSchema 注入、状态摘要渲染等逻辑，只做参数转发和状态落盘
  - 新增 visible_message.go：persistVisibleAssistantMessage 统一处理可见 assistant speak 的实时持久化，失败仅记日志不中断主流程
  - 新增 llm_debug.go：logNodeLLMContext 统一打印 LLM 上下文调试日志
  - graph_run_state.go 新增 PersistVisibleMessageFunc 类型 + AgentGraphDeps.PersistVisibleMessage 字段
  - service/agentsvc/agent_newagent.go 精简主循环，注入 PersistVisibleMessage 回调；agent_history.go 精简历史构建
  - token_budget.go 移除 Execute 专属预算检查，统一到通用预算

前端：
1. 移除 retry 相关 UI 和类型
  - agent.ts 移除 retry_group_id / retry_index / retry_total 字段及 normalize 逻辑
  - AssistantPanel.vue 移除 retry 相关 UI 和交互代码（约 700 行精简）
  - dashboard.ts 移除 retry 相关类型定义
  - AssistantView.vue 微调
2. ContextWindowMeter 压缩次数展示和数值格式优化
  - 新增 formatCompactCount 工具函数，千位以上用 k 单位压缩（如 80k）
  - 新增压缩次数显示
3.修复了新对话发消息时，user和assistant消息被自动调换的bug

仓库：无
This commit is contained in:
Losita
2026-04-17 22:19:38 +08:00
parent d47a8bcabd
commit d8280cc647
39 changed files with 2095 additions and 2386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

271
backend/newAgent/node/agent_nodes.go
View File

@@ -12,12 +12,12 @@ import (
"github.com/LoveLosita/smartflow/backend/newAgent/tools/schedule"
)
// AgentNodes 是 newAgent 通用图的节点容器。
// AgentNodes 负责把 graph 层的节点调用统一转成 node 层真正的执行入口。
//
// 职责边界：
// 1. 负责把 node 层真正实现的方法统一暴露给 graph 注册；
// 2. 负责收口"graph 只编排、node 真执行"的结构约束；
// 3. 负责在每个节点执行成功后统一做状态持久化（Save/Delete）。
// 1. 这里只做参数转发、依赖注入和状态落盘，不承载业务决策。
// 2. 各节点真正的执行逻辑仍在对应的 RunXXXNode 内。
// 3. 节点成功后统一保存快照，方便断线恢复。
type AgentNodes struct{}
// NewAgentNodes 创建通用节点容器。
@@ -25,104 +25,71 @@ func NewAgentNodes() *AgentNodes {
return &AgentNodes{}
}
// Chat 是聊天入口的正式节点方法。
//
// 职责边界：
// 1. 这里只做 graph -> node 的参数转接；
// 2. 真正的入口逻辑仍由 RunChatNode 负责；
// 3. Chat 的 Save 交给 Service 层处理，这里不做持久化。
// Chat 负责把 graph 的 chat 节点请求转给 RunChatNode。
func (n *AgentNodes) Chat(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("chat node: state is nil")
}
// 注入工具 schema 到 ConversationContext，让路由决策更智能。
if st.Deps.ToolRegistry != nil {
schemas := st.Deps.ToolRegistry.Schemas()
toolSchemas := make([]newagentmodel.ToolSchemaContext, len(schemas))
for i, s := range schemas {
toolSchemas[i] = newagentmodel.ToolSchemaContext{
Name: s.Name,
Desc: s.Desc,
SchemaText: s.SchemaText,
}
}
st.EnsureConversationContext().SetToolSchemas(toolSchemas)
}
// 1. Chat 阶段只负责路由与纯对话，不需要看到工具目录，避免能力细节干扰判断。
st.EnsureConversationContext().SetToolSchemas(nil)
if err := RunChatNode(
ctx,
ChatNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
UserInput: st.Request.UserInput,
ConfirmAction: st.Request.ConfirmAction,
Client: st.Deps.ResolveChatClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
},
); err != nil {
if err := RunChatNode(ctx, ChatNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
UserInput: st.Request.UserInput,
ConfirmAction: st.Request.ConfirmAction,
Client: st.Deps.ResolveChatClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
CompactionStore: st.Deps.CompactionStore,
PersistVisibleMessage: st.Deps.PersistVisibleMessage,
}); err != nil {
return nil, err
}
saveAgentState(ctx, st)
return st, nil
}
// Confirm 是确认阶段的正式节点方法。
//
// 职责边界：
// 1. 这里只做 graph -> node 的参数转接；
// 2. 真正的确认逻辑仍由 RunConfirmNode 负责；
// 3. 不需要 LLM Client — 确认内容由已有状态机械格式化。
// 4. Confirm 执行成功后保存状态，因为它创建了 PendingInteraction。
// Confirm 负责把 graph 的 confirm 节点请求转给 RunConfirmNode。
func (n *AgentNodes) Confirm(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("confirm node: state is nil")
}
if err := RunConfirmNode(
ctx,
ConfirmNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
ChunkEmitter: st.EnsureChunkEmitter(),
},
); err != nil {
if err := RunConfirmNode(ctx, ConfirmNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
ChunkEmitter: st.EnsureChunkEmitter(),
}); err != nil {
return nil, err
} else if st.Deps.WriteSchedulePreview != nil && st.ScheduleState == nil {
flowState := st.EnsureFlowState()
log.Printf("[WARN] deliver: schedule state is nil, skip preview write chat=%s", flowState.ConversationID)
}
saveAgentState(ctx, st)
return st, nil
}
// Plan 是规划阶段的正式节点方法。
//
// 职责边界：
// 1. 这里只做 graph -> node 的参数转接；
// 2. 真正的单轮规划逻辑仍由 RunPlanNode 负责；
// 3. Plan 执行成功后保存状态，支持意外断线恢复。
// Plan 负责把 graph 的 plan 节点请求转给 RunPlanNode。
func (n *AgentNodes) Plan(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("plan node: state is nil")
}
// 等待后台记忆检索完成，注入最新记忆后再启动 Plan。
// 等待后端记忆检索完成，再把最新结果注入上下文。
ensureFreshMemory(st)
if err := RunPlanNode(
ctx,
PlanNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
UserInput: st.Request.UserInput,
Client: st.Deps.ResolvePlanClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
ResumeNode: "plan",
AlwaysExecute: st.Request.AlwaysExecute,
ThinkingEnabled: st.Deps.ThinkingPlan,
},
); err != nil {
if err := RunPlanNode(ctx, PlanNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
UserInput: st.Request.UserInput,
Client: st.Deps.ResolvePlanClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
ResumeNode: "plan",
AlwaysExecute: st.Request.AlwaysExecute,
ThinkingEnabled: st.Deps.ThinkingPlan,
CompactionStore: st.Deps.CompactionStore,
PersistVisibleMessage: st.Deps.PersistVisibleMessage,
}); err != nil {
return nil, err
}
@@ -130,12 +97,7 @@ func (n *AgentNodes) Plan(ctx context.Context, st *newagentmodel.AgentGraphState
return st, nil
}
// RoughBuild 是粗排阶段的正式节点方法。
//
// 职责边界：
// 1. 调用注入的 RoughBuildFunc 执行粗排算法；
// 2. 把粗排结果写入 ScheduleState；
// 3. 完成后保存状态，支持意外断线恢复。
// RoughBuild 负责把 graph 的 rough_build 节点请求转给 RunRoughBuildNode。
func (n *AgentNodes) RoughBuild(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("rough_build node: state is nil")
@@ -149,48 +111,31 @@ func (n *AgentNodes) RoughBuild(ctx context.Context, st *newagentmodel.AgentGrap
return st, nil
}
// Interrupt 是中断阶段的正式节点方法。
//
// 职责边界：
// 1. 这里只做 graph -> node 的参数转接；
// 2. 真正的中断逻辑仍由 RunInterruptNode 负责；
// 3. 不需要 LLM Client — 所有文本已在 PendingInteraction 里。
// 4. 不需要 Save — 上游节点（Plan/Execute/Confirm）已经存过了。
// Interrupt 负责把 graph 的 interrupt 节点请求转给 RunInterruptNode。
func (n *AgentNodes) Interrupt(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("interrupt node: state is nil")
}
if err := RunInterruptNode(
ctx,
InterruptNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
ChunkEmitter: st.EnsureChunkEmitter(),
},
); err != nil {
if err := RunInterruptNode(ctx, InterruptNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
ChunkEmitter: st.EnsureChunkEmitter(),
PersistVisibleMessage: st.Deps.PersistVisibleMessage,
}); err != nil {
return nil, err
}
return st, nil
}
// Execute 是执行阶段的正式节点方法。
//
// 职责边界：
// 1. 这里只做 graph -> node 的参数转接；
// 2. 真正的单轮执行逻辑仍由 RunExecuteNode 负责。
//
// 设计原则：
// 1. LLM 主导：LLM 自己判断 done_when 是否满足，自己决定何时推进/完成；
// 2. 后端兜底：只做资源控制、安全兜底、证据记录；
// 3. 不做硬校验：后端不质疑 LLM 的 advance/complete 决策。
// 4. Execute 每轮执行成功后保存状态，支持意外断线恢复。
// Execute 负责把 graph 的 execute 节点请求转给 RunExecuteNode。
func (n *AgentNodes) Execute(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("execute node: state is nil")
}
// 按需加载 ScheduleState（首次执行时从 DB 加载，后续复用内存中的 state）。
// 1. 首次进入时按需加载日程状态，后续轮次复用内存状态。
var scheduleState *schedule.ScheduleState
if ss, loadErr := st.EnsureScheduleState(ctx); loadErr != nil {
return nil, fmt.Errorf("execute node: 加载日程状态失败: %w", loadErr)
@@ -198,7 +143,7 @@ func (n *AgentNodes) Execute(ctx context.Context, st *newagentmodel.AgentGraphSt
scheduleState = ss
}
// 注入工具 schema 到 ConversationContext，让 LLM 能看到可用工具列表。
// 2. 把工具 schema 注入上下文，供 LLM 看到真实工具边界。
if st.Deps.ToolRegistry != nil {
schemas := st.Deps.ToolRegistry.Schemas()
toolSchemas := make([]newagentmodel.ToolSchemaContext, len(schemas))
@@ -212,28 +157,26 @@ func (n *AgentNodes) Execute(ctx context.Context, st *newagentmodel.AgentGraphSt
st.EnsureConversationContext().SetToolSchemas(toolSchemas)
}
// 等待后台记忆检索完成，注入最新记忆后再启动 Execute。
// 3. 等待后端记忆检索结果，再把最新结果注入上下文。
ensureFreshMemory(st)
if err := RunExecuteNode(
ctx,
ExecuteNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
UserInput: st.Request.UserInput,
Client: st.Deps.ResolveExecuteClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
ResumeNode: "execute",
ToolRegistry: st.Deps.ToolRegistry,
ScheduleState: scheduleState,
SchedulePersistor: st.Deps.SchedulePersistor,
CompactionStore: st.Deps.CompactionStore,
WriteSchedulePreview: st.Deps.WriteSchedulePreview,
OriginalScheduleState: st.OriginalScheduleState,
AlwaysExecute: st.Request.AlwaysExecute,
ThinkingEnabled: st.Deps.ThinkingExecute,
},
); err != nil {
if err := RunExecuteNode(ctx, ExecuteNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
UserInput: st.Request.UserInput,
Client: st.Deps.ResolveExecuteClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
ResumeNode: "execute",
ToolRegistry: st.Deps.ToolRegistry,
ScheduleState: scheduleState,
SchedulePersistor: st.Deps.SchedulePersistor,
CompactionStore: st.Deps.CompactionStore,
WriteSchedulePreview: st.Deps.WriteSchedulePreview,
OriginalScheduleState: st.OriginalScheduleState,
AlwaysExecute: st.Request.AlwaysExecute,
ThinkingEnabled: st.Deps.ThinkingExecute,
PersistVisibleMessage: st.Deps.PersistVisibleMessage,
}); err != nil {
return nil, err
}
@@ -241,12 +184,7 @@ func (n *AgentNodes) Execute(ctx context.Context, st *newagentmodel.AgentGraphSt
return st, nil
}
// OrderGuard 是顺序守卫阶段的正式节点方法。
//
// 职责边界：
// 1. 只负责调用 RunOrderGuardNode 做 suggested 相对顺序校验；
// 2. 不负责交付文案生成，校验结果统一交给 Deliver 节点收口；
// 3. 节点执行后保存状态，保证异常中断后仍可复盘守卫结果。
// OrderGuard 负责把 graph 的 order_guard 节点请求转给 RunOrderGuardNode。
func (n *AgentNodes) OrderGuard(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("order_guard node: state is nil")
@@ -260,38 +198,32 @@ func (n *AgentNodes) OrderGuard(ctx context.Context, st *newagentmodel.AgentGrap
return st, nil
}
// Deliver 是交付阶段的正式节点方法。
//
// 职责边界：
// 1. 这里只做 graph -> node 的参数转接；
// 2. 真正的交付逻辑仍由 RunDeliverNode 负责；
// 3. 调 LLM 生成任务总结，失败时降级到机械格式化。
// 4. 任务完成后保存最终状态到 Redis（2h TTL），支持断线恢复和 MySQL outbox 异步持久化。
// Deliver 负责把 graph 的 deliver 节点请求转给 RunDeliverNode。
func (n *AgentNodes) Deliver(ctx context.Context, st *newagentmodel.AgentGraphState) (*newagentmodel.AgentGraphState, error) {
if st == nil {
return nil, errors.New("deliver node: state is nil")
}
if err := RunDeliverNode(
ctx,
DeliverNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
Client: st.Deps.ResolveDeliverClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
ThinkingEnabled: st.Deps.ThinkingDeliver,
},
); err != nil {
// 1. Deliver 只做最终收口总结，不需要工具目录，避免无关能力信息污染总结。
st.EnsureConversationContext().SetToolSchemas(nil)
if err := RunDeliverNode(ctx, DeliverNodeInput{
RuntimeState: st.EnsureRuntimeState(),
ConversationContext: st.EnsureConversationContext(),
Client: st.Deps.ResolveDeliverClient(),
ChunkEmitter: st.EnsureChunkEmitter(),
ThinkingEnabled: st.Deps.ThinkingDeliver,
CompactionStore: st.Deps.CompactionStore,
PersistVisibleMessage: st.Deps.PersistVisibleMessage,
}); err != nil {
return nil, err
}
// 任务完成后写排程预览缓存：只有走到 Deliver 才代表排程结果已稳定，
// 中断（confirm/ask_user）路径不写，避免把中间态暴露给前端。
// 只有真正完成时才写入排程预览，避免中间态污染前端展示。
if st.Deps.WriteSchedulePreview != nil && st.ScheduleState != nil {
flowState := st.EnsureFlowState()
if flowState != nil && flowState.IsCompleted() {
if err := st.Deps.WriteSchedulePreview(ctx, st.ScheduleState, flowState.UserID, flowState.ConversationID, flowState.TaskClassIDs); err != nil {
// 写缓存失败不阻断主流程，降级为仅 log。
log.Printf("[WARN] deliver: 写入排程预览缓存失败 chat=%s: %v", flowState.ConversationID, err)
}
} else if flowState != nil {
@@ -303,19 +235,16 @@ func (n *AgentNodes) Deliver(ctx context.Context, st *newagentmodel.AgentGraphSt
return st, nil
}
// --- 记忆预取消费辅助 ---
// ensureFreshMemory 等待后台记忆检索完成，将最新结果注入 ConversationContext。
// ensureFreshMemory 等待后端记忆检索完成，并把最新结果写入 ConversationContext。
//
// 设计说明：
// 1. 只在首次调用时等待 channel（最多 500ms），后续调用直接跳过；
// 2. 覆盖 ConversationContext 中已有的缓存记忆（UpsertPinnedBlock 按 key 覆盖）；
// 3. timeout 后保留缓存记忆不替换，保证 Execute ReAct 循环不会因超时丢失记忆。
// 1. 只在首次调用时等待 channel，后续调用直接跳过。
// 2. 超时后保留原有上下文，不额外覆盖。
// 3. 记忆为空时也不做额外写入，避免污染 prompt。
func ensureFreshMemory(st *newagentmodel.AgentGraphState) {
if st == nil || st.Deps.MemoryConsumed || st.Deps.MemoryFuture == nil {
return
}
st.Deps.MemoryConsumed = true // 标记已消费，后续调用直接跳过
st.Deps.MemoryConsumed = true
select {
case content := <-st.Deps.MemoryFuture:
@@ -327,20 +256,11 @@ func ensureFreshMemory(st *newagentmodel.AgentGraphState) {
})
}
case <-time.After(newagentmodel.MemoryFreshTimeout):
// timeout：保留 ConversationContext 中已有的缓存记忆，不做额外操作
// 超时后保留原有上下文即可。
}
}
// --- 持久化辅助 ---
// saveAgentState 在节点执行成功后，将当前运行态快照保存到 Redis。
//
// 设计原则：
// 1. Save 失败只记日志，不中断 Graph 流程；
// 2. StateStore 为空时静默跳过（骨架期 / 测试环境）；
// 3. conversationID 为空时也静默跳过，避免写入无效 key。
//
// TODO: 接入项目统一的日志框架后，把 _ = err 改成结构化日志。
// saveAgentState 在节点成功执行后保存运行快照。
func saveAgentState(ctx context.Context, st *newagentmodel.AgentGraphState) {
if st == nil {
return
@@ -370,14 +290,7 @@ func saveAgentState(ctx context.Context, st *newagentmodel.AgentGraphState) {
_ = store.Save(ctx, flowState.ConversationID, snapshot)
}
// deleteAgentState 在任务完成后，删除 Redis 中的运行态快照。
//
// 设计原则：
// 1. Delete 失败只记日志，不中断 Graph 流程；
// 2. 删除是幂等的，key 不存在也视为成功；
// 3. StateStore 为空时静默跳过。
//
// TODO: 接入项目统一的日志框架后，把 _ = err 改成结构化日志。
// deleteAgentState 在任务完成后删除运行快照。
func deleteAgentState(ctx context.Context, st *newagentmodel.AgentGraphState) {
if st == nil {
return

57
backend/newAgent/node/chat.go
View File

@@ -45,12 +45,14 @@ const (
// 3. ConversationContext 提供历史对话；
// 4. ConfirmAction 仅在 confirm 恢复场景下由前端传入 "accept" / "reject"。
type ChatNodeInput struct {
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
UserInput string
ConfirmAction string
Client *infrallm.Client
ChunkEmitter *newagentstream.ChunkEmitter
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
UserInput string
ConfirmAction string
Client *infrallm.Client
ChunkEmitter *newagentstream.ChunkEmitter
CompactionStore newagentmodel.CompactionStore // 上下文压缩持久化
PersistVisibleMessage newagentmodel.PersistVisibleMessageFunc
}
// RunChatNode 执行一轮聊天节点逻辑。
@@ -94,6 +96,15 @@ func RunChatNode(ctx context.Context, input ChatNodeInput) error {
}
nonce := uuid.NewString()
messages := newagentprompt.BuildChatRoutingMessages(conversationContext, input.UserInput, flowState, nonce)
messages = compactUnifiedMessagesIfNeeded(ctx, messages, UnifiedCompactInput{
Client: input.Client,
CompactionStore: input.CompactionStore,
FlowState: flowState,
Emitter: emitter,
StageName: chatStageName,
StatusBlockID: chatStatusBlockID,
})
logNodeLLMContext(chatStageName, "routing", flowState, messages)
reader, err := input.Client.Stream(ctx, messages, infrallm.GenerateOptions{
Temperature: 0.7,
@@ -281,7 +292,7 @@ func handleDirectReplyStream(
if effectiveThinking {
return handleThinkingReplyStream(ctx, reader, input, emitter, conversationContext, flowState)
}
return handleDirectReplyContinueStream(ctx, reader, emitter, conversationContext, flowState, firstVisible)
return handleDirectReplyContinueStream(ctx, reader, input, emitter, conversationContext, flowState, firstVisible)
}
// handleThinkingReplyStream 处理需要思考的回复：关闭路由流 → 第二次 thinking 流式调用。
@@ -295,7 +306,16 @@ func handleThinkingReplyStream(
) error {
_ = reader.Close()
deepMessages := newagentprompt.BuildDeepAnswerMessages(conversationContext, input.UserInput)
deepMessages := newagentprompt.BuildDeepAnswerMessages(flowState, conversationContext, input.UserInput)
deepMessages = compactUnifiedMessagesIfNeeded(ctx, deepMessages, UnifiedCompactInput{
Client: input.Client,
CompactionStore: input.CompactionStore,
FlowState: flowState,
Emitter: emitter,
StageName: chatStageName,
StatusBlockID: chatStatusBlockID,
})
logNodeLLMContext(chatStageName, "direct_reply_thinking", flowState, deepMessages)
deepReader, err := input.Client.Stream(ctx, deepMessages, infrallm.GenerateOptions{
Temperature: 0.5,
MaxTokens: 2000,
@@ -322,6 +342,7 @@ func handleThinkingReplyStream(
deepText = strings.TrimSpace(deepText)
if deepText != "" {
conversationContext.AppendHistory(schema.AssistantMessage(deepText, nil))
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, schema.AssistantMessage(deepText, nil))
}
flowState.Phase = newagentmodel.PhaseChatting
@@ -332,6 +353,7 @@ func handleThinkingReplyStream(
func handleDirectReplyContinueStream(
ctx context.Context,
reader infrallm.StreamReader,
input ChatNodeInput,
emitter *newagentstream.ChunkEmitter,
conversationContext *newagentmodel.ConversationContext,
flowState *newagentmodel.CommonState,
@@ -370,7 +392,9 @@ func handleDirectReplyContinueStream(
text := fullText.String()
if strings.TrimSpace(text) != "" {
conversationContext.AppendHistory(schema.AssistantMessage(text, nil))
msg := schema.AssistantMessage(text, nil)
conversationContext.AppendHistory(msg)
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, msg)
}
flowState.Phase = newagentmodel.PhaseChatting
@@ -568,7 +592,16 @@ func handleDeepAnswerStream(
if effectiveThinking {
thinkingOpt = infrallm.ThinkingModeEnabled
}
deepMessages := newagentprompt.BuildDeepAnswerMessages(conversationContext, input.UserInput)
deepMessages := newagentprompt.BuildDeepAnswerMessages(flowState, conversationContext, input.UserInput)
deepMessages = compactUnifiedMessagesIfNeeded(ctx, deepMessages, UnifiedCompactInput{
Client: input.Client,
CompactionStore: input.CompactionStore,
FlowState: flowState,
Emitter: emitter,
StageName: chatStageName,
StatusBlockID: chatStatusBlockID,
})
logNodeLLMContext(chatStageName, "deep_answer", flowState, deepMessages)
deepReader, err := input.Client.Stream(ctx, deepMessages, infrallm.GenerateOptions{
Temperature: 0.5,
MaxTokens: 2000,
@@ -601,7 +634,9 @@ func handleDeepAnswerStream(
}
// 4. 完整回复写入 history。
conversationContext.AppendHistory(schema.AssistantMessage(deepText, nil))
msg := schema.AssistantMessage(deepText, nil)
conversationContext.AppendHistory(msg)
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, msg)
flowState.Phase = newagentmodel.PhaseChatting
return nil

29
backend/newAgent/node/deliver.go
View File

@@ -28,11 +28,13 @@ const (
// 3. ConversationContext 提供执行阶段的对话历史；
// 4. 交付完成后标记流程结束。
type DeliverNodeInput struct {
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
Client *infrallm.Client
ChunkEmitter *newagentstream.ChunkEmitter
ThinkingEnabled bool // 是否开启 thinking，由 config.yaml 的 agent.thinking.deliver 注入
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
Client *infrallm.Client
ChunkEmitter *newagentstream.ChunkEmitter
ThinkingEnabled bool // 是否开启 thinking，由 config.yaml 的 agent.thinking.deliver 注入
CompactionStore newagentmodel.CompactionStore // 上下文压缩持久化
PersistVisibleMessage newagentmodel.PersistVisibleMessageFunc
}
// RunDeliverNode 执行一轮交付节点逻辑。
@@ -65,10 +67,11 @@ func RunDeliverNode(ctx context.Context, input DeliverNodeInput) error {
}
// 2. 调 LLM 生成交付总结。
summary := generateDeliverSummary(ctx, input.Client, flowState, conversationContext, input.ThinkingEnabled)
summary := generateDeliverSummary(ctx, input.Client, flowState, conversationContext, input.ThinkingEnabled, input.CompactionStore, emitter)
// 3. 伪流式推送总结。
if strings.TrimSpace(summary) != "" {
msg := schema.AssistantMessage(summary, nil)
if err := emitter.EmitPseudoAssistantText(
ctx,
deliverSpeakBlockID,
@@ -78,7 +81,8 @@ func RunDeliverNode(ctx context.Context, input DeliverNodeInput) error {
); err != nil {
return fmt.Errorf("交付总结推送失败: %w", err)
}
conversationContext.AppendHistory(schema.AssistantMessage(summary, nil))
conversationContext.AppendHistory(msg)
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, msg)
}
// 4. 推送最终完成状态。
@@ -100,6 +104,8 @@ func generateDeliverSummary(
flowState *newagentmodel.CommonState,
conversationContext *newagentmodel.ConversationContext,
thinkingEnabled bool,
compactionStore newagentmodel.CompactionStore,
emitter *newagentstream.ChunkEmitter,
) string {
if flowState != nil {
switch {
@@ -115,6 +121,15 @@ func generateDeliverSummary(
}
messages := newagentprompt.BuildDeliverMessages(flowState, conversationContext)
messages = compactUnifiedMessagesIfNeeded(ctx, messages, UnifiedCompactInput{
Client: client,
CompactionStore: compactionStore,
FlowState: flowState,
Emitter: emitter,
StageName: deliverStageName,
StatusBlockID: deliverStatusBlockID,
})
logNodeLLMContext(deliverStageName, "summarizing", flowState, messages)
result, err := client.GenerateText(
ctx,
messages,

100
backend/newAgent/node/execute.go
View File

@@ -60,6 +60,7 @@ type ExecuteNodeInput struct {
OriginalScheduleState *schedule.ScheduleState
AlwaysExecute bool // true 时写工具跳过确认闸门直接执行
ThinkingEnabled bool // 是否开启 thinking，由 config.yaml 的 agent.thinking.execute 注入
PersistVisibleMessage newagentmodel.PersistVisibleMessageFunc
}
// ExecuteRoundObservation 记录执行阶段每轮的关键观察。
@@ -184,19 +185,16 @@ func RunExecuteNode(ctx context.Context, input ExecuteNodeInput) error {
messages := newagentprompt.BuildExecuteMessages(flowState, conversationContext)
// 5.1 Token 预算检查 & 上下文压缩。
messages = compactExecuteMessagesIfNeeded(
ctx, messages, input, flowState, emitter,
)
messages = compactUnifiedMessagesIfNeeded(ctx, messages, UnifiedCompactInput{
Client: input.Client,
CompactionStore: input.CompactionStore,
FlowState: flowState,
Emitter: emitter,
StageName: executeStageName,
StatusBlockID: executeStatusBlockID,
})
log.Printf(
"[DEBUG] execute LLM context begin chat=%s round=%d message_count=%d\n%s\n[DEBUG] execute LLM context end chat=%s round=%d",
flowState.ConversationID,
flowState.RoundUsed,
len(messages),
formatExecuteLLMMessagesForDebug(messages),
flowState.ConversationID,
flowState.RoundUsed,
)
logNodeLLMContext(executeStageName, "decision", flowState, messages)
decision, rawResult, err := infrallm.GenerateJSON[newagentmodel.ExecuteDecision](
ctx,
input.Client,
@@ -337,6 +335,7 @@ func RunExecuteNode(ctx context.Context, input ExecuteNodeInput) error {
if !isConfirmWithCard && !isAskUser && !isAbort {
// 推流给前端
msg := schema.AssistantMessage(speakText, nil)
if err := emitter.EmitPseudoAssistantText(
ctx,
executeSpeakBlockID,
@@ -346,6 +345,7 @@ func RunExecuteNode(ctx context.Context, input ExecuteNodeInput) error {
); err != nil {
return fmt.Errorf("执行文案推送失败: %w", err)
}
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, msg)
}
// 1. confirm / ask_user 的 speak 仍要写入历史，避免下一轮 LLM 丢失自己的执行上下文。
// 2. abort 不在这里写历史，避免先输出中间 speak，再在 deliver 收到第二份终止文案。
@@ -1674,79 +1674,3 @@ func flattenForLog(text string) string {
text = strings.ReplaceAll(text, "\r", " ")
return strings.TrimSpace(text)
}
// formatExecuteLLMMessagesForDebug 将本轮送入 LLM 的完整消息上下文展开成可读多行日志。
//
// 说明：
// 1. 按消息索引逐条输出，便于和上游上下文构造步骤逐项对齐；
// 2. 完整输出 content / reasoning_content / tool_calls / extra，不做截断；
// 3. 仅用于调试打点，不参与业务决策。
func formatExecuteLLMMessagesForDebug(messages []*schema.Message) string {
if len(messages) == 0 {
return "(empty messages)"
}
var sb strings.Builder
for i, msg := range messages {
sb.WriteString(fmt.Sprintf("----- message[%d] -----\n", i))
if msg == nil {
sb.WriteString("role: <nil>\n\n")
continue
}
sb.WriteString(fmt.Sprintf("role: %s\n", msg.Role))
if strings.TrimSpace(msg.ToolCallID) != "" {
sb.WriteString(fmt.Sprintf("tool_call_id: %s\n", msg.ToolCallID))
}
if strings.TrimSpace(msg.ToolName) != "" {
sb.WriteString(fmt.Sprintf("tool_name: %s\n", msg.ToolName))
}
if len(msg.ToolCalls) > 0 {
sb.WriteString("tool_calls:\n")
for j, call := range msg.ToolCalls {
sb.WriteString(fmt.Sprintf(" - [%d] id=%s type=%s function=%s\n", j, call.ID, call.Type, call.Function.Name))
sb.WriteString(" arguments:\n")
sb.WriteString(indentMultilineForDebug(call.Function.Arguments, " "))
sb.WriteString("\n")
}
}
if strings.TrimSpace(msg.ReasoningContent) != "" {
sb.WriteString("reasoning_content:\n")
sb.WriteString(indentMultilineForDebug(msg.ReasoningContent, " "))
sb.WriteString("\n")
}
sb.WriteString("content:\n")
sb.WriteString(indentMultilineForDebug(msg.Content, " "))
sb.WriteString("\n")
if len(msg.Extra) > 0 {
sb.WriteString("extra:\n")
raw, err := json.MarshalIndent(msg.Extra, "", " ")
if err != nil {
sb.WriteString(indentMultilineForDebug("<marshal_error>", " "))
} else {
sb.WriteString(indentMultilineForDebug(string(raw), " "))
}
sb.WriteString("\n")
}
sb.WriteString("\n")
}
return sb.String()
}
// indentMultilineForDebug 为多行文本统一添加前缀缩进，避免日志折行后难以阅读。
func indentMultilineForDebug(text, prefix string) string {
if text == "" {
return prefix + "<empty>"
}
lines := strings.Split(text, "\n")
for i := range lines {
lines[i] = prefix + lines[i]
}
return strings.Join(lines, "\n")
}

197
backend/newAgent/node/execute_compact.go
View File

@@ -1,197 +0,0 @@
package newagentnode
import (
"context"
"encoding/json"
"fmt"
"log"
newagentmodel "github.com/LoveLosita/smartflow/backend/newAgent/model"
newagentprompt "github.com/LoveLosita/smartflow/backend/newAgent/prompt"
newagentstream "github.com/LoveLosita/smartflow/backend/newAgent/stream"
"github.com/LoveLosita/smartflow/backend/pkg"
"github.com/cloudwego/eino/schema"
)
// compactExecuteMessagesIfNeeded 检查 Execute prompt 的 token 预算，
// 超限时对 msg1（历史对话）和 msg2（ReAct Loop）执行 LLM 压缩。
//
// 消息布局约定（由 BuildExecuteMessages 返回）：
//
// [0] system — msg0: 系统规则
// [1] assistant — msg1: 历史对话上下文
// [2] assistant — msg2: 当轮 ReAct Loop 记录
// [3] system — msg3: 当前状态 + 用户提示
func compactExecuteMessagesIfNeeded(
ctx context.Context,
messages []*schema.Message,
input ExecuteNodeInput,
flowState *newagentmodel.CommonState,
emitter *newagentstream.ChunkEmitter,
) []*schema.Message {
if len(messages) != 4 {
return messages
}
// 提取四条消息的文本内容
msg0 := messages[0].Content
msg1 := messages[1].Content
msg2 := messages[2].Content
msg3 := messages[3].Content
// Token 预算检查
breakdown, overBudget, needCompactMsg1, needCompactMsg2 := pkg.CheckExecuteTokenBudget(msg0, msg1, msg2, msg3)
log.Printf(
"[COMPACT] token budget check: total=%d budget=%d over=%v compactMsg1=%v compactMsg2=%v (msg0=%d msg1=%d msg2=%d msg3=%d)",
breakdown.Total, breakdown.Budget, overBudget, needCompactMsg1, needCompactMsg2,
breakdown.Msg0, breakdown.Msg1, breakdown.Msg2, breakdown.Msg3,
)
if !overBudget {
// 未超限，记录 token 分布后直接返回
saveTokenStats(ctx, input, flowState, breakdown)
return messages
}
// ---- msg1 压缩 ----
if needCompactMsg1 {
msg1 = compactMsg1IfNeeded(ctx, input, flowState, emitter, msg1)
messages[1].Content = msg1
// 压缩 msg1 后重算预算
breakdown = pkg.EstimateExecuteMessagesTokens(msg0, msg1, msg2, msg3)
}
// ---- msg2 压缩 ----
if needCompactMsg2 || breakdown.Total > pkg.ExecuteTokenBudget {
msg2 = compactMsg2IfNeeded(ctx, input, flowState, emitter, msg2)
messages[2].Content = msg2
breakdown = pkg.EstimateExecuteMessagesTokens(msg0, msg1, msg2, msg3)
}
// 记录最终 token 分布
saveTokenStats(ctx, input, flowState, breakdown)
log.Printf(
"[COMPACT] after compaction: total=%d budget=%d (msg0=%d msg1=%d msg2=%d msg3=%d)",
breakdown.Total, breakdown.Budget,
breakdown.Msg0, breakdown.Msg1, breakdown.Msg2, breakdown.Msg3,
)
return messages
}
// compactMsg1IfNeeded 对 msg1（历史对话）执行 LLM 压缩。
func compactMsg1IfNeeded(
ctx context.Context,
input ExecuteNodeInput,
flowState *newagentmodel.CommonState,
emitter *newagentstream.ChunkEmitter,
msg1 string,
) string {
compactionStore := input.CompactionStore
if compactionStore == nil {
log.Printf("[COMPACT] CompactionStore is nil, skip msg1 compaction")
return msg1
}
// 加载已有压缩摘要
existingSummary, _, err := compactionStore.LoadCompaction(ctx, flowState.UserID, flowState.ConversationID)
if err != nil {
log.Printf("[COMPACT] load existing compaction failed: %v, proceed without cache", err)
}
// SSE: 压缩开始
tokenBefore := pkg.EstimateTextTokens(msg1)
_ = emitter.EmitStatus(
executeStatusBlockID, "compact_msg1", "context_compact_start",
fmt.Sprintf("正在压缩对话历史（%d tokens）...", tokenBefore),
false,
)
// 调用 LLM 压缩
newSummary, err := newagentprompt.CompactMsg1(ctx, input.Client, msg1, existingSummary)
if err != nil {
log.Printf("[COMPACT] compact msg1 failed: %v", err)
_ = emitter.EmitStatus(
executeStatusBlockID, "compact_msg1", "context_compact_done",
"对话历史压缩失败，使用原始文本",
false,
)
return msg1
}
// SSE: 压缩完成
tokenAfter := pkg.EstimateTextTokens(newSummary)
_ = emitter.EmitStatus(
executeStatusBlockID, "compact_msg1", "context_compact_done",
fmt.Sprintf("对话历史已压缩：%d → %d tokens", tokenBefore, tokenAfter),
false,
)
// 持久化压缩结果
if err := compactionStore.SaveCompaction(ctx, flowState.UserID, flowState.ConversationID, newSummary, flowState.RoundUsed); err != nil {
log.Printf("[COMPACT] save compaction failed: %v", err)
}
return newSummary
}
// compactMsg2IfNeeded 对 msg2（ReAct Loop 记录）执行 LLM 压缩。
func compactMsg2IfNeeded(
ctx context.Context,
input ExecuteNodeInput,
flowState *newagentmodel.CommonState,
emitter *newagentstream.ChunkEmitter,
msg2 string,
) string {
// SSE: 压缩开始
tokenBefore := pkg.EstimateTextTokens(msg2)
_ = emitter.EmitStatus(
executeStatusBlockID, "compact_msg2", "context_compact_start",
fmt.Sprintf("正在压缩执行记录（%d tokens）...", tokenBefore),
false,
)
// 调用 LLM 压缩
compressed, err := newagentprompt.CompactMsg2(ctx, input.Client, msg2)
if err != nil {
log.Printf("[COMPACT] compact msg2 failed: %v", err)
_ = emitter.EmitStatus(
executeStatusBlockID, "compact_msg2", "context_compact_done",
"执行记录压缩失败，使用原始文本",
false,
)
return msg2
}
// SSE: 压缩完成
tokenAfter := pkg.EstimateTextTokens(compressed)
_ = emitter.EmitStatus(
executeStatusBlockID, "compact_msg2", "context_compact_done",
fmt.Sprintf("执行记录已压缩：%d → %d tokens", tokenBefore, tokenAfter),
false,
)
return compressed
}
// saveTokenStats 持久化当前 token 分布到 DB。
func saveTokenStats(
ctx context.Context,
input ExecuteNodeInput,
flowState *newagentmodel.CommonState,
breakdown pkg.ExecuteTokenBreakdown,
) {
compactionStore := input.CompactionStore
if compactionStore == nil {
return
}
statsJSON, err := json.Marshal(breakdown)
if err != nil {
log.Printf("[COMPACT] marshal token stats failed: %v", err)
return
}
if err := compactionStore.SaveContextTokenStats(ctx, flowState.UserID, flowState.ConversationID, string(statsJSON)); err != nil {
log.Printf("[COMPACT] save token stats failed: %v", err)
}
}

15
backend/newAgent/node/interrupt.go
View File

@@ -24,9 +24,10 @@ const (
// 2. RuntimeState 提供 PendingInteraction；
// 3. ChunkEmitter 负责推送收尾消息。
type InterruptNodeInput struct {
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
ChunkEmitter *newagentstream.ChunkEmitter
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
ChunkEmitter *newagentstream.ChunkEmitter
PersistVisibleMessage newagentmodel.PersistVisibleMessageFunc
}
// RunInterruptNode 执行一轮中断节点逻辑。
@@ -55,7 +56,7 @@ func RunInterruptNode(ctx context.Context, input InterruptNodeInput) error {
switch pending.Type {
case newagentmodel.PendingInteractionTypeAskUser:
return handleInterruptAskUser(ctx, pending, conversationContext, emitter)
return handleInterruptAskUser(ctx, runtimeState, input.PersistVisibleMessage, pending, conversationContext, emitter)
case newagentmodel.PendingInteractionTypeConfirm:
return handleInterruptConfirm(pending, emitter)
default:
@@ -70,6 +71,8 @@ func RunInterruptNode(ctx context.Context, input InterruptNodeInput) error {
// 写入历史，然后结束。用户体验和正常对话一样 — 助手问了问题，停下来等回复。
func handleInterruptAskUser(
ctx context.Context,
runtimeState *newagentmodel.AgentRuntimeState,
persist newagentmodel.PersistVisibleMessageFunc,
pending *newagentmodel.PendingInteraction,
conversationContext *newagentmodel.ConversationContext,
emitter *newagentstream.ChunkEmitter,
@@ -89,7 +92,9 @@ func handleInterruptAskUser(
}
// 写入对话历史，下一轮 resume 时 LLM 能看到这个上下文。
conversationContext.AppendHistory(schema.AssistantMessage(text, nil))
msg := schema.AssistantMessage(text, nil)
conversationContext.AppendHistory(msg)
persistVisibleAssistantMessage(ctx, persist, runtimeState.EnsureCommonState(), msg)
// 状态持久化已由 agent_nodes 层统一处理，此处不再需要自行存快照。

121
backend/newAgent/node/llm_debug.go Normal file
View File

@@ -0,0 +1,121 @@
package newagentnode
import (
"encoding/json"
"fmt"
"log"
"strings"
newagentmodel "github.com/LoveLosita/smartflow/backend/newAgent/model"
"github.com/cloudwego/eino/schema"
)
// logNodeLLMContext 将某个节点即将送入 LLM 的完整消息上下文按统一格式打印到日志。
//
// 步骤化说明：
// 1. 统一输出 stage / phase / chat / round，方便按一次请求内的多次 LLM 调用串联排查；
// 2. 完整展开 messages，不做截断，保证问题复现时能直接对照 prompt 组装结果；
// 3. 该函数只负责调试日志，不参与任何业务判断，也不修改上下文内容。
func logNodeLLMContext(
stage string,
phase string,
flowState *newagentmodel.CommonState,
messages []*schema.Message,
) {
chatID := ""
roundUsed := 0
if flowState != nil {
chatID = flowState.ConversationID
roundUsed = flowState.RoundUsed
}
log.Printf(
"[DEBUG] %s LLM context begin phase=%s chat=%s round=%d message_count=%d\n%s\n[DEBUG] %s LLM context end phase=%s chat=%s round=%d",
stage,
strings.TrimSpace(phase),
chatID,
roundUsed,
len(messages),
formatLLMMessagesForDebug(messages),
stage,
strings.TrimSpace(phase),
chatID,
roundUsed,
)
}
// formatLLMMessagesForDebug 将本轮送入 LLM 的完整消息上下文展开成可读多行日志。
//
// 说明：
// 1. 按消息索引逐条输出，便于和上游上下文构造步骤逐项对齐；
// 2. 完整输出 content / reasoning_content / tool_calls / extra，不做截断；
// 3. 仅用于调试打点，不参与业务决策。
func formatLLMMessagesForDebug(messages []*schema.Message) string {
if len(messages) == 0 {
return "(empty messages)"
}
var sb strings.Builder
for i, msg := range messages {
sb.WriteString(fmt.Sprintf("----- message[%d] -----\n", i))
if msg == nil {
sb.WriteString("role: <nil>\n\n")
continue
}
sb.WriteString(fmt.Sprintf("role: %s\n", msg.Role))
if strings.TrimSpace(msg.ToolCallID) != "" {
sb.WriteString(fmt.Sprintf("tool_call_id: %s\n", msg.ToolCallID))
}
if strings.TrimSpace(msg.ToolName) != "" {
sb.WriteString(fmt.Sprintf("tool_name: %s\n", msg.ToolName))
}
if len(msg.ToolCalls) > 0 {
sb.WriteString("tool_calls:\n")
for j, call := range msg.ToolCalls {
sb.WriteString(fmt.Sprintf(" - [%d] id=%s type=%s function=%s\n", j, call.ID, call.Type, call.Function.Name))
sb.WriteString(" arguments:\n")
sb.WriteString(indentMultilineForDebug(call.Function.Arguments, " "))
sb.WriteString("\n")
}
}
if strings.TrimSpace(msg.ReasoningContent) != "" {
sb.WriteString("reasoning_content:\n")
sb.WriteString(indentMultilineForDebug(msg.ReasoningContent, " "))
sb.WriteString("\n")
}
sb.WriteString("content:\n")
sb.WriteString(indentMultilineForDebug(msg.Content, " "))
sb.WriteString("\n")
if len(msg.Extra) > 0 {
sb.WriteString("extra:\n")
raw, err := json.MarshalIndent(msg.Extra, "", " ")
if err != nil {
sb.WriteString(indentMultilineForDebug("<marshal_error>", " "))
} else {
sb.WriteString(indentMultilineForDebug(string(raw), " "))
}
sb.WriteString("\n")
}
sb.WriteString("\n")
}
return sb.String()
}
// indentMultilineForDebug 为多行文本统一添加前缀缩进，避免日志折行后难以阅读。
func indentMultilineForDebug(text, prefix string) string {
if text == "" {
return prefix + "<empty>"
}
lines := strings.Split(text, "\n")
for i := range lines {
lines[i] = prefix + lines[i]
}
return strings.Join(lines, "\n")
}

35
backend/newAgent/node/plan.go
View File

@@ -28,14 +28,16 @@ const (
// PlanNodeInput 描述单轮规划节点执行所需的最小依赖。
type PlanNodeInput struct {
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
UserInput string
Client *infrallm.Client
ChunkEmitter *newagentstream.ChunkEmitter
ResumeNode string
AlwaysExecute bool // true 时计划生成后自动确认，不进入 confirm 节点
ThinkingEnabled bool // 是否开启 thinking，由 config.yaml 的 agent.thinking.plan 注入
RuntimeState *newagentmodel.AgentRuntimeState
ConversationContext *newagentmodel.ConversationContext
UserInput string
Client *infrallm.Client
ChunkEmitter *newagentstream.ChunkEmitter
ResumeNode string
AlwaysExecute bool // true 时计划生成后自动确认，不进入 confirm 节点
ThinkingEnabled bool // 是否开启 thinking，由 config.yaml 的 agent.thinking.plan 注入
CompactionStore newagentmodel.CompactionStore // 上下文压缩持久化
PersistVisibleMessage newagentmodel.PersistVisibleMessageFunc
}
// RunPlanNode 执行一轮规划节点逻辑。
@@ -68,6 +70,15 @@ func RunPlanNode(ctx context.Context, input PlanNodeInput) error {
// 2. 构造本轮规划输入。
messages := newagentprompt.BuildPlanMessages(flowState, conversationContext, input.UserInput)
messages = compactUnifiedMessagesIfNeeded(ctx, messages, UnifiedCompactInput{
Client: input.Client,
CompactionStore: input.CompactionStore,
FlowState: flowState,
Emitter: emitter,
StageName: planStageName,
StatusBlockID: planStatusBlockID,
})
logNodeLLMContext(planStageName, "planning", flowState, messages)
// 3. 单轮深度规划：由配置决定是否开启 thinking，不做 token 上限约束。
decision, rawResult, err := infrallm.GenerateJSON[newagentmodel.PlanDecision](
@@ -95,6 +106,7 @@ func RunPlanNode(ctx context.Context, input PlanNodeInput) error {
// 4. 若模型先对用户说了话，且不是 ask_user（ask_user 交给 interrupt 收口），则先以伪流式推送，再写回 history。
if strings.TrimSpace(decision.Speak) != "" && decision.Action != newagentmodel.PlanActionAskUser {
msg := schema.AssistantMessage(decision.Speak, nil)
if err := emitter.EmitPseudoAssistantText(
ctx,
planSpeakBlockID,
@@ -104,7 +116,8 @@ func RunPlanNode(ctx context.Context, input PlanNodeInput) error {
); err != nil {
return fmt.Errorf("规划文案推送失败: %w", err)
}
conversationContext.AppendHistory(schema.AssistantMessage(decision.Speak, nil))
conversationContext.AppendHistory(msg)
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, msg)
}
// 5. 按规划动作推进流程状态。
@@ -139,6 +152,7 @@ func RunPlanNode(ctx context.Context, input PlanNodeInput) error {
// 3. 推流后同步写入历史，确保后续 Execute 阶段的上下文也能看到这份计划。
summary := strings.TrimSpace(buildPlanSummary(decision.PlanSteps))
if summary != "" {
msg := schema.AssistantMessage(summary, nil)
if err := emitter.EmitPseudoAssistantText(
ctx,
planSummaryBlockID,
@@ -148,7 +162,8 @@ func RunPlanNode(ctx context.Context, input PlanNodeInput) error {
); err != nil {
return fmt.Errorf("自动执行前计划摘要推送失败: %w", err)
}
conversationContext.AppendHistory(schema.AssistantMessage(summary, nil))
conversationContext.AppendHistory(msg)
persistVisibleAssistantMessage(ctx, input.PersistVisibleMessage, flowState, msg)
}
flowState.ConfirmPlan()

301
backend/newAgent/node/unified_compact.go Normal file
View File

@@ -0,0 +1,301 @@
package newagentnode
import (
"context"
"encoding/json"
"fmt"
"log"
infrallm "github.com/LoveLosita/smartflow/backend/infra/llm"
newagentmodel "github.com/LoveLosita/smartflow/backend/newAgent/model"
newagentprompt "github.com/LoveLosita/smartflow/backend/newAgent/prompt"
newagentstream "github.com/LoveLosita/smartflow/backend/newAgent/stream"
"github.com/LoveLosita/smartflow/backend/pkg"
"github.com/cloudwego/eino/schema"
)
// UnifiedCompactInput 是统一压缩入口的参数。
//
// 设计说明：
// 1. 从 ExecuteNodeInput 中提取压缩所需的公共字段，消除对 Execute 的直接依赖；
// 2. 各节点（Plan/Chat/Deliver）构造此参数时从自己的 NodeInput 中提取对应字段；
// 3. StageName 和 StatusBlockID 用于区分日志来源和 SSE 状态推送。
type UnifiedCompactInput struct {
// Client 用于调用 LLM 压缩 msg1/msg2。
Client *infrallm.Client
// CompactionStore 用于持久化压缩摘要和 token 统计，为 nil 时跳过持久化。
CompactionStore newagentmodel.CompactionStore
// FlowState 提供 userID / chatID / roundUsed 等定位信息。
FlowState *newagentmodel.CommonState
// Emitter 用于推送压缩进度 SSE 事件。
Emitter *newagentstream.ChunkEmitter
// StageName 标识当前阶段（如 "execute"/"plan"/"chat"/"deliver"），用于日志和缓存 key。
StageName string
// StatusBlockID 是 SSE 状态推送的 block ID，各节点使用自己的 block ID。
StatusBlockID string
}
// compactUnifiedMessagesIfNeeded 检查统一消息结构的 token 预算，
// 超限时对 msg1（历史对话）和 msg2（阶段工作区）执行 LLM 压缩。
//
// 消息布局约定（由 buildUnifiedStageMessages 返回）：
//
// [0] system — msg0: 系统规则 + 工具简表
// [1] assistant — msg1: 历史对话上下文
// [2] assistant — msg2: 阶段工作区（Execute=ReAct Loop，其余="暂无"）
// [3] system — msg3: 阶段状态 + 记忆 + 指令
//
// 压缩策略：
// 1. msg1 超过可用预算一半时触发 LLM 压缩（合并已有摘要 + 新内容）；
// 2. msg1 压缩后仍超限，则对 msg2 也做 LLM 压缩；
// 3. 压缩结果持久化到 CompactionStore，下一轮可复用摘要避免重复计算。
func compactUnifiedMessagesIfNeeded(
ctx context.Context,
messages []*schema.Message,
input UnifiedCompactInput,
) []*schema.Message {
if input.FlowState == nil {
log.Printf("[COMPACT:%s] FlowState is nil, skip token stats refresh", input.StageName)
return messages
}
// 1. 非严格 4 段式时，退化成按角色汇总的统计，确保 context_token_stats 仍然刷新。
if len(messages) != 4 {
breakdown := estimateFallbackStageTokenBreakdown(messages)
log.Printf(
"[COMPACT:%s] fallback token stats refresh: total=%d budget=%d count=%d (msg0=%d msg1=%d msg2=%d msg3=%d)",
input.StageName, breakdown.Total, breakdown.Budget, len(messages),
breakdown.Msg0, breakdown.Msg1, breakdown.Msg2, breakdown.Msg3,
)
saveUnifiedTokenStats(ctx, input, breakdown)
return messages
}
// 2. 提取四条消息的文本内容。
msg0 := messages[0].Content
msg1 := messages[1].Content
msg2 := messages[2].Content
msg3 := messages[3].Content
// 3. Token 预算检查。
breakdown, overBudget, needCompactMsg1, needCompactMsg2 := pkg.CheckStageTokenBudget(msg0, msg1, msg2, msg3)
log.Printf(
"[COMPACT:%s] token budget check: total=%d budget=%d over=%v compactMsg1=%v compactMsg2=%v (msg0=%d msg1=%d msg2=%d msg3=%d)",
input.StageName, breakdown.Total, breakdown.Budget, overBudget, needCompactMsg1, needCompactMsg2,
breakdown.Msg0, breakdown.Msg1, breakdown.Msg2, breakdown.Msg3,
)
if !overBudget {
// 4. 未超限，记录 token 分布后直接返回。
saveUnifiedTokenStats(ctx, input, breakdown)
return messages
}
// 5. msg1 压缩（历史对话 → LLM 摘要）。
if needCompactMsg1 {
msg1 = compactUnifiedMsg1(ctx, input, msg1)
messages[1].Content = msg1
// 压缩 msg1 后重算预算。
breakdown = pkg.EstimateStageMessagesTokens(msg0, msg1, msg2, msg3)
}
// 6. msg2 压缩（阶段工作区 → LLM 摘要）。
if needCompactMsg2 || breakdown.Total > pkg.StageTokenBudget {
msg2 = compactUnifiedMsg2(ctx, input, msg2)
messages[2].Content = msg2
breakdown = pkg.EstimateStageMessagesTokens(msg0, msg1, msg2, msg3)
}
// 7. 记录最终 token 分布。
saveUnifiedTokenStats(ctx, input, breakdown)
log.Printf(
"[COMPACT:%s] after compaction: total=%d budget=%d (msg0=%d msg1=%d msg2=%d msg3=%d)",
input.StageName, breakdown.Total, breakdown.Budget,
breakdown.Msg0, breakdown.Msg1, breakdown.Msg2, breakdown.Msg3,
)
return messages
}
// estimateFallbackStageTokenBreakdown 在非统一 4 段式场景下按消息角色做近似统计。
//
// 步骤说明：
// 1. 先按消息类型汇总 token，保证总量准确；
// 2. 再把最后一个 user 消息尽量视作 msg3，保留阶段指令语义；
// 3. 其他历史内容归入 msg1 / msg2，确保上下文统计不会因为结构不标准而断更。
func estimateFallbackStageTokenBreakdown(messages []*schema.Message) pkg.StageTokenBreakdown {
breakdown := pkg.StageTokenBreakdown{Budget: pkg.StageTokenBudget}
if len(messages) == 0 {
return breakdown
}
lastUserIndex := -1
for i := len(messages) - 1; i >= 0; i-- {
msg := messages[i]
if msg == nil {
continue
}
if msg.Role == schema.User {
lastUserIndex = i
break
}
}
for i, msg := range messages {
if msg == nil {
continue
}
tokens := pkg.EstimateMessageTokens(msg)
breakdown.Total += tokens
switch msg.Role {
case schema.System:
breakdown.Msg0 += tokens
case schema.User:
if i == lastUserIndex {
breakdown.Msg3 += tokens
} else {
breakdown.Msg1 += tokens
}
case schema.Tool:
breakdown.Msg2 += tokens
case schema.Assistant:
if len(msg.ToolCalls) > 0 {
breakdown.Msg2 += tokens
} else {
breakdown.Msg1 += tokens
}
default:
breakdown.Msg1 += tokens
}
}
return breakdown
}
// compactUnifiedMsg1 对 msg1（历史对话）执行 LLM 压缩。
//
// 步骤化说明：
// 1. CompactionStore 为 nil 时跳过（测试环境 / 骨架期）；
// 2. 先加载该阶段已有的压缩摘要，与当前 msg1 合并后调 LLM 压缩；
// 3. 压缩失败时降级为原始文本，不中断主流程；
// 4. 压缩成功后持久化新摘要，供下一轮复用。
func compactUnifiedMsg1(
ctx context.Context,
input UnifiedCompactInput,
msg1 string,
) string {
// 1. CompactionStore 为 nil 时无法加载/保存摘要，跳过压缩。
if input.CompactionStore == nil {
log.Printf("[COMPACT:%s] CompactionStore is nil, skip msg1 compaction", input.StageName)
return msg1
}
// 2. 加载该阶段已有的压缩摘要（可能为空）。
existingSummary, _, err := input.CompactionStore.LoadStageCompaction(ctx, input.FlowState.UserID, input.FlowState.ConversationID, input.StageName)
if err != nil {
log.Printf("[COMPACT:%s] load existing compaction failed: %v, proceed without cache", input.StageName, err)
}
// 3. SSE: 压缩开始。
tokenBefore := pkg.EstimateTextTokens(msg1)
_ = input.Emitter.EmitStatus(
input.StatusBlockID, input.StageName, "context_compact_start",
fmt.Sprintf("正在压缩对话历史（%d tokens）...", tokenBefore),
false,
)
// 4. 调用 LLM 压缩：将 msg1 全文 + 已有摘要合并为一份紧凑摘要。
newSummary, err := newagentprompt.CompactMsg1(ctx, input.Client, msg1, existingSummary)
if err != nil {
log.Printf("[COMPACT:%s] compact msg1 failed: %v", input.StageName, err)
_ = input.Emitter.EmitStatus(
input.StatusBlockID, input.StageName, "context_compact_done",
"对话历史压缩失败，使用原始文本",
false,
)
return msg1
}
// 5. SSE: 压缩完成。
tokenAfter := pkg.EstimateTextTokens(newSummary)
_ = input.Emitter.EmitStatus(
input.StatusBlockID, input.StageName, "context_compact_done",
fmt.Sprintf("对话历史已压缩：%d → %d tokens", tokenBefore, tokenAfter),
false,
)
// 6. 持久化压缩结果，下一轮可直接复用摘要。
if err := input.CompactionStore.SaveStageCompaction(ctx, input.FlowState.UserID, input.FlowState.ConversationID, input.StageName, newSummary, input.FlowState.RoundUsed); err != nil {
log.Printf("[COMPACT:%s] save compaction failed: %v", input.StageName, err)
}
return newSummary
}
// compactUnifiedMsg2 对 msg2（阶段工作区）执行 LLM 压缩。
//
// 步骤化说明：
// 1. 非 Execute 阶段的 msg2 通常是"暂无"，压缩无意义但不会出错；
// 2. Execute 阶段的 msg2 包含 ReAct loop 记录，压缩可显著节省 token；
// 3. 压缩失败时降级为原始文本，不中断主流程。
func compactUnifiedMsg2(
ctx context.Context,
input UnifiedCompactInput,
msg2 string,
) string {
// 1. SSE: 压缩开始。
tokenBefore := pkg.EstimateTextTokens(msg2)
_ = input.Emitter.EmitStatus(
input.StatusBlockID, input.StageName, "context_compact_start",
fmt.Sprintf("正在压缩执行记录（%d tokens）...", tokenBefore),
false,
)
// 2. 调用 LLM 压缩。
compressed, err := newagentprompt.CompactMsg2(ctx, input.Client, msg2)
if err != nil {
log.Printf("[COMPACT:%s] compact msg2 failed: %v", input.StageName, err)
_ = input.Emitter.EmitStatus(
input.StatusBlockID, input.StageName, "context_compact_done",
"执行记录压缩失败，使用原始文本",
false,
)
return msg2
}
// 3. SSE: 压缩完成。
tokenAfter := pkg.EstimateTextTokens(compressed)
_ = input.Emitter.EmitStatus(
input.StatusBlockID, input.StageName, "context_compact_done",
fmt.Sprintf("执行记录已压缩：%d → %d tokens", tokenBefore, tokenAfter),
false,
)
return compressed
}
// saveUnifiedTokenStats 持久化当前 token 分布到 DB。
//
// 步骤化说明：
// 1. CompactionStore 为 nil 时跳过（测试环境 / 骨架期）；
// 2. 序列化失败只记日志，不中断主流程；
// 3. 写入失败只记日志，不中断主流程。
func saveUnifiedTokenStats(
ctx context.Context,
input UnifiedCompactInput,
breakdown pkg.StageTokenBreakdown,
) {
if input.CompactionStore == nil || input.FlowState == nil {
return
}
statsJSON, err := json.Marshal(breakdown)
if err != nil {
log.Printf("[COMPACT:%s] marshal token stats failed: %v", input.StageName, err)
return
}
if err := input.CompactionStore.SaveContextTokenStats(ctx, input.FlowState.UserID, input.FlowState.ConversationID, string(statsJSON)); err != nil {
log.Printf("[COMPACT:%s] save token stats failed: %v", input.StageName, err)
}
}

37
backend/newAgent/node/visible_message.go Normal file
View File

@@ -0,0 +1,37 @@
package newagentnode
import (
"context"
"log"
"strings"
newagentmodel "github.com/LoveLosita/smartflow/backend/newAgent/model"
"github.com/cloudwego/eino/schema"
)
// persistVisibleAssistantMessage 负责把“真正要展示给用户”的 assistant 文本交给 service 层持久化。
//
// 职责边界：
// 1. 只处理可见的 assistant 消息，不处理内部纠错提示、工具调用结果和纯状态文案；
// 2. 持久化失败只记日志，不反向中断节点主流程，避免“已经对外输出但后端补写失败”时把用户请求打断；
// 3. 具体的 Redis / MySQL / 乐观缓存写入由 service 回调统一完成。
func persistVisibleAssistantMessage(
ctx context.Context,
persist newagentmodel.PersistVisibleMessageFunc,
state *newagentmodel.CommonState,
msg *schema.Message,
) {
if persist == nil || state == nil || msg == nil {
return
}
role := strings.TrimSpace(string(msg.Role))
content := strings.TrimSpace(msg.Content)
if role != string(schema.Assistant) || content == "" {
return
}
if err := persist(ctx, state, msg); err != nil {
log.Printf("[WARN] persist visible assistant message failed chat=%s phase=%s err=%v", state.ConversationID, state.Phase, err)
}
}