chore: sync upstream v0.3.6 changes

This commit is contained in:
pphua committed 2026-08-28 14:03:07 +08:00
1 parent 9da521ee4e
commit 22132a4d21
34 files changed
+897 -127

No files matched your search

+43
View File
@@ -4,6 +4,49 @@
## [Unreleased] ## [Unreleased]
## [0.3.6] - 2026-08-27
### LLM
#### 修复的问题
- 修复「非流式」开关保存后重新打开又变回「流式」的问题(#69):配置列表接口的 DTO 漏了 `streaming` 字段,响应从不返回它,前端读到 `undefined` 后一律回落成默认流式;实际写入与 DB 存储本无问题,只是读不回来。DTO 补回该字段(不加 `omitempty`,`false` 也必须出现在响应里)。
- 测试连接改为校验模型确有回复、并按该配置真实的收发模式来测(#65):此前只判 HTTP 是否成功,请求通但模型零回复(思考烧光预算 / 正文被安全策略吞掉 / 兼容层丢 `content`)也报「连接成功」,与会话里「不回话」的表现割裂;且测试恒走流式,非流式配置测的其实是另一条通道。现在空回复直接判失败、成功时在提示里展示模型回复,并把当前流式开关一并带入测试,让「测试通过」与「会话跑得通」保持一致。
### Agent
#### 修改的功能
- `list_facts` 改为分页 + 关键词过滤,避免事实多时单次调用撑爆上下文(#74):默认返回最新 20 条,支持 `limit`(上限 100)/ `before` 游标 / `q` 摘要关键词,返回 `{facts, total, has_more, next_before}`;单条 summary 过长按字数截断(全文仍走 `node_detail`)。worker / planner 提示词同步为分页语义。旧库通过一次性迁移把新参数 schema 刷进工具目录表(`SeedTool` 首插入only,否则工具管理页显示「无参数」)。
### 工具
#### 新增的功能
- 「工具执行」页新增工具调用次数统计(#72):工具栏「统计」按钮打开弹窗,按工具展示调用次数、占比与失败次数(次数降序);沿用列表的任务 / 关键词筛选,统计整个结果集而非当前页,弹窗打开时才拉取。
### 依赖
#### 修改的功能
- 升级 norma v0.2.9 → v0.3.2:修复 OpenAI 系响应里被静默丢弃的 reasoning/refusal(Responses 补 `reasoning_text`)、reasoning 字段三名去重、空响应(零内容块)有界重试、压缩边界切断工具配对导致的网关 400(同时修复已烤进 transcript 的存量孤儿)。
### 贡献者
- [@Autumn-27](https://github.com/Autumn-27)
## [0.3.5] - 2026-08-25
### LLM
#### 新增的功能
- 每个 LLM 配置支持流式/非流式切换(默认流式):开走流式 SSE;关走真·非流式(`stream:false`、一次性返回完整 JSON),可绕开部分网关糟糕的 SSE 实现(空帧、思考字段丢帧),代价是失去运行中的实时进度与实时 Token 计数。worker / planner / mainagent / chat / goals 均按当前激活配置动态取值;`llmpool` / `llmrec` / 任务运行时三层 Provider 包装均兼容非流式;`llm_profiles` 新增 `streaming` 列并 `ALTER` 补旧库(默认 `true`,旧配置无感)。
### 贡献者
- [@Autumn-27](https://github.com/Autumn-27)
## [0.3.4] - 2026-08-24 ## [0.3.4] - 2026-08-24
### UI ### UI
+1 -1
View File
@@ -326,7 +326,7 @@ func TestBlackboardToolsReadDirectSources(t *testing.T) {
t.Fatalf("inherited host context missing: %#v", coverage) t.Fatalf("inherited host context missing: %#v", coverage)
} }
facts := callReadJSON(t, tools.listFacts(), `{}`).([]any) facts := callReadJSON(t, tools.listFacts(), `{}`).(map[string]any)["facts"].([]any)
seenSource, seenGrand := false, false seenSource, seenGrand := false, false
for _, raw := range facts { for _, raw := range facts {
item := raw.(map[string]any) item := raw.(map[string]any)
+11
View File
@@ -19,6 +19,17 @@ func (p captureUsageProvider) Stream(ctx context.Context, _ llm.CompletionReques
return func(yield func(llm.StreamEvent, error) bool) { p.stream(ctx, yield) } return func(yield func(llm.StreamEvent, error) bool) { p.stream(ctx, yield) }
} }
func (p captureUsageProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
acc := llm.NewAccumulator()
for ev, err := range p.Stream(ctx, req) {
if err != nil {
return llm.Message{}, "", llm.Usage{}, err
}
acc.Add(ev)
}
return acc.Message(), acc.StopReason, acc.Usage, nil
}
func TestCaptureRunPersistsUsageOnProviderFailure(t *testing.T) { func TestCaptureRunPersistsUsageOnProviderFailure(t *testing.T) {
wantErr := errors.New("provider failed after reporting usage") wantErr := errors.New("provider failed after reporting usage")
provider := captureUsageProvider{stream: func(_ context.Context, yield func(llm.StreamEvent, error) bool) { provider := captureUsageProvider{stream: func(_ context.Context, yield func(llm.StreamEvent, error) bool) {
+26 -18
View File
@@ -6,13 +6,13 @@ import (
"path/filepath" "path/filepath"
"time" "time"
"github.com/Autumn-27/artex/db"
"github.com/Autumn-27/artex/guard"
"github.com/Autumn-27/norma/agentcore" "github.com/Autumn-27/norma/agentcore"
"github.com/Autumn-27/norma/llm" "github.com/Autumn-27/norma/llm"
"github.com/Autumn-27/norma/permission" "github.com/Autumn-27/norma/permission"
actool "github.com/Autumn-27/norma/tool" actool "github.com/Autumn-27/norma/tool"
"github.com/Autumn-27/norma/transcript" "github.com/Autumn-27/norma/transcript"
"github.com/Autumn-27/artex/db"
"github.com/Autumn-27/artex/guard"
) )
// ChatAgent is the generic, task-independent conversational runner behind the chat // ChatAgent is the generic, task-independent conversational runner behind the chat
@@ -22,21 +22,28 @@ import (
// LS/Glob/Grep) plus whatever skills/MCP the key is made visible; NO pentest // LS/Glob/Grep) plus whatever skills/MCP the key is made visible; NO pentest
// graph/task context is injected (that stays exclusive to MainAgent). // graph/task context is injected (that stays exclusive to MainAgent).
type ChatAgent struct { type ChatAgent struct {
prov llm.Provider prov llm.Provider
model string model string
workDir string workDir string
tx *transcript.Store tx *transcript.Store
window int window int
proxyAddr string proxyAddr string
proxyCACert string proxyCACert string
webSearch WebSearchOpts webSearch WebSearchOpts
guard *guard.Guard // optional; nil disables intercept hooks for chat guard *guard.Guard // optional; nil disables intercept hooks for chat
nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming)
} }
func NewChatAgent(prov llm.Provider, model, workDir string, tx *transcript.Store, window int) *ChatAgent { func NewChatAgent(prov llm.Provider, model, workDir string, tx *transcript.Store, window int) *ChatAgent {
return &ChatAgent{prov: prov, model: model, workDir: workDir, tx: tx, window: window} return &ChatAgent{prov: prov, model: model, workDir: workDir, tx: tx, window: window}
} }
// SetNonStreaming wires a resolver deciding whether chat runs use the
// non-streaming model path (true = non-streaming). nil/unset = streaming.
func (c *ChatAgent) SetNonStreaming(fn func() bool) { c.nonStreamingFn = fn }
func (c *ChatAgent) nonStreaming() bool { return c.nonStreamingFn != nil && c.nonStreamingFn() }
// SetProxy points the chat agent's WebFetch/Bash at the recording proxy plus the // SetProxy points the chat agent's WebFetch/Bash at the recording proxy plus the
// CA cert it trusts (empty addr = direct). Kept for parity with the other agents. // CA cert it trusts (empty addr = direct). Kept for parity with the other agents.
func (c *ChatAgent) SetProxy(addr, caCert string) { c.proxyAddr, c.proxyCACert = addr, caCert } func (c *ChatAgent) SetProxy(addr, caCert string) { c.proxyAddr, c.proxyCACert = addr, caCert }
@@ -115,18 +122,19 @@ func (c *ChatAgent) Chat(ctx context.Context, agentKey, sessionID, message strin
BraveSearchAPIKey: ws.BraveKey, BraveSearchAPIKey: ws.BraveKey,
TavilySearchAPIKey: ws.TavilyKey, TavilySearchAPIKey: ws.TavilyKey,
WebSearchProxy: ws.Proxy, WebSearchProxy: ws.Proxy,
BashEnv: proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA BashEnv: proxyEnv(c.proxyAddr, c.proxyCACert), // Bash 子命令默认走代理+信任 CA
WorkingDir: sessionWorkDir, WorkingDir: sessionWorkDir,
MaxTurns: maxTurns, MaxTurns: maxTurns,
MaxDuration: maxDuration, MaxDuration: maxDuration,
Compaction: compactionConfig(c.window), Compaction: compactionConfig(c.window),
Todos: actool.NewTodoStore(), Todos: actool.NewTodoStore(),
// large tool output spills to cmd-output/ under the session dir. // large tool output spills to cmd-output/ under the session dir.
// 截断上限用 SDK 默认(tool.Capture 的 30000 字符)。 // 截断上限用 SDK 默认(tool.Capture 的 30000 字符)。
ToolOutputDir: filepath.Join(sessionWorkDir, "cmd-output"), ToolOutputDir: filepath.Join(sessionWorkDir, "cmd-output"),
// 命中预算(步数)→ SDK 跑收尾:输出一句总结。Prompt 与收尾轮数按本 agent key 后台可编辑 // 命中预算(步数)→ SDK 跑收尾:输出一句总结。Prompt 与收尾轮数按本 agent key 后台可编辑
// (自定义 agent 各自一份;留空/0 用通用默认:10 轮)。 // (自定义 agent 各自一份;留空/0 用通用默认:10 轮)。
Settlement: wrapupSettlement(agentKey, nil), Settlement: wrapupSettlement(agentKey, nil),
NonStreaming: c.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete
} }
if c.guard != nil { if c.guard != nil {
opts.Hooks = c.guard.Hooks() opts.Hooks = c.guard.Hooks()
+4 -3
View File
@@ -98,13 +98,13 @@ func DecomposeGoals(ctx context.Context, prov llm.Provider, dataDir, goalText, d
if prov == nil { if prov == nil {
return nil return nil
} }
return DecomposeGoalsWithProvider(ctx, prov, dataDir, goalText, desc, as, ts, taskID, emit) return DecomposeGoalsWithProvider(ctx, prov, dataDir, goalText, desc, as, ts, taskID, false, emit)
} }
// DecomposeGoalsWithProvider is the task-runtime variant used when a task has an // DecomposeGoalsWithProvider is the task-runtime variant used when a task has an
// ordered provider chain. It preserves the same tools and write behavior while // ordered provider chain. It preserves the same tools and write behavior while
// letting the caller own provider selection/failover. // letting the caller own provider selection/failover.
func DecomposeGoalsWithProvider(ctx context.Context, prov llm.Provider, dataDir, goalText, desc string, as *db.AssetStore, ts *db.ExplorationStore, taskID int64, emit func(db.Activity)) []GoalSpec { func DecomposeGoalsWithProvider(ctx context.Context, prov llm.Provider, dataDir, goalText, desc string, as *db.AssetStore, ts *db.ExplorationStore, taskID int64, nonStreaming bool, emit func(db.Activity)) []GoalSpec {
if prov == nil { if prov == nil {
return nil return nil
} }
@@ -145,7 +145,8 @@ func DecomposeGoalsWithProvider(ctx context.Context, prov llm.Provider, dataDir,
PermissionMode: acperm.ModeBypass, PermissionMode: acperm.ModeBypass,
DisableBackgroundTasks: true, DisableBackgroundTasks: true,
// 3 步(抽约束 → 登记范围 → 拆目标)各需一次工具调用,给足回合避免收尾前漏调 set_goals。 // 3 步(抽约束 → 登记范围 → 拆目标)各需一次工具调用,给足回合避免收尾前漏调 set_goals。
MaxTurns: 8, MaxTurns: 8,
NonStreaming: nonStreaming, // 该 profile 选非流式时走 Provider.Complete
}, userMsg, captureEmit) }, userMsg, captureEmit)
// set_goals persisted the goals directly; read them back so the caller sees what // set_goals persisted the goals directly; read them back so the caller sees what
// was written (empty slice ⇒ the LLM produced nothing ⇒ caller falls back). // was written (empty slice ⇒ the LLM produced nothing ⇒ caller falls back).
+20 -12
View File
@@ -17,19 +17,26 @@ import (
// (→planner) or direct high-priority intents (→frontier). It does NOT run the // (→planner) or direct high-priority intents (→frontier). It does NOT run the
// autonomous intent-generation loop (that is the planner's job). // autonomous intent-generation loop (that is the planner's job).
type MainAgent struct { type MainAgent struct {
prov llm.Provider prov llm.Provider
model string model string
tx *transcript.Store // raw LLM conversation persistence (nil = off) tx *transcript.Store // raw LLM conversation persistence (nil = off)
window int // context window in tokens (for compaction) window int // context window in tokens (for compaction)
windowFn func() int // optional dynamic task-chain minimum windowFn func() int // optional dynamic task-chain minimum
maxTurns int // max agent turns per run (0 = unlimited) maxTurns int // max agent turns per run (0 = unlimited)
proxyAddr string // recording proxy for WebFetch (empty = direct) proxyAddr string // recording proxy for WebFetch (empty = direct)
proxyCACert string // recording proxy's CA cert path (HTTPS verify) proxyCACert string // recording proxy's CA cert path (HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default) webSearch WebSearchOpts // web_search tool backend selection (off by default)
workDir string // shared work dir (surfaced in prompt as artifact-output target) workDir string // shared work dir (surfaced in prompt as artifact-output target)
steerWork func(intentID int64, msg string) error // engine callback: steer a running work (nil = off) steerWork func(intentID int64, msg string) error // engine callback: steer a running work (nil = off)
nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming)
} }
// SetNonStreaming wires a resolver deciding whether runs use the non-streaming
// model path (true = non-streaming). nil/unset = streaming (default).
func (m *MainAgent) SetNonStreaming(fn func() bool) { m.nonStreamingFn = fn }
func (m *MainAgent) nonStreaming() bool { return m.nonStreamingFn != nil && m.nonStreamingFn() }
func NewMainAgent(prov llm.Provider, model, workDir string, tx *transcript.Store, window, maxTurns int) *MainAgent { func NewMainAgent(prov llm.Provider, model, workDir string, tx *transcript.Store, window, maxTurns int) *MainAgent {
return &MainAgent{prov: prov, model: model, workDir: workDir, tx: tx, window: window, maxTurns: maxTurns} return &MainAgent{prov: prov, model: model, workDir: workDir, tx: tx, window: window, maxTurns: maxTurns}
} }
@@ -128,7 +135,8 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, as *db.AssetStore, t
Compaction: compactionConfig(m.compactionWindow()), // long chats stay within the window Compaction: compactionConfig(m.compactionWindow()), // long chats stay within the window
Todos: actool.NewTodoStore(), // 会话级临时待办(TodoWrite),纯规划用,退出即丢 Todos: actool.NewTodoStore(), // 会话级临时待办(TodoWrite),纯规划用,退出即丢
// 命中预算(步数)→ SDK 跑收尾:向用户输出一句进展总结。Prompt 与收尾轮数可后台编辑(默认 10 轮)。 // 命中预算(步数)→ SDK 跑收尾:向用户输出一句进展总结。Prompt 与收尾轮数可后台编辑(默认 10 轮)。
Settlement: wrapupSettlement("mainagent", nil), Settlement: wrapupSettlement("mainagent", nil),
NonStreaming: m.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete
} }
if m.tx != nil { // persist raw human↔AI conversation; one accumulating file per task if m.tx != nil { // persist raw human↔AI conversation; one accumulating file per task
opts.Transcript = m.tx opts.Transcript = m.tx
+23 -15
View File
@@ -20,19 +20,20 @@ import (
// assets, judges whether the task goal is met, and emits 0..N exploration intents // assets, judges whether the task goal is met, and emits 0..N exploration intents
// into the frontier. It is the sole intent generator. // into the frontier. It is the sole intent generator.
type Planner struct { type Planner struct {
prov llm.Provider prov llm.Provider
model string model string
tx *transcript.Store // raw LLM conversation persistence (nil = off) tx *transcript.Store // raw LLM conversation persistence (nil = off)
window int // context window in tokens (for compaction) window int // context window in tokens (for compaction)
windowFn func() int // optional dynamic task-chain minimum windowFn func() int // optional dynamic task-chain minimum
maxTurns int // max agent turns per run (0 = unlimited) maxTurns int // max agent turns per run (0 = unlimited)
killWork func(intentID int64) error // engine callback to terminate a running work (nil = off) killWork func(intentID int64) error // engine callback to terminate a running work (nil = off)
steerWork func(intentID int64, msg string) error // engine callback to steer a running work mid-run (nil = off) steerWork func(intentID int64, msg string) error // engine callback to steer a running work mid-run (nil = off)
proxyAddr string // recording proxy for WebFetch (empty = direct) proxyAddr string // recording proxy for WebFetch (empty = direct)
proxyCACert string // recording proxy's CA cert path (HTTPS verify) proxyCACert string // recording proxy's CA cert path (HTTPS verify)
webSearch WebSearchOpts // web_search tool backend selection (off by default) webSearch WebSearchOpts // web_search tool backend selection (off by default)
workDir string // shared work dir (surfaced in prompt as artifact-output target) workDir string // shared work dir (surfaced in prompt as artifact-output target)
injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes) injectConstraints func() bool // resolver: inject task operation constraints into system prompt? (nil = yes)
nonStreamingFn func() bool // resolver: use non-streaming (Complete) path? (nil = streaming)
// todos keeps ONE plan-scratchpad per task (keyed by exploration id) so the // todos keeps ONE plan-scratchpad per task (keyed by exploration id) so the
// planner's multi-step plan survives across wake-ups — each Plan() is a fresh // planner's multi-step plan survives across wake-ups — each Plan() is a fresh
@@ -48,6 +49,12 @@ func NewPlanner(prov llm.Provider, model, workDir string, tx *transcript.Store,
func (p *Planner) SetCompactionWindowResolver(fn func() int) { p.windowFn = fn } func (p *Planner) SetCompactionWindowResolver(fn func() int) { p.windowFn = fn }
// SetNonStreaming wires a resolver deciding whether runs use the non-streaming
// model path (true = non-streaming). nil/unset = streaming (default).
func (p *Planner) SetNonStreaming(fn func() bool) { p.nonStreamingFn = fn }
func (p *Planner) nonStreaming() bool { return p.nonStreamingFn != nil && p.nonStreamingFn() }
func (p *Planner) compactionWindow() int { func (p *Planner) compactionWindow() int {
if p.windowFn != nil { if p.windowFn != nil {
return p.windowFn() return p.windowFn()
@@ -238,7 +245,7 @@ const plannerDefaultTmpl = `你是一个授权渗透测试系统的"规划者"
⚠️ 最重要的原则:**这一轮生成 0 个意图是完全正常、而且是最常见的结果。** 你不是"每次都要产出意图"的机器。绝大多数唤醒,frontier 里已有的意图已经覆盖了所有已知方向,你应当什么都不加、直接结束。重复/换措辞地生成已经存在的意图是严重错误。 ⚠️ 最重要的原则:**这一轮生成 0 个意图是完全正常、而且是最常见的结果。** 你不是"每次都要产出意图"的机器。绝大多数唤醒,frontier 里已有的意图已经覆盖了所有已知方向,你应当什么都不加、直接结束。重复/换措辞地生成已经存在的意图是严重错误。
决策流程(每次唤醒): 决策流程(每次唤醒):
1. **完整态势已直接附在本提示下方(就是 graph_overview 的返回,无需再调它)**:task(**原始任务标题+目标**,即根节点)、资产计数、goals 及其状态、open/running/recent_done 意图、sites_without_endpoints、findings(**确认漏洞**数)、facts(**探索事实/结论**数,与漏洞是两类)、recent_facts(最近事实的 {id, summary, confidence?},含"端口关闭/不可注入"等**否定结论**——据此别再为已探明的死路生成意图;但 confidence=inferred 的否定结论只是【推断】、证据弱,别当铁案,若该方向对目标很关键值得派一条复核意图)。**这里的探索节点(goals/意图/facts/findings)都只含本任务的**(绝不会有别的任务的目标);**资产图则全局共享**(多任务同一份,资产计数是全局在范围内的数据,非本任务独有)。需要更深的细节时才按需调:list_facts 列全部事实、list_findings 列全部漏洞、**node_detail(id)** 取某条的完整证据/详情(列表/recent_facts 只给摘要)。 1. **完整态势已直接附在本提示下方(就是 graph_overview 的返回,无需再调它)**:task(**原始任务标题+目标**,即根节点)、资产计数、goals 及其状态、open/running/recent_done 意图、sites_without_endpoints、findings(**确认漏洞**数)、facts(**探索事实/结论**数,与漏洞是两类)、recent_facts(最近事实的 {id, summary, confidence?},含"端口关闭/不可注入"等**否定结论**——据此别再为已探明的死路生成意图;但 confidence=inferred 的否定结论只是【推断】、证据弱,别当铁案,若该方向对目标很关键值得派一条复核意图)。**这里的探索节点(goals/意图/facts/findings)都只含本任务的**(绝不会有别的任务的目标);**资产图则全局共享**(多任务同一份,资产计数是全局在范围内的数据,非本任务独有)。需要更深的细节时才按需调:list_facts 分页列事实(最新在前,默认 20 条,可传 q 关键词过滤、before 翻页,返回带 total/has_more)、list_findings 列全部漏洞、**node_detail(id)** 取某条的完整证据/详情(列表/recent_facts 只给摘要)。
- open_intents / running_intents / recent_done_intents——"哪些方向已经有意图在覆盖/已尝试"。每个意图还带 **parents(上游:它派生自哪些事实/意图)和 yields(下游:它产生了哪些事实/发现)**——这就是探索图的**血缘关系**,据此理解"哪些事实来自哪个方向、能否综合成新方向"。recent_facts 里每个事实带 **from_intent**(由哪个意图产生)。 - open_intents / running_intents / recent_done_intents——"哪些方向已经有意图在覆盖/已尝试"。每个意图还带 **parents(上游:它派生自哪些事实/意图)和 yields(下游:它产生了哪些事实/发现)**——这就是探索图的**血缘关系**,据此理解"哪些事实来自哪个方向、能否综合成新方向"。recent_facts 里每个事实带 **from_intent**(由哪个意图产生)。
- sites_without_endpoints / findings——"哪些方向【可能】需要探索"。 - sites_without_endpoints / findings——"哪些方向【可能】需要探索"。
- 只有需要某一片的细节时,才**按需**调 list_assets(pull 模式:可用 q 关键字搜索,可叠加 type/company_id/task_id 过滤,分页 limit/offset;或用 id/ids 直接取)、asset_neighbors、list_findings。资产图全局共享,别默认拉全量。资产中可能包含非本次任务涉及到的资产,所以需要主要出现非本次任务相关的资产时忽略这些资产。 - 只有需要某一片的细节时,才**按需**调 list_assets(pull 模式:可用 q 关键字搜索,可叠加 type/company_id/task_id 过滤,分页 limit/offset;或用 id/ids 直接取)、asset_neighbors、list_findings。资产图全局共享,别默认拉全量。资产中可能包含非本次任务涉及到的资产,所以需要主要出现非本次任务相关的资产时忽略这些资产。
@@ -353,7 +360,8 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts
// 命中【本轮】步数预算→ SDK 跑收尾:把本轮已想清楚的结论落地(该派的 add_intent、 // 命中【本轮】步数预算→ SDK 跑收尾:把本轮已想清楚的结论落地(该派的 add_intent、
// 能证的 prove_goal、串行链记 TodoWrite),而非停止规划——planner 之后仍会被反复唤醒。 // 能证的 prove_goal、串行链记 TodoWrite),而非停止规划——planner 之后仍会被反复唤醒。
// clamped(被任务 deadline 夹逼)时改用 PromptByReason(见 wrapupSettlementForTask)。 // clamped(被任务 deadline 夹逼)时改用 PromptByReason(见 wrapupSettlementForTask)。
Settlement: settle, Settlement: settle,
NonStreaming: p.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete
} }
if p.tx != nil { // persist raw LLM conversation; one accumulating file per task's planner if p.tx != nil { // persist raw LLM conversation; one accumulating file per task's planner
opts.Transcript = p.tx opts.Transcript = p.tx
+36 -5
View File
@@ -51,6 +51,11 @@ type Config struct {
// "" = 不发送(默认); "low"/"medium"/"high"/"xhigh"/"max" = 对应强度. // "" = 不发送(默认); "low"/"medium"/"high"/"xhigh"/"max" = 对应强度.
// OpenAI 映射为顶层 reasoning_effort;Anthropic 映射为 output_config.effort. // OpenAI 映射为顶层 reasoning_effort;Anthropic 映射为 output_config.effort.
ReasoningEffort string ReasoningEffort string
// Stream 控制该 profile 是否使用流式(SSE)接口。true(默认)= 流式;false = 真·
// 非流式(发 stream:false,一次性拿完整 JSON,走 Provider.Complete)。非流式可绕开
// 某些网关糟糕的 SSE 实现(空帧、思考字段丢帧),代价是失去运行中的实时进度/实时
// token 计数。映射为 agentcore.Options.NonStreaming = !Stream。
Stream bool
} }
// compaction window resolution bounds (in K tokens). Below the floor the // compaction window resolution bounds (in K tokens). Below the floor the
@@ -116,6 +121,8 @@ func FromEnv() (Config, bool) {
BaseURL: os.Getenv("ARTEX_LLM_BASE_URL"), BaseURL: os.Getenv("ARTEX_LLM_BASE_URL"),
Model: os.Getenv("ARTEX_LLM_MODEL"), Model: os.Getenv("ARTEX_LLM_MODEL"),
Proxy: strings.TrimSpace(os.Getenv("ARTEX_LLM_PROXY")), Proxy: strings.TrimSpace(os.Getenv("ARTEX_LLM_PROXY")),
// 默认流式;ARTEX_LLM_STREAM=false/0/off 显式关闭走非流式。
Stream: !isFalsy(os.Getenv("ARTEX_LLM_STREAM")),
} }
switch prov { switch prov {
case "openai": case "openai":
@@ -147,6 +154,7 @@ func ConfigFrom(provider, model, baseURL, apiKey, proxy string) Config {
BaseURL: strings.TrimRight(strings.TrimSpace(baseURL), "/"), BaseURL: strings.TrimRight(strings.TrimSpace(baseURL), "/"),
APIKey: strings.TrimSpace(apiKey), APIKey: strings.TrimSpace(apiKey),
Proxy: strings.TrimSpace(proxy), Proxy: strings.TrimSpace(proxy),
Stream: true, // 默认流式;调用方按 profile 覆盖
} }
switch strings.TrimSpace(provider) { switch strings.TrimSpace(provider) {
case "openai": case "openai":
@@ -167,6 +175,16 @@ func ConfigFrom(provider, model, baseURL, apiKey, proxy string) Config {
return c return c
} }
// isFalsy reports whether an env-var string explicitly requests "off". Empty or
// unrecognized → false (so an unset var keeps the streaming default).
func isFalsy(s string) bool {
switch strings.ToLower(strings.TrimSpace(s)) {
case "0", "false", "off", "no":
return true
}
return false
}
// Provider returns the short provider name ("anthropic"/"openai"). // Provider returns the short provider name ("anthropic"/"openai").
func (c Config) Provider() string { func (c Config) Provider() string {
if c.Format == llm.FormatOpenAI { if c.Format == llm.FormatOpenAI {
@@ -288,11 +306,12 @@ func quotaAwareHTTPClient(proxy string) (*http.Client, error) {
} }
// TestConnection makes a minimal real completion to verify the provider/model/ // TestConnection makes a minimal real completion to verify the provider/model/
// endpoint/key actually work. Returns the round-trip latency. // endpoint/key actually work. Returns the round-trip latency and the model's
func TestConnection(ctx context.Context, c Config) (time.Duration, error) { // reply text.
func TestConnection(ctx context.Context, c Config) (time.Duration, string, error) {
prov, err := c.NewProvider() prov, err := c.NewProvider()
if err != nil { if err != nil {
return 0, err return 0, "", err
} }
ctx, cancel := context.WithTimeout(ctx, 30*time.Second) ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel() defer cancel()
@@ -302,12 +321,24 @@ func TestConnection(ctx context.Context, c Config) (time.Duration, error) {
// 就撞到输出上限(finish=length)、被截断,连接测试虽仍算通(err=nil)但显示成 // 就撞到输出上限(finish=length)、被截断,连接测试虽仍算通(err=nil)但显示成
// "已中断/length/resume" 一团糟。给足预算让它把 OK 干净吐完(finish=stop)。 // "已中断/length/resume" 一团糟。给足预算让它把 OK 干净吐完(finish=stop)。
// EscalateMaxTokens 保持 false:不因截断而抬额重试,避免 resume 循环空烧。 // EscalateMaxTokens 保持 false:不因截断而抬额重试,避免 resume 循环空烧。
_, err = agentcore.Run(ctx, agentcore.Options{ reply, err := agentcore.Run(ctx, agentcore.Options{
Provider: prov, Provider: prov,
SystemPrompt: []string{"你是连接测试。直接输出两个字符 OK 即可,不要思考、不要解释、不要别的。"}, SystemPrompt: []string{"你是连接测试。直接输出两个字符 OK 即可,不要思考、不要解释、不要别的。"},
PermissionMode: acperm.ModeBypass, PermissionMode: acperm.ModeBypass,
MaxTurns: 1, MaxTurns: 1,
MaxTokens: 8192, MaxTokens: 8192,
NonStreaming: !c.Stream, // 用该 profile 的真实收发模式做连接测试
}, "ping") }, "ping")
return time.Since(start), err lat := time.Since(start)
if err != nil {
return lat, "", err
}
// err==nil 还不够:请求通了但模型一个字都不吐的情况真实存在(思考把预算烧光、
// 正文被安全策略吞掉、兼容层把 content 丢了)。这种配置在会话里就是"不回话",
// 测试却报成功——正是本项要消除的落差。没有可见正文一律判失败。
reply = strings.TrimSpace(reply)
if reply == "" {
return lat, "", fmt.Errorf("模型无回复内容(请求已通,但未返回任何文本)")
}
return lat, reply, nil
} }
+51 -6
View File
@@ -898,19 +898,64 @@ func (t *ToolSet) listFindings() actool.CoreTool {
}) })
} }
// factsPageSize is the default page size for list_facts. Facts pile up on long
// tasks; returning all of them at once (the old behaviour) could blow up the
// context, so default to the newest page and let the agent page/filter for more.
const factsPageSize = 20
func (t *ToolSet) listFacts() actool.CoreTool { func (t *ToolSet) listFacts() actool.CoreTool {
return readTool("list_facts", "列本任务及直接关联任务的【探索事实/结论】(紧凑:id+摘要+状态)。关联任务条目带 source_task_id/inherited=true 且只读。详情用 node_detail(id),漏洞看 list_findings。", return readTool("list_facts", "分页列本任务及直接关联任务的【探索事实/结论】,最新在前(紧凑:id+摘要+状态,摘要过长会截断,全文用 node_detail(id))。参数均可选:limit(默认 20,上限 100)、before(游标,传上一页返回的 next_before 取更旧的一页;省略/0=最新一页)、q(按摘要关键词过滤)。返回 {facts, total, has_more, next_before}:total 是过滤后的总数,has_more=true 时用 next_before 继续翻页。关联任务条目带 source_task_id/inherited=true 且只读。漏洞看 list_findings。",
obj(map[string]any{}), obj(map[string]any{
func(context.Context, json.RawMessage) (actool.Result, error) { "limit": intp("返回条数,默认 20,上限 100"),
f, _ := t.ts.ListByKindWithSources(db.KindFact, 500) "before": intp("分页游标:只返回 id 小于该值的更旧事实;省略或 0 = 最新一页"),
"q": str("按事实摘要关键词过滤(不区分大小写);省略 = 不过滤"),
}),
func(_ context.Context, in json.RawMessage) (actool.Result, error) {
var a struct {
Limit int `json:"limit"`
Before int64 `json:"before"`
Q string `json:"q"`
}
_ = json.Unmarshal(in, &a)
limit := a.Limit
if limit <= 0 {
limit = factsPageSize
}
if limit > 100 {
limit = 100
}
f, hasMore, total, err := t.ts.ListByKindPageWithSources(db.KindFact, a.Before, limit, strings.TrimSpace(a.Q))
if err != nil {
return actool.Result{}, err
}
out := make([]map[string]any, 0, len(f)) out := make([]map[string]any, 0, len(f))
for _, n := range f { for _, n := range f {
out = append(out, compactNode(n)) out = append(out, compactFact(n))
} }
return jsonResult(out) res := map[string]any{"facts": out, "total": total, "has_more": hasMore}
if hasMore && len(f) > 0 {
res["next_before"] = f[len(f)-1].ID // 传回它取下一页(更旧的)
}
return jsonResult(res)
}) })
} }
// factSummaryMax caps a fact summary in list_facts output. Facts carry one-line
// conclusions, but nothing enforces brevity; a runaway summary must not bloat a
// whole page. Full text stays available via node_detail(id).
const factSummaryMax = 160
// compactFact is compactNode with the summary rune-capped for list_facts, so a
// page of facts stays bounded regardless of how long any single summary grew.
func compactFact(n *db.Node) map[string]any {
m := compactNode(n)
if s, ok := m["summary"].(string); ok && len([]rune(s)) > factSummaryMax {
m["summary"] = string([]rune(s)[:factSummaryMax]) + "…"
m["summary_truncated"] = true
}
return m
}
func (t *ToolSet) nodeDetail() actool.CoreTool { func (t *ToolSet) nodeDetail() actool.CoreTool {
return readTool("node_detail", "按 id 取本任务或直接关联任务的【探索图节点】完整内容。继承节点带 source_task_id/inherited=true 且只读。仅限 list_facts/list_findings/graph_overview 返回的探索节点 id;资产请用 list_assets/asset_neighbors。", return readTool("node_detail", "按 id 取本任务或直接关联任务的【探索图节点】完整内容。继承节点带 source_task_id/inherited=true 且只读。仅限 list_facts/list_findings/graph_overview 返回的探索节点 id;资产请用 list_assets/asset_neighbors。",
obj(map[string]any{"id": idp("探索图节点 id(非资产 id)")}, "id"), obj(map[string]any{"id": idp("探索图节点 id(非资产 id)")}, "id"),
+13 -1
View File
@@ -62,8 +62,19 @@ type Worker struct {
// injected into the worker system prompt. Read per run so the settings toggle // injected into the worker system prompt. Read per run so the settings toggle
// takes effect without rebuilding the agent. nil = inject (default). // takes effect without rebuilding the agent. nil = inject (default).
injectConstraints func() bool injectConstraints func() bool
// nonStreamingFn resolves whether this run uses the non-streaming (Complete)
// path. Read per run so a profile/task toggle takes effect without rebuilding
// the agent. nil = streaming (default).
nonStreamingFn func() bool
} }
// SetNonStreaming wires a resolver deciding whether runs use the non-streaming
// model path (true = non-streaming). nil/unset = streaming (default). Read per
// run so a profile or task-chain toggle takes effect without rebuilding.
func (w *Worker) SetNonStreaming(fn func() bool) { w.nonStreamingFn = fn }
func (w *Worker) nonStreaming() bool { return w.nonStreamingFn != nil && w.nonStreamingFn() }
// SetConstraintInject wires a resolver deciding whether this task's operation // SetConstraintInject wires a resolver deciding whether this task's operation
// constraints get injected into the worker system prompt. nil = inject (default). // constraints get injected into the worker system prompt. nil = inject (default).
func (w *Worker) SetConstraintInject(fn func() bool) { w.injectConstraints = fn } func (w *Worker) SetConstraintInject(fn func() bool) { w.injectConstraints = fn }
@@ -160,7 +171,7 @@ const workerDefaultTmpl = `你是一个授权渗透测试系统的"执行者"(wo
- insert_assets:登记新资产(资产图)。**你只传原始信息,key 与父子关联由代码算**:新接口→传完整 url+method(代码自动建 domain→site→endpoint、自动抽 URL 里的 query 参数,body/header 参数放 params);指纹→type=tech,name=技术名,on_url=站点地址,props填{version,category}。多个资产放 assets 数组一次批量登记。属性写在资产自己的 props 上,探索结论不要写这里。 - insert_assets:登记新资产(资产图)。**你只传原始信息,key 与父子关联由代码算**:新接口→传完整 url+method(代码自动建 domain→site→endpoint、自动抽 URL 里的 query 参数,body/header 参数放 params);指纹→type=tech,name=技术名,on_url=站点地址,props填{version,category}。多个资产放 assets 数组一次批量登记。属性写在资产自己的 props 上,探索结论不要写这里。
- record_fact:把探索【事实/结论】写入探索图并连到意图(传 intent_id)。正向/否定结论、观察、判断用它;**一次探索的多个观察汇总成一条事实**(summary 总结一句话 + detail 放细节),不要一个属性一条。真有多条不同结论才用 facts 数组。**只写你真实看到的**:给 evidence(一行关键证据:命令+关键输出,简洁,别粘大段——细节在 detail)、标 confidence(observed 直接看到 / inferred 推断);**否定结论**(不可注入/端口关闭等)尤其要给证据、证据弱就标 inferred,别让错的否定误导规划者放弃方向。 - record_fact:把探索【事实/结论】写入探索图并连到意图(传 intent_id)。正向/否定结论、观察、判断用它;**一次探索的多个观察汇总成一条事实**(summary 总结一句话 + detail 放细节),不要一个属性一条。真有多条不同结论才用 facts 数组。**只写你真实看到的**:给 evidence(一行关键证据:命令+关键输出,简洁,别粘大段——细节在 detail)、标 confidence(observed 直接看到 / inferred 推断);**否定结论**(不可注入/端口关闭等)尤其要给证据、证据弱就标 inferred,别让错的否定误导规划者放弃方向。
- report_finding:确认漏洞 → 记录(含 PoC,传 intent_id=你领到的意图id)。**只有你在本次运行里真实触发过该漏洞、拿到了可复现的证据(请求/响应或命令输出)才用它。** 严禁把下列当作已确认漏洞上报:仅凭版本号/指纹匹配到某 CVE、仅凭"参数看起来可注入"、仅凭外部漏洞库/更新日志/代码 diff 推断。**不要用查 CVE 库或"对比补丁版本"替代实际触发。** 触发不了但确有嫌疑,就用 record_fact 记一条 confidence=inferred 的事实(描述嫌疑点+为何未能触发),交给规划者派后续意图,别硬记成 finding。 - report_finding:确认漏洞 → 记录(含 PoC,传 intent_id=你领到的意图id)。**只有你在本次运行里真实触发过该漏洞、拿到了可复现的证据(请求/响应或命令输出)才用它。** 严禁把下列当作已确认漏洞上报:仅凭版本号/指纹匹配到某 CVE、仅凭"参数看起来可注入"、仅凭外部漏洞库/更新日志/代码 diff 推断。**不要用查 CVE 库或"对比补丁版本"替代实际触发。** 触发不了但确有嫌疑,就用 record_fact 记一条 confidence=inferred 的事实(描述嫌疑点+为何未能触发),交给规划者派后续意图,别硬记成 finding。
- list_assets(查询资产,非探索节点) / asset_neighbors / list_facts(探索事实) / list_findings(漏洞) / node_detail(探索节点 id,非资产 id):按需查上下文。 - list_assets(查询资产,非探索节点) / asset_neighbors / list_facts(探索事实,分页最新在前;默认 20 条,可传 q 关键词过滤、before 翻页) / list_findings(漏洞) / node_detail(探索节点 id,非资产 id):按需查上下文。
只在授权范围内操作;若系统提示顶部附有【操作约束】,那是最高优先级红线——任何命令/探测在执行前先自检是否违反,违反即不做(哪怕它落在你领到的意图里)。完成本意图后用一句话总结你做了什么、写回了哪些事实。务实、克制、聚焦这一条意图。` 只在授权范围内操作;若系统提示顶部附有【操作约束】,那是最高优先级红线——任何命令/探测在执行前先自检是否违反,违反即不做(哪怕它落在你领到的意图里)。完成本意图后用一句话总结你做了什么、写回了哪些事实。务实、克制、聚焦这一条意图。`
@@ -337,6 +348,7 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.
ToolOutputDir: cmdOutDir(runDir), ToolOutputDir: cmdOutDir(runDir),
Compaction: compactionConfig(w.compactionWindow()), // long tool-heavy runs stay within the window Compaction: compactionConfig(w.compactionWindow()), // long tool-heavy runs stay within the window
Todos: actool.NewTodoStore(), // 会话级临时待办(TodoWrite),纯规划用,退出即丢 Todos: actool.NewTodoStore(), // 会话级临时待办(TodoWrite),纯规划用,退出即丢
NonStreaming: w.nonStreaming(), // 该 profile 选非流式时走 Provider.Complete
} }
if hooks != nil { // typed-nil guard: only set when concrete (avoids harness panic) if hooks != nil { // typed-nil guard: only set when concrete (avoids harness panic)
opts.Hooks = hooks opts.Hooks = hooks
+57 -12
View File
@@ -19,18 +19,10 @@ type CommandRecord struct {
CreatedAt time.Time `json:"created_at"` CreatedAt time.Time `json:"created_at"`
} }
// ListCommands returns tool executions (tool_use + paired tool_result) across all // commandFilter builds the WHERE clause shared by the tool-execution list and
// explorations, with optional filtering and pagination. Covers every tool, not // its per-tool tally, so the summary always describes exactly the rows the table
// just Bash; q matches the tool name or its input. // pages through. Returns the clause, its args, and the next placeholder index.
func (d *DB) ListCommands(expID *int64, q string, page, size int) ([]CommandRecord, int, error) { func commandFilter(expID *int64, q string) (string, []any, int) {
if size <= 0 {
size = 50
}
if page < 0 {
page = 0
}
offset := page * size
where := `WHERE u.kind = 'tool_use'` where := `WHERE u.kind = 'tool_use'`
args := []any{} args := []any{}
argN := 1 argN := 1
@@ -45,6 +37,59 @@ func (d *DB) ListCommands(expID *int64, q string, page, size int) ([]CommandReco
args = append(args, "%"+q+"%") args = append(args, "%"+q+"%")
argN++ argN++
} }
return where, args, argN
}
// ToolStat is one tool's execution tally for the usage summary.
type ToolStat struct {
Tool string `json:"tool"`
Total int `json:"total"`
Errors int `json:"errors"`
}
// ToolStats counts executions grouped by tool under the same filters
// ListCommands takes. Unpaginated on purpose: the tally describes the whole
// filtered set, not the page currently on screen.
func (d *DB) ToolStats(expID *int64, q string) ([]ToolStat, error) {
where, args, _ := commandFilter(expID, q)
rows, err := d.Query(`
SELECT COALESCE(NULLIF(u.tool,''),'-') AS tool, COUNT(*) AS total,
COUNT(*) FILTER (WHERE COALESCE(r.is_error,false)) AS errors
FROM activity u
LEFT JOIN activity r ON r.tool_use_id = u.tool_use_id AND r.kind = 'tool_result'
`+where+`
GROUP BY 1
ORDER BY total DESC, tool ASC`, args...)
if err != nil {
return nil, err
}
defer rows.Close()
out := []ToolStat{}
for rows.Next() {
var s ToolStat
if err := rows.Scan(&s.Tool, &s.Total, &s.Errors); err != nil {
return nil, err
}
out = append(out, s)
}
return out, rows.Err()
}
// ListCommands returns tool executions (tool_use + paired tool_result) across all
// explorations, with optional filtering and pagination. Covers every tool, not
// just Bash; q matches the tool name or its input.
func (d *DB) ListCommands(expID *int64, q string, page, size int) ([]CommandRecord, int, error) {
if size <= 0 {
size = 50
}
if page < 0 {
page = 0
}
offset := page * size
where, args, argN := commandFilter(expID, q)
// count // count
var total int var total int
+15 -10
View File
@@ -39,18 +39,23 @@ type LLMProfile struct {
// usable when an agent/task binds it explicitly, it just never gets picked up // usable when an agent/task binds it explicitly, it just never gets picked up
// as a fallback target. // as a fallback target.
PoolExclude bool `json:"pool_exclude"` PoolExclude bool `json:"pool_exclude"`
// Streaming selects the wire protocol: true (default) = streaming (SSE);
// false = real non-streaming (stream:false, single JSON response via
// Provider.Complete). Non-streaming sidesteps flaky gateway SSE at the cost of
// live in-run progress. Maps to agent.Config.Stream.
Streaming bool `json:"streaming"`
} }
// profileCols is the read column list (hint variant, no api key) shared by the // profileCols is the read column list (hint variant, no api key) shared by the
// list query; profileColsKey is the same with api_key for the single-row loads. // list query; profileColsKey is the same with api_key for the single-row loads.
const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,'')` const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true)`
const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,'')` const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true)`
// scanProfile reads one row in the profileCols / profileColsKey column order. The // scanProfile reads one row in the profileCols / profileColsKey column order. The
// 7th column lands in APIKeyHint or APIKey depending on which list the caller used. // 7th column lands in APIKeyHint or APIKey depending on which list the caller used.
func scanProfile(sc interface{ Scan(...any) error }, into *string, p *LLMProfile) error { func scanProfile(sc interface{ Scan(...any) error }, into *string, p *LLMProfile) error {
return sc.Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, into, return sc.Scan(&p.ID, &p.Name, &p.Format, &p.BaseURL, &p.Proxy, &p.Model, into,
&p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude, &p.ThinkingType) &p.RatePerSecond, &p.RatePerMinute, &p.ContextWindowK, &p.ReasoningEffort, &p.IsDefault, &p.Priority, &p.PoolExclude, &p.ThinkingType, &p.Streaming)
} }
func (d *DB) ListProfiles() ([]*LLMProfile, error) { func (d *DB) ListProfiles() ([]*LLMProfile, error) {
@@ -123,18 +128,18 @@ func (d *DB) SaveProfile(p *LLMProfile) (int64, error) {
} }
if p.ID == 0 { if p.ID == 0 {
var id int64 var id int64
err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude,thinking_type) err := d.QueryRow(`INSERT INTO llm_profiles(name,format,base_url,proxy,model,api_key,api_key_hint,rate_per_second,rate_per_minute,context_window_k,reasoning_effort,priority,pool_exclude,thinking_type,streaming)
VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14) RETURNING id`, VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14,$15) RETURNING id`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType).Scan(&id) p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming).Scan(&id)
return id, err return id, err
} }
if p.APIKey == "" { if p.APIKey == "" {
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11,thinking_type=$12 WHERE id=$13`, _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11,thinking_type=$12,streaming=$13 WHERE id=$14`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.ID) p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.ID)
return p.ID, err return p.ID, err
} }
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13,thinking_type=$14 WHERE id=$15`, _, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13,thinking_type=$14,streaming=$15 WHERE id=$16`,
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.ID) p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.ID)
return p.ID, err return p.ID, err
} }
+34
View File
@@ -672,6 +672,40 @@ ORDER BY id DESC LIMIT $4`, s.expID, kind, before, limit+1)
return nodes, hasMore, nil return nodes, hasMore, nil
} }
// listByKindPageFiltered is ListByKindPage with an optional summary keyword
// filter (payload->>'summary' ILIKE %q%). Fetches one extra row so the caller can
// probe hasMore. Empty q = no filter. Newest-first by id.
func (s *ExplorationStore) listByKindPageFiltered(kind string, before int64, limit int, q string) ([]*Node, error) {
where := `exploration_id=$1 AND kind=$2 AND ($3 <= 0 OR id < $3)`
args := []any{s.expID, kind, before}
if q != "" {
args = append(args, "%"+q+"%")
where += fmt.Sprintf(` AND payload->>'summary' ILIKE $%d`, len(args))
}
args = append(args, limit+1)
rows, err := s.db.Query(`SELECT `+nodeCols+` FROM exploration_nodes
WHERE `+where+` ORDER BY id DESC LIMIT $`+fmt.Sprintf("%d", len(args)), args...)
if err != nil {
return nil, err
}
defer rows.Close()
return scanNodes(rows)
}
// countByKindFiltered counts nodes of a kind under the same summary filter, for a
// page's total. Empty q = count all of that kind.
func (s *ExplorationStore) countByKindFiltered(kind, q string) (int, error) {
where := `exploration_id=$1 AND kind=$2`
args := []any{s.expID, kind}
if q != "" {
args = append(args, "%"+q+"%")
where += fmt.Sprintf(` AND payload->>'summary' ILIKE $%d`, len(args))
}
var n int
err := s.db.QueryRow(`SELECT COUNT(*) FROM exploration_nodes WHERE `+where, args...).Scan(&n)
return n, err
}
// GetNode returns one node of this exploration by id (nil, nil if not found). // GetNode returns one node of this exploration by id (nil, nil if not found).
func (s *ExplorationStore) GetNode(id int64) (*Node, error) { func (s *ExplorationStore) GetNode(id int64) (*Node, error) {
n, err := scanNode(s.db.QueryRow(`SELECT `+nodeCols+` FROM exploration_nodes WHERE id=$1 AND exploration_id=$2`, id, s.expID)) n, err := scanNode(s.db.QueryRow(`SELECT `+nodeCols+` FROM exploration_nodes WHERE id=$1 AND exploration_id=$2`, id, s.expID))
+47
View File
@@ -108,6 +108,53 @@ func (s *ExplorationStore) ListByKindWithSources(kind string, limit int) ([]*Nod
return out, nil return out, nil
} }
// ListByKindPageWithSources is the paginated, keyword-filterable sibling of
// ListByKindWithSources: it returns one newest-first page (id < before, before<=0
// = newest) spanning this exploration and its direct sources, plus hasMore and the
// filtered total across all of them. Node ids are globally unique, so merging each
// store's own page and re-sorting by id DESC yields the true global page; fetching
// limit+1 per store guarantees the merged top-`limit` is complete.
func (s *ExplorationStore) ListByKindPageWithSources(kind string, before int64, limit int, q string) (nodes []*Node, hasMore bool, total int, err error) {
if limit <= 0 {
limit = 20
}
own, err := s.listByKindPageFiltered(kind, before, limit, q)
if err != nil {
return nil, false, 0, err
}
merged := own
total, err = s.countByKindFiltered(kind, q)
if err != nil {
return nil, false, 0, err
}
sources, err := s.DirectSourceStores()
if err != nil {
return nil, false, 0, err
}
for _, source := range sources {
page, err := source.Store.listByKindPageFiltered(kind, before, limit, q)
if err != nil {
return nil, false, 0, err
}
for _, n := range page {
merged = append(merged, markInheritedNode(n, source.Task.TaskID))
}
cnt, err := source.Store.countByKindFiltered(kind, q)
if err != nil {
return nil, false, 0, err
}
total += cnt
}
sort.Slice(merged, func(i, j int) bool { return merged[i].ID > merged[j].ID })
hasMore = len(merged) > limit
if hasMore {
merged = merged[:limit]
}
return merged, hasMore, total, nil
}
// GetNodeWithSources reads a node only when it belongs to this exploration or // GetNodeWithSources reads a node only when it belongs to this exploration or
// one of its direct sources. Inherited nodes are tagged so tool callers can keep // one of its direct sources. Inherited nodes are tagged so tool callers can keep
// them read-only and show their provenance. // them read-only and show their provenance.
+4
View File
@@ -310,6 +310,8 @@ CREATE TABLE IF NOT EXISTS llm_profiles (
-- pool_exclude true=不作为故障转移目标(仍可被 agent/任务显式绑定使用)。 -- pool_exclude true=不作为故障转移目标(仍可被 agent/任务显式绑定使用)。
priority INTEGER NOT NULL DEFAULT 0, priority INTEGER NOT NULL DEFAULT 0,
pool_exclude BOOLEAN NOT NULL DEFAULT false, pool_exclude BOOLEAN NOT NULL DEFAULT false,
-- streaming=true(默认)走流式 SSE;false 走真·非流式(stream:false,一次性 JSON)。
streaming BOOLEAN NOT NULL DEFAULT true,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now() updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
); );
@@ -320,6 +322,8 @@ CREATE TRIGGER trg_llm_upd BEFORE UPDATE ON llm_profiles
-- 轮询顺位/排除标记;补旧库。默认 0 / false = 全部配置都参与轮询。 -- 轮询顺位/排除标记;补旧库。默认 0 / false = 全部配置都参与轮询。
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 0; ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 0;
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS pool_exclude BOOLEAN NOT NULL DEFAULT false; ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS pool_exclude BOOLEAN NOT NULL DEFAULT false;
-- 流式开关;补旧库。默认 true = 保持既有的流式行为,旧配置无感升级。
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS streaming BOOLEAN NOT NULL DEFAULT true;
-- 思考开关字段 thinking_type,从旧的单一 reasoning_effort 语义一次性拆分而来。 -- 思考开关字段 thinking_type,从旧的单一 reasoning_effort 语义一次性拆分而来。
-- schema.sql 每次启动都执行,故迁移必须只跑一次:仅当该列尚不存在时才回填, -- schema.sql 每次启动都执行,故迁移必须只跑一次:仅当该列尚不存在时才回填,
+2 -2
View File
@@ -3,9 +3,10 @@ module github.com/Autumn-27/artex
go 1.26.3 go 1.26.3
require ( require (
github.com/Autumn-27/norma v0.2.9 github.com/Autumn-27/norma v0.3.2
github.com/golang-jwt/jwt/v5 v5.3.1 github.com/golang-jwt/jwt/v5 v5.3.1
github.com/jackc/pgx/v5 v5.10.0 github.com/jackc/pgx/v5 v5.10.0
github.com/klauspost/compress v1.18.6
github.com/lqqyt2423/go-mitmproxy v1.9.2 github.com/lqqyt2423/go-mitmproxy v1.9.2
github.com/miekg/dns v1.1.72 github.com/miekg/dns v1.1.72
github.com/projectdiscovery/dnsx v1.2.3 github.com/projectdiscovery/dnsx v1.2.3
@@ -29,7 +30,6 @@ require (
github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect
github.com/klauspost/compress v1.18.6 // indirect
github.com/kr/pretty v0.3.1 // indirect github.com/kr/pretty v0.3.1 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-isatty v0.0.20 // indirect
github.com/microcosm-cc/bluemonday v1.0.27 // indirect github.com/microcosm-cc/bluemonday v1.0.27 // indirect
+2 -2
View File
@@ -1,5 +1,5 @@
github.com/Autumn-27/norma v0.2.9 h1:VTt8Oe7XoiV85Dc+QOPu+zQ7IykQRm2aDTEbLEzbBe4= github.com/Autumn-27/norma v0.3.2 h1:3gW84SbSrz8luW8kbi2tv3O7CLdRMQvyylCHj1QUdNk=
github.com/Autumn-27/norma v0.2.9/go.mod h1:BudWrf7iHl96FgHqFr/HwN3mKYmGjJDt+xL2ZHXfFZw= github.com/Autumn-27/norma v0.3.2/go.mod h1:BudWrf7iHl96FgHqFr/HwN3mKYmGjJDt+xL2ZHXfFZw=
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro= github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so=
+38
View File
@@ -122,6 +122,44 @@ func (p *Pool) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[
} }
} }
// Complete implements llm.Provider with failover for non-streaming calls. A
// non-streaming request is atomic — it never delivers partial output — so every
// failover-eligible failure lands in the safe window and the next member can be
// tried without risk of duplicated output. Mirrors Stream's health-tripping and
// chain-exhaustion behavior.
func (p *Pool) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
order := p.order(req)
var lastErr error
for i, m := range order {
msg, sr, usage, err := m.Prov.Complete(ctx, req)
if err == nil {
p.health.Pass(m.ID)
return msg, sr, usage, nil
}
if !shouldFailover(ctx, err) {
// Non-failover error (e.g. ctx cancel, deterministic 4xx): surface as-is
// without tripping health, matching Stream's non-failover path.
p.health.Pass(m.ID)
return llm.Message{}, "", llm.Usage{}, err
}
lastErr = err
hard := isHardFailure(err)
if p.health.Trip(m.ID, trimErr(err), hard) {
log.Printf("[llmpool] 配置 %q(%s) 已熔断:%s", m.Name, m.Model, trimErr(err))
}
if i+1 < len(order) {
n := order[i+1]
log.Printf("[llmpool] LLM 故障转移:%q(%s) → %q(%s),原因:%s",
m.Name, m.Model, n.Name, n.Model, trimErr(err))
}
}
if lastErr == nil {
lastErr = ErrExhausted
}
log.Printf("[llmpool] 轮询链已耗尽(%d 个配置全部失败),最后错误:%s", len(order), trimErr(lastErr))
return llm.Message{}, "", llm.Usage{}, fmt.Errorf("%w:%v", ErrExhausted, lastErr)
}
// order picks the members to try, in order: skip those in a cooling-off window // order picks the members to try, in order: skip those in a cooling-off window
// and those whose context window can't hold this request, then rotate within each // and those whose context window can't hold this request, then rotate within each
// equal-priority group so same-priority profiles share the load. Never returns an // equal-priority group so same-priority profiles share the load. Never returns an
+11
View File
@@ -55,6 +55,17 @@ func (f *fakeProv) Stream(ctx context.Context, req llm.CompletionRequest) iter.S
} }
} }
func (f *fakeProv) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
acc := llm.NewAccumulator()
for ev, err := range f.Stream(ctx, req) {
if err != nil {
return llm.Message{}, "", llm.Usage{}, err
}
acc.Add(ev)
}
return acc.Message(), acc.StopReason, acc.Usage, nil
}
// ok builds a provider that always succeeds with one text delta. // ok builds a provider that always succeeds with one text delta.
func okProv(name, text string) *fakeProv { func okProv(name, text string) *fakeProv {
return &fakeProv{name: name, events: [][]llm.StreamEvent{{{Type: llm.SETextDelta, Text: text}}}} return &fakeProv{name: name, events: [][]llm.StreamEvent{{{Type: llm.SETextDelta, Text: text}}}}
+43 -4
View File
@@ -39,10 +39,10 @@ func TaskIDFrom(ctx context.Context) string {
// Recorder wraps an llm.Provider and records every completion call. // Recorder wraps an llm.Provider and records every completion call.
type Recorder struct { type Recorder struct {
inner llm.Provider inner llm.Provider
pg *db.DB pg *db.DB
model string // model name (from config, not in CompletionRequest) model string // model name (from config, not in CompletionRequest)
prof string // LLM profile name (from llm_profiles) prof string // LLM profile name (from llm_profiles)
// thinkingType / reasoningEffort 是配置级思考参数(思考开关 / 思考强度)。它们在 // thinkingType / reasoningEffort 是配置级思考参数(思考开关 / 思考强度)。它们在
// norma 的 buildBody() 里从 provider 配置注入真正的 HTTP body,不出现在 // norma 的 buildBody() 里从 provider 配置注入真正的 HTTP body,不出现在
// CompletionRequest 上,故 Recorder 需在此单独带一份,序列化时写进录制。 // CompletionRequest 上,故 Recorder 需在此单独带一份,序列化时写进录制。
@@ -179,6 +179,45 @@ func (r *Recorder) Stream(ctx context.Context, req llm.CompletionRequest) iter.S
} }
} }
// Complete implements the atomic completion path while preserving the same
// usage metering and optional body recording semantics as Stream.
func (r *Recorder) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
recordBodies := r.enabled == nil || r.enabled()
start := time.Now()
session := transcript.SessionIDFrom(ctx)
parsedID, worker := parseSession(session)
expID := db.ParseExpID(parsedID)
taskID := TaskIDFrom(ctx)
if taskID == "" {
taskID = parsedID
}
reqBody := ""
if recordBodies {
reqBody = r.serializeRequest(req)
}
msg, stopReason, usage, err := r.inner.Complete(ctx, req)
status := "ok"
if err != nil {
status = "error"
}
r.recordUsage(taskID, expID, worker, usage, int(time.Since(start).Milliseconds()), status)
if recordBodies {
var textBuf, thinkingBuf strings.Builder
for _, block := range msg.Content {
switch block.Type {
case llm.BlockText:
textBuf.WriteString(block.Text)
case llm.BlockThinking:
thinkingBuf.WriteString(block.Thinking)
}
}
r.record(req, session, taskID, worker, reqBody, start, textBuf.String(), thinkingBuf.String(), usage, stopReason, err)
}
return msg, stopReason, usage, err
}
// recordUsage appends one lightweight metering row to llm_usage (no bodies). Skips // recordUsage appends one lightweight metering row to llm_usage (no bodies). Skips
// zero-token calls with no model, which carry nothing worth metering. // zero-token calls with no model, which carry nothing worth metering.
func (r *Recorder) recordUsage(taskID string, expID int64, worker string, usage llm.Usage, latencyMs int, status string) { func (r *Recorder) recordUsage(taskID string, expID int64, worker string, usage llm.Usage, latencyMs int, status string) {
+36
View File
@@ -2,9 +2,28 @@ package llmrec
import ( import (
"context" "context"
"iter"
"testing" "testing"
"github.com/Autumn-27/norma/llm"
) )
type completeProvider struct{}
func (completeProvider) Stream(context.Context, llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] {
return func(func(llm.StreamEvent, error) bool) {}
}
func (completeProvider) Complete(context.Context, llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
return llm.Message{
Role: llm.RoleAssistant,
Content: []llm.ContentBlock{
{Type: llm.BlockThinking, Thinking: "reasoning"},
llm.TextBlock("answer"),
},
}, "stop", llm.Usage{InputTokens: 7, OutputTokens: 3}, nil
}
func TestTaskIDContextUsesExplicitRegistryID(t *testing.T) { func TestTaskIDContextUsesExplicitRegistryID(t *testing.T) {
ctx := WithTaskID(context.Background(), " 42 ") ctx := WithTaskID(context.Background(), " 42 ")
if got := TaskIDFrom(ctx); got != "42" { if got := TaskIDFrom(ctx); got != "42" {
@@ -27,3 +46,20 @@ func TestParseSessionFallbackIsExplorationScoped(t *testing.T) {
t.Fatalf("unexpected non-session parse: (%q,%q)", taskID, worker) t.Fatalf("unexpected non-session parse: (%q,%q)", taskID, worker)
} }
} }
func TestCompleteForwardsAtomicResponse(t *testing.T) {
recorder := Wrap(completeProvider{}, nil, "model", "profile", "", "", func() bool { return false })
msg, stopReason, usage, err := recorder.Complete(context.Background(), llm.CompletionRequest{})
if err != nil {
t.Fatalf("Complete() error: %v", err)
}
if got := msg.Text(); got != "answer" {
t.Fatalf("Complete() text=%q want answer", got)
}
if stopReason != "stop" {
t.Fatalf("Complete() stop reason=%q want stop", stopReason)
}
if usage.InputTokens != 7 || usage.OutputTokens != 3 {
t.Fatalf("Complete() usage=%+v", usage)
}
}
+32 -8
View File
@@ -15,14 +15,7 @@ func (s *Server) pgListCommands(w http.ResponseWriter, r *http.Request) {
size := atoiDefault(q.Get("size"), 50) size := atoiDefault(q.Get("size"), 50)
keyword := q.Get("q") keyword := q.Get("q")
var expID *int64 records, total, err := s.m.PG().ListCommands(commandTaskFilter(q.Get("task")), keyword, page, size)
if tv := q.Get("task"); tv != "" {
if n, err := strconv.ParseInt(tv, 10, 64); err == nil {
expID = &n
}
}
records, total, err := s.m.PG().ListCommands(expID, keyword, page, size)
if err != nil { if err != nil {
writeErr(w, 500, err.Error()) writeErr(w, 500, err.Error())
return return
@@ -30,6 +23,37 @@ func (s *Server) pgListCommands(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"commands": records, "total": total}) writeJSON(w, 200, map[string]any{"commands": records, "total": total})
} }
// pgToolStats returns per-tool call counts for the tool-execution history, under
// the same task/keyword filters the list takes — the summary describes the whole
// filtered set, not the page on screen.
func (s *Server) pgToolStats(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
pg := s.m.PG()
if pg == nil {
writeJSON(w, 200, map[string]any{"stats": []db.ToolStat{}})
return
}
stats, err := pg.ToolStats(commandTaskFilter(q.Get("task")), q.Get("q"))
if err != nil {
writeErr(w, 500, err.Error())
return
}
writeJSON(w, 200, map[string]any{"stats": stats})
}
// commandTaskFilter parses the ?task= exploration id. Absent or unparsable → nil
// (no filter), matching the list endpoint's long-standing lenient behaviour.
func commandTaskFilter(v string) *int64 {
if v == "" {
return nil
}
n, err := strconv.ParseInt(v, 10, 64)
if err != nil {
return nil
}
return &n
}
// pgListLLMRecords returns paginated LLM call records. // pgListLLMRecords returns paginated LLM call records.
func (s *Server) pgListLLMRecords(w http.ResponseWriter, r *http.Request) { func (s *Server) pgListLLMRecords(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query() q := r.URL.Query()
+4
View File
@@ -580,6 +580,9 @@ type LLMProfileDTO struct {
// pool_exclude=true 则不作为故障转移目标,但仍可被 agent/任务显式绑定。 // pool_exclude=true 则不作为故障转移目标,但仍可被 agent/任务显式绑定。
Priority int `json:"priority"` Priority int `json:"priority"`
PoolExclude bool `json:"pool_exclude"` PoolExclude bool `json:"pool_exclude"`
// 收发模式:true=流式(SSE) | false=非流式。没有 omitempty —— false 必须出现在
// 响应里,否则前端读不到「非流式」,开关会回落成默认的流式。
Streaming bool `json:"streaming"`
} }
func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO { func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO {
@@ -599,6 +602,7 @@ func llmProfileDTO(p *db.LLMProfile) LLMProfileDTO {
IsDefault: p.IsDefault, IsDefault: p.IsDefault,
Priority: p.Priority, Priority: p.Priority,
PoolExclude: p.PoolExclude, PoolExclude: p.PoolExclude,
Streaming: p.Streaming,
} }
} }
+2 -1
View File
@@ -383,7 +383,8 @@ func (s *Server) createGoals(ctx context.Context, t *Task, emit func(db.Activity
} }
taskID, _ := strconv.ParseInt(t.ID, 10, 64) taskID, _ := strconv.ParseInt(t.ID, 10, 64)
s.engine.BeginLLMCall(t.ID) s.engine.BeginLLMCall(t.ID)
decomposed := agent.DecomposeGoalsWithProvider(ctx, s.agentsForTask(t).runtime, s.m.dir, t.Goal, t.Description, as, t.Store, taskID, emit) goalRuntime := s.agentsForTask(t).runtime
decomposed := agent.DecomposeGoalsWithProvider(ctx, goalRuntime, s.m.dir, t.Goal, t.Description, as, t.Store, taskID, goalRuntime.nonStreaming(), emit)
s.engine.EndLLMCall(t.ID) s.engine.EndLLMCall(t.ID)
for _, g := range decomposed { for _, g := range decomposed {
if strings.TrimSpace(g.Text) != "" { if strings.TrimSpace(g.Text) != "" {
+6 -4
View File
@@ -450,9 +450,9 @@ func (s *Server) seedOrchestrationTools() {
s.seedPlannerDefaultBindings() s.seedPlannerDefaultBindings()
s.seedAutoReportFindingBinding() s.seedAutoReportFindingBinding()
s.unbindGoalMetDefault() s.unbindGoalMetDefault()
s.reseedGoalsPrompt() // goals 提示词加入「抽操作约束」步 → 旧库追加一版新默认(一次性) s.reseedGoalsPrompt() // goals 提示词加入「抽操作约束」步 → 旧库追加一版新默认(一次性)
s.reseedMainAgentPrompt() // mainagent 提示词加入「目标达成后 add_intent 反问是否建目标」(一次性) s.reseedMainAgentPrompt() // mainagent 提示词加入「目标达成后 add_intent 反问是否建目标」(一次性)
s.seedReporterAgent() // 预置「报告撰写」agent + 工具绑定 + finding 触发器(一次性) s.seedReporterAgent() // 预置「报告撰写」agent + 工具绑定 + finding 触发器(一次性)
// 注:pentest 的默认工具绑定无需迁移——BuiltinToolSeeds 在全新初始化时就把 // 注:pentest 的默认工具绑定无需迁移——BuiltinToolSeeds 在全新初始化时就把
// list_assets/insert_assets/report_finding/list_findings/list_companies 连同 // list_assets/insert_assets/report_finding/list_findings/list_companies 连同
// pentest 一起 seed 好了(项目尚无旧库,不做迁移)。 // pentest 一起 seed 好了(项目尚无旧库,不做迁移)。
@@ -464,7 +464,7 @@ func (s *Server) seedOrchestrationTools() {
// reaches an old DB otherwise. Preserves each tool's agent binding + enabled flag. // reaches an old DB otherwise. Preserves each tool's agent binding + enabled flag.
// Bump the flag whenever these tools' schemas/descriptions change in code. // Bump the flag whenever these tools' schemas/descriptions change in code.
func (s *Server) refreshBuiltinToolSchemas() { func (s *Server) refreshBuiltinToolSchemas() {
const flag = "tool_schema_refresh_v6_insert_assets_related" const flag = "tool_schema_refresh_v7_list_facts_paging"
if v, _, _ := s.m.pg.GetSetting(flag); v == "true" { if v, _, _ := s.m.pg.GetSetting(flag); v == "true" {
return return
} }
@@ -480,7 +480,9 @@ func (s *Server) refreshBuiltinToolSchemas() {
// “结束空轮”的手段、刚开跑就误判整个任务完成。 // “结束空轮”的手段、刚开跑就误判整个任务完成。
// - insert_assets:新增 related 入参(标记资产是否与当前任务相关、决定是否入覆盖度), // - insert_assets:新增 related 入参(标记资产是否与当前任务相关、决定是否入覆盖度),
// SeedTool 首插入only,旧库已 seed 的 schema 否则收不到这个新参数。 // SeedTool 首插入only,旧库已 seed 的 schema 否则收不到这个新参数。
refreshBuiltin := map[string]bool{"goal_met": true, "insert_assets": true} // - list_facts:改为分页,新增 limit/before/q 入参;旧库已 seed 的空 schema 否则
// 在工具管理页显示「无参数」,模型也拿不到这几个参数说明。
refreshBuiltin := map[string]bool{"goal_met": true, "insert_assets": true, "list_facts": true}
for _, sd := range agent.BuiltinToolSeeds() { for _, sd := range agent.BuiltinToolSeeds() {
if !refreshBuiltin[sd.Key] { if !refreshBuiltin[sd.Key] {
continue continue
+44 -6
View File
@@ -311,6 +311,7 @@ func (s *Server) loadLLMConfig() (agent.Config, bool) {
cfg.ContextWindowK = p.ContextWindowK cfg.ContextWindowK = p.ContextWindowK
cfg.ThinkingType = p.ThinkingType cfg.ThinkingType = p.ThinkingType
cfg.ReasoningEffort = p.ReasoningEffort cfg.ReasoningEffort = p.ReasoningEffort
cfg.Stream = p.Streaming
if cfg.APIKey == "" { if cfg.APIKey == "" {
return cfg, false return cfg, false
} }
@@ -328,14 +329,15 @@ func (s *Server) saveLLMConfig(cfg agent.Config) error {
} }
var id int64 var id int64
// agent.Config carries no failover fields, so carry the stored ones forward — // agent.Config carries no failover fields, so carry the stored ones forward —
// otherwise this legacy endpoint would silently reset the profile's priority // otherwise this legacy endpoint would silently reset the profile's priority,
// and pool_exclude to zero values on every save. // pool_exclude, and streaming to zero values on every save.
var priority int var priority int
var poolExclude bool var poolExclude bool
streaming := true // 旧库/新建默认流式
if profs, _ := s.m.pg.ListProfiles(); profs != nil { if profs, _ := s.m.pg.ListProfiles(); profs != nil {
for _, p := range profs { for _, p := range profs {
if p.Name == "default" { if p.Name == "default" {
id, priority, poolExclude = p.ID, p.Priority, p.PoolExclude id, priority, poolExclude, streaming = p.ID, p.Priority, p.PoolExclude, p.Streaming
break break
} }
} }
@@ -344,7 +346,7 @@ func (s *Server) saveLLMConfig(cfg agent.Config) error {
ID: id, Name: "default", Format: format, Model: cfg.Model, BaseURL: cfg.BaseURL, Proxy: cfg.Proxy, ID: id, Name: "default", Format: format, Model: cfg.Model, BaseURL: cfg.BaseURL, Proxy: cfg.Proxy,
APIKey: cfg.APIKey, RatePerSecond: cfg.RatePerSecond, RatePerMinute: cfg.RatePerMinute, APIKey: cfg.APIKey, RatePerSecond: cfg.RatePerSecond, RatePerMinute: cfg.RatePerMinute,
ContextWindowK: cfg.ContextWindowK, ThinkingType: cfg.ThinkingType, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true, ContextWindowK: cfg.ContextWindowK, ThinkingType: cfg.ThinkingType, ReasoningEffort: cfg.ReasoningEffort, IsDefault: true,
Priority: priority, PoolExclude: poolExclude, Priority: priority, PoolExclude: poolExclude, Streaming: streaming,
}) })
if err != nil { if err != nil {
return err return err
@@ -394,6 +396,7 @@ func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent
wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory"))) wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory")))
wk.SetWebSearch(s.webSearchFor("worker")) wk.SetWebSearch(s.webSearchFor("worker"))
wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开;每轮读) wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开;每轮读)
wk.SetNonStreaming(nonStreamingResolver(wCfg)) // 该 profile 选非流式时走 Provider.Complete
pProv, pCfg := s.providerForAgent("planner", pinID, gProv, gCfg) pProv, pCfg := s.providerForAgent("planner", pinID, gProv, gCfg)
pl := agent.NewPlanner(pProv, pCfg.Model, s.m.dir, tx, pCfg.CompactionWindow(), s.agentMaxTurns("planner")) pl := agent.NewPlanner(pProv, pCfg.Model, s.m.dir, tx, pCfg.CompactionWindow(), s.agentMaxTurns("planner"))
pl.SetKillWork(s.engine.KillWork) // planner kill_work → terminate a running work pl.SetKillWork(s.engine.KillWork) // planner kill_work → terminate a running work
@@ -401,9 +404,18 @@ func (s *Server) buildPlannerWorker(pinID *int64, gProv llm.Provider, gCfg agent
pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy
pl.SetWebSearch(s.webSearchFor("planner")) pl.SetWebSearch(s.webSearchFor("planner"))
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开;每轮读) pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开;每轮读)
pl.SetNonStreaming(nonStreamingResolver(pCfg)) // 该 profile 选非流式时走 Provider.Complete
return pl, wk return pl, wk
} }
// nonStreamingResolver returns a resolver capturing a profile's streaming choice.
// The agents read it per run; a profile change rebuilds the agents (applyLLM),
// so the captured value is always the one in effect for this build.
func nonStreamingResolver(cfg agent.Config) func() bool {
nonStreaming := !cfg.Stream
return func() bool { return nonStreaming }
}
// applyLLM (re)builds the planner/worker/main-agent from cfg and installs them on // applyLLM (re)builds the planner/worker/main-agent from cfg and installs them on
// the running engine as the GLOBAL active pair. Safe to call at runtime (UI configures LLM). // the running engine as the GLOBAL active pair. Safe to call at runtime (UI configures LLM).
func (s *Server) applyLLM(cfg agent.Config) error { func (s *Server) applyLLM(cfg agent.Config) error {
@@ -447,12 +459,14 @@ func (s *Server) applyLLM(cfg agent.Config) error {
s.mainAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy s.mainAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) // WebFetch through the recording proxy
s.mainAgent.SetWebSearch(s.webSearchFor("mainagent")) s.mainAgent.SetWebSearch(s.webSearchFor("mainagent"))
s.mainAgent.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏 s.mainAgent.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏
s.mainAgent.SetNonStreaming(nonStreamingResolver(mCfg))
// chat agent serves MANY custom agents by key → it holds the GLOBAL opts // chat agent serves MANY custom agents by key → it holds the GLOBAL opts
// (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。 // (backend/key) and gates Enabled per-conversation-agent at Chat time. 对话始终用激活配置。
s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner s.chatAgent = agent.NewChatAgent(prov, cfg.Model, s.m.dir, tx, win) // chat page runner
s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) s.chatAgent.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
s.chatAgent.SetWebSearch(s.m.WebSearchOpts()) s.chatAgent.SetWebSearch(s.m.WebSearchOpts())
s.chatAgent.SetGuard(s.chatGuard()) s.chatAgent.SetGuard(s.chatGuard())
s.chatAgent.SetNonStreaming(nonStreamingResolver(cfg))
s.llmProv = prov s.llmProv = prov
s.llmCfg = cfg s.llmCfg = cfg
s.llmOn = true s.llmOn = true
@@ -478,6 +492,7 @@ func (s *Server) loadProfileConfig(id int64) (agent.Config, bool) {
cfg.ContextWindowK = p.ContextWindowK cfg.ContextWindowK = p.ContextWindowK
cfg.ThinkingType = p.ThinkingType cfg.ThinkingType = p.ThinkingType
cfg.ReasoningEffort = p.ReasoningEffort cfg.ReasoningEffort = p.ReasoningEffort
cfg.Stream = p.Streaming
if cfg.APIKey == "" { if cfg.APIKey == "" {
return cfg, false return cfg, false
} }
@@ -629,6 +644,7 @@ func (s *Server) chatAgentForProfile(id int64) *agent.ChatAgent {
ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) ca.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
ca.SetWebSearch(s.m.WebSearchOpts()) ca.SetWebSearch(s.m.WebSearchOpts())
ca.SetGuard(s.chatGuard()) ca.SetGuard(s.chatGuard())
ca.SetNonStreaming(nonStreamingResolver(cfg))
s.profMu.Lock() s.profMu.Lock()
if ex := s.profChatAgents[id]; ex != nil { // lost the race → keep the winner if ex := s.profChatAgents[id]; ex != nil { // lost the race → keep the winner
ca = ex ca = ex
@@ -786,6 +802,7 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("GET /api/traffic/exchange", s.getTrafficExchange) mux.HandleFunc("GET /api/traffic/exchange", s.getTrafficExchange)
mux.HandleFunc("GET /api/traffic/blob", s.getTrafficBlob) mux.HandleFunc("GET /api/traffic/blob", s.getTrafficBlob)
mux.HandleFunc("GET /api/commands", s.pgListCommands) mux.HandleFunc("GET /api/commands", s.pgListCommands)
mux.HandleFunc("GET /api/commands/stats", s.pgToolStats) // 按工具聚合调用次数
mux.HandleFunc("GET /api/llm/records", s.pgListLLMRecords) mux.HandleFunc("GET /api/llm/records", s.pgListLLMRecords)
mux.HandleFunc("DELETE /api/llm/records", s.pgDeleteLLMRecords) mux.HandleFunc("DELETE /api/llm/records", s.pgDeleteLLMRecords)
mux.HandleFunc("GET /api/llm/records/tasks", s.pgLLMTasks) mux.HandleFunc("GET /api/llm/records/tasks", s.pgLLMTasks)
@@ -1335,6 +1352,7 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) {
ThinkingType string `json:"thinking_type"` ThinkingType string `json:"thinking_type"`
ReasoningEffort string `json:"reasoning_effort"` ReasoningEffort string `json:"reasoning_effort"`
ProfileID *int64 `json:"profile_id"` // 测已存 profile 时传入:api_key 为空则用它存的 key ProfileID *int64 `json:"profile_id"` // 测已存 profile 时传入:api_key 为空则用它存的 key
Streaming *bool `json:"streaming"` // 省略=流式,与保存 profile 时同一套默认
} }
if err := json.NewDecoder(r.Body).Decode(&req); err != nil { if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeErr(w, 400, err.Error()) writeErr(w, 400, err.Error())
@@ -1345,6 +1363,11 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) {
// reasoning_effort/thinking field fails the test too (no false "test ok, run 400"). // reasoning_effort/thinking field fails the test too (no false "test ok, run 400").
cfg.ThinkingType = req.ThinkingType cfg.ThinkingType = req.ThinkingType
cfg.ReasoningEffort = req.ReasoningEffort cfg.ReasoningEffort = req.ReasoningEffort
// 同理,收发模式也照该 profile 的选择来:只支持其中一种通道的端点必须在这里就
// 暴露,而不是等会话里才发现"测试通过的配置根本跑不动"。
if req.Streaming != nil {
cfg.Stream = *req.Streaming
}
// API Key 解析优先级:表单输入 > 指定 profile 存的 key > 全局配置的 key。 // API Key 解析优先级:表单输入 > 指定 profile 存的 key > 全局配置的 key。
// 已存 profile 的 key 不回传浏览器,所以测试已存配置时表单为空,需从 DB 取。 // 已存 profile 的 key 不回传浏览器,所以测试已存配置时表单为空,需从 DB 取。
if cfg.APIKey == "" && req.ProfileID != nil { if cfg.APIKey == "" && req.ProfileID != nil {
@@ -1361,12 +1384,27 @@ func (s *Server) testLLM(w http.ResponseWriter, r *http.Request) {
writeJSON(w, 200, map[string]any{"ok": false, "error": "未提供 API Key"}) writeJSON(w, 200, map[string]any{"ok": false, "error": "未提供 API Key"})
return return
} }
lat, err := agent.TestConnection(r.Context(), cfg) lat, reply, err := agent.TestConnection(r.Context(), cfg)
if err != nil { if err != nil {
writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error()}) writeJSON(w, 200, map[string]any{"ok": false, "error": err.Error()})
return return
} }
writeJSON(w, 200, map[string]any{"ok": true, "latency_ms": lat.Milliseconds(), "model": cfg.Model}) // 回传模型实际回复,让"测试通过"有据可查:看得见它确实说了话,而不只是 HTTP 200。
writeJSON(w, 200, map[string]any{
"ok": true, "latency_ms": lat.Milliseconds(), "model": cfg.Model, "reply": truncateReply(reply),
})
}
// truncateReply clips a connection-test reply for display. A model told to answer
// "OK" can still ramble (or think out loud); the UI only needs enough to show it
// really said something. Rune-based so multibyte text never splits mid-character.
func truncateReply(s string) string {
const max = 200
r := []rune(s)
if len(r) <= max {
return s
}
return string(r[:max]) + "…"
} }
type createTaskReq struct { type createTaskReq struct {
+7 -2
View File
@@ -1736,10 +1736,14 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) {
return return
} }
// APIKey has json:"-" on db.LLMProfile so it is never leaked to the UI on // APIKey has json:"-" on db.LLMProfile so it is never leaked to the UI on
// read; accept it here via a sibling field for create/update. // read; accept it here via a sibling field for create/update. Streaming is a
// *bool shadowing the embedded json:"streaming": an absent field must default
// to streaming (true), which a plain bool zero value (false = non-streaming)
// would get wrong — legacy/partial clients that never send it stay streaming.
var body struct { var body struct {
db.LLMProfile db.LLMProfile
APIKey string `json:"api_key"` APIKey string `json:"api_key"`
Streaming *bool `json:"streaming"`
} }
if err := decode(r, &body); err != nil { if err := decode(r, &body); err != nil {
writeErr(w, 400, err.Error()) writeErr(w, 400, err.Error())
@@ -1747,6 +1751,7 @@ func (s *Server) pgSaveProfile(w http.ResponseWriter, r *http.Request) {
} }
p := body.LLMProfile p := body.LLMProfile
p.APIKey = body.APIKey p.APIKey = body.APIKey
p.Streaming = body.Streaming == nil || *body.Streaming
id, err := pg.SaveProfile(&p) id, err := pg.SaveProfile(&p)
if err != nil { if err != nil {
writeErr(w, 500, err.Error()) writeErr(w, 500, err.Error())
+102 -4
View File
@@ -139,6 +139,36 @@ func (r *taskLLMRuntime) current() (*Task, int64, int64, llm.Provider, error) {
return t, 0, pt.LLMChainRevision, prov, nil return t, 0, pt.LLMChainRevision, prov, nil
} }
// nonStreaming reports whether the task's currently-active LLM source is set to
// non-streaming. It mirrors current()'s source precedence (agent binding →
// active chain profile → global) but reads only the streaming flag. Read-only
// and best-effort: any resolution error falls back to streaming (false), the
// safe default. If failover switches to a profile with a different streaming
// setting, the change takes effect on the next agent run (a fresh Session is
// built per run in captureRun).
func (r *taskLLMRuntime) nonStreaming() bool {
taskNum, err := parseTaskID(r.taskID)
if err != nil {
return false
}
pt, err := r.s.m.pg.GetTask(taskNum)
if err != nil || pt == nil {
return false
}
if _, cfg, ok := r.s.agentBindingProvider(r.agentKey); ok {
return !cfg.Stream
}
if len(pt.LLMProfileIDs) > 0 && pt.ActiveLLMProfileID != nil {
if _, cfg, ok := r.s.providerForProfile(*pt.ActiveLLMProfileID); ok {
return !cfg.Stream
}
}
if _, cfg, ok := r.s.globalProvider(); ok {
return !cfg.Stream
}
return false
}
func parseTaskID(id string) (int64, error) { func parseTaskID(id string) (int64, error) {
n, err := strconv.ParseInt(id, 10, 64) n, err := strconv.ParseInt(id, 10, 64)
if err != nil || n <= 0 { if err != nil || n <= 0 {
@@ -147,9 +177,11 @@ func parseTaskID(id string) (int64, error) {
return n, nil return n, nil
} }
func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { // streamHooks builds the failover/exhaustion callbacks shared by Stream and
ctx = llmrec.WithTaskID(ctx, r.taskID) // Complete: how to read the current profile selection, how to mark it quota
hooks := taskLLMStreamHooks{ // exhausted, and how to emit a failover transition.
func (r *taskLLMRuntime) streamHooks() taskLLMStreamHooks {
return taskLLMStreamHooks{
current: func() (taskLLMSelection, error) { current: func() (taskLLMSelection, error) {
task, profileID, revision, provider, err := r.current() task, profileID, revision, provider, err := r.current()
return taskLLMSelection{task: task, profileID: profileID, revision: revision, provider: provider}, err return taskLLMSelection{task: task, profileID: profileID, revision: revision, provider: provider}, err
@@ -169,7 +201,70 @@ func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest)
r.s.emitTaskLLMTransition(selection.task, transition, cause) r.s.emitTaskLLMTransition(selection.task, transition, cause)
}, },
} }
return streamTaskLLM(ctx, r.taskID, req, hooks) }
func (r *taskLLMRuntime) Stream(ctx context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] {
ctx = llmrec.WithTaskID(ctx, r.taskID)
return streamTaskLLM(ctx, r.taskID, req, r.streamHooks())
}
// Complete is the non-streaming counterpart of Stream. A non-streaming call is
// atomic — it never delivers partial output — so every failure is safe to retry
// on the same provider or fail over to the next profile without risking
// duplicated model output or tool execution (the "committed" bookkeeping the
// streaming path needs is unnecessary here).
func (r *taskLLMRuntime) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
ctx = llmrec.WithTaskID(ctx, r.taskID)
return completeTaskLLM(ctx, r.taskID, req, r.streamHooks())
}
func completeTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) (llm.Message, string, llm.Usage, error) {
for {
selection, err := hooks.current()
if err != nil {
return llm.Message{}, "", llm.Usage{}, err
}
var (
msg llm.Message
sr string
usage llm.Usage
callErr error
)
// 同 provider 安全窗口重试:非流式调用要么整体成功、要么整体失败,没有
// 中途已交付输出的问题,所以任何瞬时失败都可原样重试。
for attempt := 0; ; attempt++ {
msg, sr, usage, callErr = selection.provider.Complete(ctx, req)
if callErr != nil && ctx.Err() == nil &&
attempt < sameProviderStreamRetries && isRetryableStreamError(callErr) {
backoff := sameProviderRetryBackoff(attempt)
log.Printf("[task-llm] task %s 非流式调用失败,%v 后同 provider 重试 (%d/%d): %v",
taskID, backoff, attempt+1, sameProviderStreamRetries, callErr)
if sleepCtx(ctx, backoff) {
break // 退避期间 ctx 取消 → 停止重试
}
continue
}
break
}
if callErr == nil {
return msg, sr, usage, nil
}
// profileID=0: 显式链已被清空;非额度错误:透传。二者都不改任务 failover 状态。
if selection.profileID == 0 || !isQuotaExhaustedError(callErr) {
return llm.Message{}, "", llm.Usage{}, callErr
}
transition, markErr := hooks.exhaust(selection, callErr)
if markErr != nil {
return llm.Message{}, "", llm.Usage{}, fmt.Errorf("mark profile quota exhausted after %v: %w", callErr, markErr)
}
if transition.Advanced && !transition.Stale && hooks.transition != nil {
hooks.transition(selection, transition, callErr)
}
if !transition.Stale && transition.NextProfileID == nil {
return llm.Message{}, "", llm.Usage{}, &taskLLMError{taskID: taskID, chainExhausted: transition.ChainExhausted, cause: callErr}
}
// 没有任何输出交付给调用方,换下一个 profile 重放同一逻辑请求是安全的。
}
} }
func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) iter.Seq2[llm.StreamEvent, error] { func streamTaskLLM(ctx context.Context, taskID string, req llm.CompletionRequest, hooks taskLLMStreamHooks) iter.Seq2[llm.StreamEvent, error] {
@@ -435,6 +530,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
window := workerRuntime.CompactionWindow() window := workerRuntime.CompactionWindow()
wk := agent.NewWorker(workerRuntime, "task-router", s.m.dir, tx, window, s.agentMaxTurns("worker")) wk := agent.NewWorker(workerRuntime, "task-router", s.m.dir, tx, window, s.agentMaxTurns("worker"))
wk.SetCompactionWindowResolver(workerRuntime.CompactionWindow) wk.SetCompactionWindowResolver(workerRuntime.CompactionWindow)
wk.SetNonStreaming(workerRuntime.nonStreaming) // 按任务当前激活 profile 的流式开关(每轮读)
wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second) wk.SetRunTimeout(time.Duration(s.agentRunSeconds("worker")) * time.Second)
wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) wk.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory"))) wk.SetMemory(memory.NewStore(filepath.Join(s.m.dir, "memory")))
@@ -442,6 +538,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开) wk.SetConstraintInject(s.constraintInjectWorker) // 操作约束注入 worker(可配置,默认开)
pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner")) pl := agent.NewPlanner(plannerRuntime, "task-router", s.m.dir, tx, plannerRuntime.CompactionWindow(), s.agentMaxTurns("planner"))
pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow) pl.SetCompactionWindowResolver(plannerRuntime.CompactionWindow)
pl.SetNonStreaming(plannerRuntime.nonStreaming)
pl.SetKillWork(s.engine.KillWork) pl.SetKillWork(s.engine.KillWork)
pl.SetSteerWork(s.engine.SteerWork) pl.SetSteerWork(s.engine.SteerWork)
pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) pl.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
@@ -449,6 +546,7 @@ func (s *Server) agentsForTask(t *Task) *taskAgentBundle {
pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开) pl.SetConstraintInject(s.constraintInjectPlanner) // 操作约束注入 planner(可配置,默认开)
main := agent.NewMainAgent(mainRuntime, "task-router", s.m.dir, tx, mainRuntime.CompactionWindow(), s.agentMaxTurns("mainagent")) main := agent.NewMainAgent(mainRuntime, "task-router", s.m.dir, tx, mainRuntime.CompactionWindow(), s.agentMaxTurns("mainagent"))
main.SetCompactionWindowResolver(mainRuntime.CompactionWindow) main.SetCompactionWindowResolver(mainRuntime.CompactionWindow)
main.SetNonStreaming(mainRuntime.nonStreaming)
main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert()) main.SetProxy(s.m.ProxyAddr(), s.m.ProxyCACert())
main.SetWebSearch(s.webSearchFor("mainagent")) main.SetWebSearch(s.webSearchFor("mainagent"))
main.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏 main.SetSteerWork(s.engine.SteerWork) // steer_work:人对运行中 work 实时纠偏
+21
View File
@@ -37,6 +37,23 @@ func (p *scriptedLLMProvider) Stream(context.Context, llm.CompletionRequest) ite
} }
} }
func (p *scriptedLLMProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
return accumulateStreamForTest(ctx, p.Stream, req)
}
// accumulateStreamForTest drains a mock's Stream to satisfy the non-streaming
// Complete method.
func accumulateStreamForTest(ctx context.Context, stream func(context.Context, llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error], req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
acc := llm.NewAccumulator()
for ev, err := range stream(ctx, req) {
if err != nil {
return llm.Message{}, "", llm.Usage{}, err
}
acc.Add(ev)
}
return acc.Message(), acc.StopReason, acc.Usage, nil
}
// withZeroRetryBackoff sets the same-provider retry backoff to zero for the // withZeroRetryBackoff sets the same-provider retry backoff to zero for the
// duration of a (serial) test and returns a restore func for defer. // duration of a (serial) test and returns a restore func for defer.
func withZeroRetryBackoff() func() { func withZeroRetryBackoff() func() {
@@ -73,6 +90,10 @@ func (p *flakyThenOKProvider) Stream(context.Context, llm.CompletionRequest) ite
} }
} }
func (p *flakyThenOKProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
return accumulateStreamForTest(ctx, p.Stream, req)
}
func collectTaskLLMStream(seq iter.Seq2[llm.StreamEvent, error]) ([]llm.StreamEvent, error) { func collectTaskLLMStream(seq iter.Seq2[llm.StreamEvent, error]) ([]llm.StreamEvent, error) {
var events []llm.StreamEvent var events []llm.StreamEvent
var streamErr error var streamErr error
+91 -2
View File
@@ -2,17 +2,18 @@
import * as React from "react"; import * as React from "react";
import { ChevronLeftIcon, ChevronRightIcon, Loader2Icon, SearchIcon, TerminalIcon } from "lucide-react"; import { BarChart3Icon, ChevronLeftIcon, ChevronRightIcon, Loader2Icon, SearchIcon, TerminalIcon } from "lucide-react";
import { Badge } from "@/components/ui/badge"; import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button"; import { Button } from "@/components/ui/button";
import { Card } from "@/components/ui/card"; import { Card } from "@/components/ui/card";
import { Dialog, DialogContent, DialogDescription, DialogHeader, DialogTitle } from "@/components/ui/dialog";
import { Input } from "@/components/ui/input"; import { Input } from "@/components/ui/input";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet"; import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@/components/ui/sheet";
import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table"; import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table";
import { api } from "@/lib/api"; import { api } from "@/lib/api";
import type { CommandRecord } from "@/lib/types"; import type { CommandRecord, ToolStat } from "@/lib/types";
import { cn } from "@/lib/utils"; import { cn } from "@/lib/utils";
function fmtTime(ts: string) { function fmtTime(ts: string) {
@@ -59,6 +60,11 @@ export default function CommandsPage() {
const [total, setTotal] = React.useState(0); const [total, setTotal] = React.useState(0);
const [loading, setLoading] = React.useState(false); const [loading, setLoading] = React.useState(false);
// 各工具调用次数。弹窗打开时才拉取(多一次聚合查询,不必每次翻页都付)。
const [statsOpen, setStatsOpen] = React.useState(false);
const [stats, setStats] = React.useState<ToolStat[]>([]);
const [statsLoading, setStatsLoading] = React.useState(false);
// Selected execution is rendered in a right-side detail sheet. // Selected execution is rendered in a right-side detail sheet.
const [selected, setSelected] = React.useState<CommandRecord | null>(null); const [selected, setSelected] = React.useState<CommandRecord | null>(null);
@@ -96,6 +102,24 @@ export default function CommandsPage() {
}; };
}, [page, size, queryQ, taskFilter]); }, [page, size, queryQ, taskFilter]);
// 统计跟随筛选条件走,和表格描述的是同一批记录(但不分页)。
React.useEffect(() => {
if (!statsOpen) return;
let alive = true;
setStatsLoading(true);
api
.commandStats({ task: taskFilter || undefined, q: queryQ || undefined })
.then((r) => alive && setStats(r.stats ?? []))
.catch(() => alive && setStats([]))
.finally(() => alive && setStatsLoading(false));
return () => {
alive = false;
};
}, [statsOpen, queryQ, taskFilter]);
const statsTotal = stats.reduce((n, s) => n + s.total, 0);
const statsErrors = stats.reduce((n, s) => n + s.errors, 0);
const statsMax = stats.reduce((n, s) => Math.max(n, s.total), 0);
const totalPages = Math.max(1, Math.ceil(total / size)); const totalPages = Math.max(1, Math.ceil(total / size));
const rangeStart = total === 0 ? 0 : page * size + 1; const rangeStart = total === 0 ? 0 : page * size + 1;
const rangeEnd = page * size + commands.length; const rangeEnd = page * size + commands.length;
@@ -141,6 +165,11 @@ export default function CommandsPage() {
</SelectContent> </SelectContent>
</Select> </Select>
<Button variant="outline" size="sm" className="h-8" onClick={() => setStatsOpen(true)}>
<BarChart3Icon className="size-4" />
统计
</Button>
<div className="ml-auto flex items-center gap-2 text-xs text-muted-foreground"> <div className="ml-auto flex items-center gap-2 text-xs text-muted-foreground">
<span className="tabular-nums"> <span className="tabular-nums">
{rangeStart}–{rangeEnd} / {total} {rangeStart}–{rangeEnd} / {total}
@@ -243,6 +272,66 @@ export default function CommandsPage() {
</Card> </Card>
</div> </div>
{/* 工具调用统计:与表格同一批记录(同筛选、不分页) */}
<Dialog open={statsOpen} onOpenChange={setStatsOpen}>
<DialogContent className="sm:max-w-lg">
<DialogHeader>
<DialogTitle>工具调用统计</DialogTitle>
<DialogDescription>
{taskFilter || queryQ ? "当前筛选条件下的全部记录" : "全部工具执行记录"}
{stats.length > 0 && (
<>
{" · "}
<span className="tabular-nums">{stats.length}</span> 个工具 ·{" "}
<span className="tabular-nums">{statsTotal}</span> 次调用
{statsErrors > 0 && (
<>
{" · 失败 "}
<span className="tabular-nums text-red-600 dark:text-red-400">{statsErrors}</span>
</>
)}
</>
)}
</DialogDescription>
</DialogHeader>
{statsLoading && stats.length === 0 ? (
<div className="py-10 text-center">
<Loader2Icon className="mx-auto h-5 w-5 animate-spin text-muted-foreground" />
</div>
) : stats.length === 0 ? (
<div className="py-10 text-center text-sm text-muted-foreground">暂无统计数据</div>
) : (
<div className="-mr-2 max-h-[55vh] space-y-1 overflow-auto pr-2">
{stats.map((s) => (
<div key={s.tool} className="grid grid-cols-[minmax(0,1fr)_auto] items-center gap-3 rounded-md p-2">
<div className="min-w-0">
<div className="flex items-center gap-2">
<span className="truncate font-mono text-xs font-medium">{s.tool}</span>
{s.errors > 0 && (
<span className="text-[11px] tabular-nums text-red-600 dark:text-red-400">失败 {s.errors}</span>
)}
</div>
<div className="mt-1.5 h-1.5 overflow-hidden rounded-full bg-muted">
<div
className="h-full rounded-full bg-primary"
style={{ width: `${statsMax > 0 ? (s.total / statsMax) * 100 : 0}%` }}
/>
</div>
</div>
<div className="text-right">
<div className="text-xs font-semibold tabular-nums">{s.total}</div>
<div className="text-[11px] tabular-nums text-muted-foreground">
{statsTotal > 0 ? ((s.total / statsTotal) * 100).toFixed(1) : "0.0"}%
</div>
</div>
</div>
))}
</div>
)}
</DialogContent>
</Dialog>
<Sheet open={selected !== null} onOpenChange={(open) => !open && setSelected(null)}> <Sheet open={selected !== null} onOpenChange={(open) => !open && setSelected(null)}>
<SheetContent className="w-full! max-w-none! gap-0 p-0 sm:w-[46rem]! sm:max-w-[46rem]!"> <SheetContent className="w-full! max-w-none! gap-0 p-0 sm:w-[46rem]! sm:max-w-[46rem]!">
{selected && ( {selected && (
+36 -7
View File
@@ -309,6 +309,7 @@ function ProfileSheet({
const [effort, setEffort] = React.useState(NONE); const [effort, setEffort] = React.useState(NONE);
const [priority, setPriority] = React.useState("0"); // 轮询顺位;越大越先 const [priority, setPriority] = React.useState("0"); // 轮询顺位;越大越先
const [poolExclude, setPoolExclude] = React.useState(false); const [poolExclude, setPoolExclude] = React.useState(false);
const [streaming, setStreaming] = React.useState(true); // true=流式(默认);false=非流式
const [testing, setTesting] = React.useState(false); const [testing, setTesting] = React.useState(false);
const [saving, setSaving] = React.useState(false); const [saving, setSaving] = React.useState(false);
const [models, setModels] = React.useState<string[]>([]); const [models, setModels] = React.useState<string[]>([]);
@@ -331,6 +332,7 @@ function ProfileSheet({
setEffort(fromStore(profile?.reasoning_effort)); setEffort(fromStore(profile?.reasoning_effort));
setPriority(String(profile?.priority ?? 0)); setPriority(String(profile?.priority ?? 0));
setPoolExclude(profile?.pool_exclude ?? false); setPoolExclude(profile?.pool_exclude ?? false);
setStreaming(profile?.streaming ?? true);
setApiKey(""); setApiKey("");
setKeyHint(profile?.api_key_hint ?? ""); setKeyHint(profile?.api_key_hint ?? "");
setModels([]); setModels([]);
@@ -365,8 +367,22 @@ function ProfileSheet({
try { try {
// 用配置实际会跑的思考参数来测,这样不支持该字段的模型在这里就失败, // 用配置实际会跑的思考参数来测,这样不支持该字段的模型在这里就失败,
// 而不是等到跑任务时才炸。传 profile id:Key 输入框留空时用已存的 Key。 // 而不是等到跑任务时才炸。传 profile id:Key 输入框留空时用已存的 Key。
const r = await api.testLLM(format, model, baseUrl, apiKey, proxy, toStore(thinkingType), toStore(effort), profileId); const r = await api.testLLM(
if (r.ok) toast.success(`连接成功 · ${r.latency_ms ?? "?"}ms · ${r.model ?? model}`); format,
model,
baseUrl,
apiKey,
proxy,
toStore(thinkingType),
toStore(effort),
profileId,
streaming,
);
// 回复内容一并展示:看得见模型确实说了话,才算和会话里跑通是一回事。
if (r.ok)
toast.success(`连接成功 · ${r.latency_ms ?? "?"}ms · ${r.model ?? model}`, {
description: r.reply ? `回复:${r.reply}` : undefined,
});
else toast.error(`连接失败:${r.error ?? "未知"}`); else toast.error(`连接失败:${r.error ?? "未知"}`);
} catch (e) { } catch (e) {
toast.error(`测试出错:${(e as Error).message}`); toast.error(`测试出错:${(e as Error).message}`);
@@ -398,6 +414,7 @@ function ProfileSheet({
reasoning_effort: toStore(effort), reasoning_effort: toStore(effort),
priority: Number(priority) || 0, priority: Number(priority) || 0,
pool_exclude: poolExclude, pool_exclude: poolExclude,
streaming,
}); });
if (isNew) toast.success(`已新建:${name.trim()}(在卡片上「设为激活」以启用)`); if (isNew) toast.success(`已新建:${name.trim()}(在卡片上「设为激活」以启用)`);
else toast.success(profile?.is_default ? "已保存,激活配置即时生效,无需重启" : "已保存"); else toast.success(profile?.is_default ? "已保存,激活配置即时生效,无需重启" : "已保存");
@@ -525,7 +542,8 @@ function ProfileSheet({
onChange={(e) => setProxy(e.target.value)} onChange={(e) => setProxy(e.target.value)}
/> />
<p className="text-muted-foreground text-xs"> <p className="text-muted-foreground text-xs">
仅 LLM 出站请求走此代理,支持 http/https/socks5,可带账号密码(如 socks5://user:pass@host:port,密码含特殊字符需 URL 编码);留空表示不使用代理(直连)。 仅 LLM 出站请求走此代理,支持 http/https/socks5,可带账号密码(如
socks5://user:pass@host:port,密码含特殊字符需 URL 编码);留空表示不使用代理(直连)。
</p> </p>
</div> </div>
@@ -595,6 +613,17 @@ function ProfileSheet({
</div> </div>
<Switch checked={poolExclude} onCheckedChange={setPoolExclude} aria-label="不参与轮询" /> <Switch checked={poolExclude} onCheckedChange={setPoolExclude} aria-label="不参与轮询" />
</div> </div>
<div className="flex items-center justify-between gap-4 border-t pt-3">
<div className="grid gap-0.5">
<Label className="text-sm">流式输出 · streaming</Label>
<p className="text-muted-foreground text-xs">
开启(默认)走流式 SSE,有运行中实时进度与实时 token 计数。 关闭则走真·非流式(stream:false,
一次性返回完整响应)——可绕开部分网关糟糕的 SSE 实现(空帧 / 思考字段丢帧),
代价是失去运行中的实时进度。
</p>
</div>
<Switch checked={streaming} onCheckedChange={setStreaming} aria-label="流式输出" />
</div>
</div> </div>
<div className="grid gap-3 rounded-lg border p-3"> <div className="grid gap-3 rounded-lg border p-3">
@@ -602,8 +631,8 @@ function ProfileSheet({
<div className="grid gap-0.5"> <div className="grid gap-0.5">
<Label className="text-sm">思考开关 · thinking.type</Label> <Label className="text-sm">思考开关 · thinking.type</Label>
<p className="text-muted-foreground text-xs"> <p className="text-muted-foreground text-xs">
控制是否发送 thinking 字段。不发送=不带该字段(兼容 MiniMax 等不支持 控制是否发送 thinking 字段。不发送=不带该字段(兼容 MiniMax 等不支持 的模型);关闭=发
的模型);关闭=发 disabled;开启=发 enabled。与下面的强度互相独立。 disabled;开启=发 enabled。与下面的强度互相独立。
</p> </p>
</div> </div>
<Select value={thinkingType} onValueChange={setThinkingType}> <Select value={thinkingType} onValueChange={setThinkingType}>
@@ -623,8 +652,8 @@ function ProfileSheet({
<div className="grid gap-0.5"> <div className="grid gap-0.5">
<Label className="text-sm">思考强度 · reasoning_effort</Label> <Label className="text-sm">思考强度 · reasoning_effort</Label>
<p className="text-muted-foreground text-xs"> <p className="text-muted-foreground text-xs">
独立的强度档位(OpenAI reasoning_effort / Anthropic output_config.effort)。 独立的强度档位(OpenAI reasoning_effort / Anthropic output_config.effort)。 有些接口没有 thinking
有些接口没有 thinking 字段、只靠强度即可激活思考,故可单独设置、不发送思考开关。 字段、只靠强度即可激活思考,故可单独设置、不发送思考开关。
</p> </p>
</div> </div>
<Select value={effort} onValueChange={setEffort}> <Select value={effort} onValueChange={setEffort}>
+13 -1
View File
@@ -74,6 +74,7 @@ import type {
TokenTotal, TokenTotal,
TokenUsage, TokenUsage,
Tool, Tool,
ToolStat,
TrafficDetail, TrafficDetail,
TrafficHost, TrafficHost,
TrafficResp, TrafficResp,
@@ -677,8 +678,10 @@ export const api = {
thinking_type = "", thinking_type = "",
reasoning_effort = "", reasoning_effort = "",
profile_id?: number, profile_id?: number,
streaming = true, // 用该配置真实的收发模式来测,别让"流式能通、非流式不通"漏到会话里
) => ) =>
post<{ ok: boolean; error?: string; latency_ms?: number; model?: string }>("/llm/test", { // reply = 模型实际回复(已截断);一个字都不回的配置后端直接判失败
post<{ ok: boolean; error?: string; latency_ms?: number; model?: string; reply?: string }>("/llm/test", {
provider, provider,
model, model,
base_url, base_url,
@@ -687,6 +690,7 @@ export const api = {
thinking_type, thinking_type,
reasoning_effort, reasoning_effort,
profile_id, profile_id,
streaming,
}), }),
llmProfiles: () => get<{ profiles: LLMProfile[] }>("/llm/profiles").then((r) => arr(r.profiles)), llmProfiles: () => get<{ profiles: LLMProfile[] }>("/llm/profiles").then((r) => arr(r.profiles)),
saveLLMProfile: (p: { saveLLMProfile: (p: {
@@ -704,6 +708,7 @@ export const api = {
reasoning_effort?: string; // ""(不发送)|"low"|"medium"|"high"|"xhigh"|"max" reasoning_effort?: string; // ""(不发送)|"low"|"medium"|"high"|"xhigh"|"max"
priority?: number; // 轮询顺位,越大越先用 priority?: number; // 轮询顺位,越大越先用
pool_exclude?: boolean; // true=不作为故障转移目标 pool_exclude?: boolean; // true=不作为故障转移目标
streaming?: boolean; // true(默认)=流式 | false=非流式
}) => post<{ id: number }>("/llm/profiles", p), }) => post<{ id: number }>("/llm/profiles", p),
deleteLLMProfile: (id: string) => del<{ deleted: number }>(`/llm/profiles/${id}`), deleteLLMProfile: (id: string) => del<{ deleted: number }>(`/llm/profiles/${id}`),
activateLLMProfile: (id: string) => post<{ ok: boolean }>("/llm/profiles/active", { id: Number(id) }), activateLLMProfile: (id: string) => post<{ ok: boolean }>("/llm/profiles/active", { id: Number(id) }),
@@ -972,6 +977,13 @@ export const api = {
sp.set("size", String(params?.size ?? 50)); sp.set("size", String(params?.size ?? 50));
return get<{ commands: CommandRecord[]; total: number }>(`/commands?${sp}`); return get<{ commands: CommandRecord[]; total: number }>(`/commands?${sp}`);
}, },
// 各工具调用次数;沿用列表的 task/q 筛选,统计的是整个结果集而非当前页。
commandStats: (params?: { task?: string; q?: string }) => {
const sp = new URLSearchParams();
if (params?.task) sp.set("task", params.task);
if (params?.q) sp.set("q", params.q);
return get<{ stats: ToolStat[] }>(`/commands/stats?${sp}`);
},
// ---- LLM records ---- // ---- LLM records ----
llmRecords: (params?: { model?: string; session?: string; task?: string; page?: number; size?: number }) => { llmRecords: (params?: { model?: string; session?: string; task?: string; page?: number; size?: number }) => {
+13 -1
View File
@@ -1493,6 +1493,17 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re
// ── 工具执行历史 ── // ── 工具执行历史 ──
if (path === "/commands" && m === "GET") return { commands: D.commandRecords, total: D.commandRecords.length }; if (path === "/commands" && m === "GET") return { commands: D.commandRecords, total: D.commandRecords.length };
if (path === "/commands/stats" && m === "GET") {
const tally = new Map<string, { tool: string; total: number; errors: number }>();
for (const c of D.commandRecords) {
const tool = c.tool || "-";
const s = tally.get(tool) ?? { tool, total: 0, errors: 0 };
s.total++;
if (c.is_error) s.errors++;
tally.set(tool, s);
}
return { stats: [...tally.values()].sort((a, b) => b.total - a.total || a.tool.localeCompare(b.tool)) };
}
// ── LLM ── // ── LLM ──
if (path === "/llm/records" && m === "GET") return { records: mockLLMRecords, total: mockLLMRecords.length }; if (path === "/llm/records" && m === "GET") return { records: mockLLMRecords, total: mockLLMRecords.length };
@@ -1509,7 +1520,8 @@ function route(m: string, path: string, seg: string[], q: URLSearchParams, b: Re
} }
if (path === "/llm" && m === "GET") return D.llmConfig; if (path === "/llm" && m === "GET") return D.llmConfig;
if (path === "/llm" && m === "POST") return { ok: true }; if (path === "/llm" && m === "POST") return { ok: true };
if (path === "/llm/test") return { ok: true, latency_ms: 128, model: String(b.model ?? "claude-opus-4-8") }; if (path === "/llm/test")
return { ok: true, latency_ms: 128, model: String(b.model ?? "claude-opus-4-8"), reply: "OK" };
if (path === "/llm/profiles" && m === "GET") return { profiles: D.llmProfiles }; if (path === "/llm/profiles" && m === "GET") return { profiles: D.llmProfiles };
if (path === "/llm/profiles" && m === "POST") return { id: Number(b.id) || 3 }; if (path === "/llm/profiles" && m === "POST") return { id: Number(b.id) || 3 };
if (path === "/llm/profiles/active") return { ok: true }; if (path === "/llm/profiles/active") return { ok: true };
+9
View File
@@ -869,6 +869,8 @@ export interface LLMProfile {
priority?: number; priority?: number;
// true = 不作为故障转移目标(仍可被 agent/任务显式绑定使用)。 // true = 不作为故障转移目标(仍可被 agent/任务显式绑定使用)。
pool_exclude?: boolean; pool_exclude?: boolean;
// true(默认)= 流式(SSE) | false = 真·非流式(stream:false,一次性返回)。
streaming?: boolean;
} }
// ---- LLM 轮询(故障转移)---- // ---- LLM 轮询(故障转移)----
@@ -1133,6 +1135,13 @@ export interface CommandRecord {
created_at: string; created_at: string;
} }
// 单个工具的调用统计(/commands/stats);errors 为其中失败的次数。
export interface ToolStat {
tool: string;
total: number;
errors: number;
}
// ---- LLM recording ---- // ---- LLM recording ----
export interface LLMRecordItem { export interface LLMRecordItem {
id: number; id: number;