feat(memory): P1 长期记忆升级 —— 异步攒批 Consolidate + 软删 + importance/last_seen #1
+3
-2
@@ -186,8 +186,9 @@ RBAC 未做,暂以单管理员账号代理;概览口径必须是**系统级*
|
|||||||
### [~] T4.E 编排引擎边角
|
### [~] T4.E 编排引擎边角
|
||||||
- [x] Map 节点错误传播 + 汇总 ✅ —— writeSection 返回 (body,err) 仅真·LLM 失败计错(预算/无模型是主动降级不算);writeSections 汇总失败数、失败项 Body 带可见标记;mapNode 全失败→置 b.fatalErr(判 failed 非静默 done-空)、部分失败→trace+流式告警。report handleReport 同步提示部分失败。3 单测(全失败/部分精确计数/mapNode 置 fatalErr)。
|
- [x] Map 节点错误传播 + 汇总 ✅ —— writeSection 返回 (body,err) 仅真·LLM 失败计错(预算/无模型是主动降级不算);writeSections 汇总失败数、失败项 Body 带可见标记;mapNode 全失败→置 b.fatalErr(判 failed 非静默 done-空)、部分失败→trace+流式告警。report handleReport 同步提示部分失败。3 单测(全失败/部分精确计数/mapNode 置 fatalErr)。
|
||||||
- [ ] 熔断器接回 failover(harness/circuitbreaker.go 与 llm/failover.go 未接合,可能长卡备用)| M
|
- [ ] 熔断器接回 failover(harness/circuitbreaker.go 与 llm/failover.go 未接合,可能长卡备用)| M
|
||||||
- [ ] Branch else 兜底 + coordinator 专家超时(专家卡死拖垮全局,coordinator.go:109)| M
|
- [x] coordinator 专家超时 ✅ —— specialistTool 加 timeout(默认 specialistTimeout=3min),InvokableRun 包 WithTimeout;超时作为"观察"跳过该专家(err=nil,lead 据其余综合,不中断协调)。2 单测(超时~50ms 跳过 / 正常不受影响)。
|
||||||
- [ ] DSL 拓扑/节点-工具映射校验(dsl/parser.go:23 TODO,现仅 JSON 格式校验)| M
|
- [ ] Branch else 兜底(compose_compiler.go:155 分支无"都不满足"收口)| M
|
||||||
|
- [ ] DSL 拓扑/节点-工具映射校验(gateway dsl/parser.go:23 TODO,现仅 JSON 格式校验)| M
|
||||||
|
|
||||||
### [ ] T4.F 健壮性 / 安全 / 性能收口(多为 S,可穿插着做)
|
### [ ] T4.F 健壮性 / 安全 / 性能收口(多为 S,可穿插着做)
|
||||||
- [ ] 关键 DB 写失败上浮 5xx(现 best-effort 返 200,前端无感,task_handler.go:69 等)| M
|
- [ ] 关键 DB 写失败上浮 5xx(现 best-effort 返 200,前端无感,task_handler.go:69 等)| M
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package eino
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
@@ -73,6 +74,7 @@ type specialistTool struct {
|
|||||||
name string
|
name string
|
||||||
run func(ctx context.Context, brief string) (string, error)
|
run func(ctx context.Context, brief string) (string, error)
|
||||||
tr *execTracer
|
tr *execTracer
|
||||||
|
timeout time.Duration // 单次派发上限;0=不限。超时作为"观察"跳过该专家,不中断协调。
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *specialistTool) Info(context.Context) (*schema.ToolInfo, error) { return s.info, nil }
|
func (s *specialistTool) Info(context.Context) (*schema.ToolInfo, error) { return s.info, nil }
|
||||||
@@ -94,8 +96,19 @@ func (s *specialistTool) InvokableRun(ctx context.Context, argsJSON string, _ ..
|
|||||||
bud.AddPrompt(brief)
|
bud.AddPrompt(brief)
|
||||||
}
|
}
|
||||||
end := s.tr.span("agent:"+s.name, "agent", "派发专家 "+s.name)
|
end := s.tr.span("agent:"+s.name, "agent", "派发专家 "+s.name)
|
||||||
out, err := s.run(ctx, brief)
|
rctx := ctx
|
||||||
|
if s.timeout > 0 {
|
||||||
|
var cancel context.CancelFunc
|
||||||
|
rctx, cancel = context.WithTimeout(ctx, s.timeout)
|
||||||
|
defer cancel()
|
||||||
|
}
|
||||||
|
out, err := s.run(rctx, brief)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
// 超时单独提示:作为"观察"跳过该专家,lead 据其余专家继续综合(不中断协调)。
|
||||||
|
if s.timeout > 0 && errors.Is(err, context.DeadlineExceeded) {
|
||||||
|
end("专家超时", err)
|
||||||
|
return fmt.Sprintf("专家 %s 超时(>%s),本轮跳过其结论。", s.name, s.timeout), nil
|
||||||
|
}
|
||||||
end("专家执行失败", err)
|
end("专家执行失败", err)
|
||||||
return "专家 " + s.name + " 执行失败:" + err.Error(), nil
|
return "专家 " + s.name + " 执行失败:" + err.Error(), nil
|
||||||
}
|
}
|
||||||
@@ -125,7 +138,7 @@ func (o *Orchestrator) buildSpecialists(ctx context.Context, specs []specialistS
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
out = append(out, &specialistTool{
|
out = append(out, &specialistTool{
|
||||||
name: spec.Name, tr: tr, run: run,
|
name: spec.Name, tr: tr, run: run, timeout: specialistTimeout,
|
||||||
info: &schema.ToolInfo{
|
info: &schema.ToolInfo{
|
||||||
Name: spec.Name,
|
Name: spec.Name,
|
||||||
Desc: firstNonEmpty(spec.Use, "专家 "+spec.Name),
|
Desc: firstNonEmpty(spec.Use, "专家 "+spec.Name),
|
||||||
|
|||||||
@@ -92,6 +92,10 @@ const taskExecTimeout = 10 * time.Minute
|
|||||||
// approvalTimeout 是单个审批节点等待人工决定的上限;超时安全默认拒绝(fail-safe)。
|
// approvalTimeout 是单个审批节点等待人工决定的上限;超时安全默认拒绝(fail-safe)。
|
||||||
const approvalTimeout = 5 * time.Minute
|
const approvalTimeout = 5 * time.Minute
|
||||||
|
|
||||||
|
// specialistTimeout 是多智能体单个专家派发的执行上限;超时即作为"观察"跳过该专家
|
||||||
|
// (不中断整个协调),避免单专家卡死拖垮 lead 无限等待。专家可能多轮 react+工具调用,故给较宽。
|
||||||
|
const specialistTimeout = 3 * time.Minute
|
||||||
|
|
||||||
// Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行(记忆召回 → 工具节点 → 注入 → 流式)。
|
// Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行(记忆召回 → 工具节点 → 注入 → 流式)。
|
||||||
type Orchestrator struct {
|
type Orchestrator struct {
|
||||||
pool LLM
|
pool LLM
|
||||||
|
|||||||
@@ -0,0 +1,56 @@
|
|||||||
|
package eino
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/cloudwego/eino/schema"
|
||||||
|
)
|
||||||
|
|
||||||
|
// 专家超时:卡死的专家应在 timeout 附近作为"观察"跳过(err=nil,不中断协调),
|
||||||
|
// 而非无限阻塞 lead。
|
||||||
|
func TestSpecialist_TimeoutSkips(t *testing.T) {
|
||||||
|
o := newOrch(&fakeLLM{}, kbTool(nil), &fakeSink{}, &fakeExec{})
|
||||||
|
st := &specialistTool{
|
||||||
|
name: "分析专家",
|
||||||
|
tr: o.tracer("t1"),
|
||||||
|
timeout: 50 * time.Millisecond,
|
||||||
|
info: &schema.ToolInfo{Name: "分析专家"},
|
||||||
|
run: func(ctx context.Context, _ string) (string, error) {
|
||||||
|
<-ctx.Done() // 模拟卡死:直到超时被取消
|
||||||
|
return "", ctx.Err()
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
out, err := st.InvokableRun(context.Background(), `{"brief":"干活"}`)
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("专家超时应作为观察返回、err=nil(不中断协调),got err=%v", err)
|
||||||
|
}
|
||||||
|
if !strings.Contains(out, "超时") {
|
||||||
|
t.Errorf("应返回超时观察,got %q", out)
|
||||||
|
}
|
||||||
|
if elapsed > 2*time.Second {
|
||||||
|
t.Errorf("应在超时(50ms)附近返回,实际耗时 %v(疑似未生效)", elapsed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 正常专家不受超时影响。
|
||||||
|
func TestSpecialist_NormalWithinTimeout(t *testing.T) {
|
||||||
|
o := newOrch(&fakeLLM{}, kbTool(nil), &fakeSink{}, &fakeExec{})
|
||||||
|
st := &specialistTool{
|
||||||
|
name: "快专家",
|
||||||
|
tr: o.tracer("t1"),
|
||||||
|
timeout: time.Second,
|
||||||
|
info: &schema.ToolInfo{Name: "快专家"},
|
||||||
|
run: func(_ context.Context, brief string) (string, error) { return "结论:" + brief, nil },
|
||||||
|
}
|
||||||
|
out, err := st.InvokableRun(context.Background(), `{"brief":"分析X"}`)
|
||||||
|
if err != nil || !strings.Contains(out, "结论:分析X") {
|
||||||
|
t.Fatalf("正常专家应返回结论,got out=%q err=%v", out, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user