Agent-System-Backend: review fixes for chat-process-split-r1 (symlink, live interleaving, artifact process) #21

Merged
weichen merged 4 commits from feature/chat-process-split-r1 into main 2026-09-30 16:15:19 +00:00
2 changed files with 242 additions and 4 deletions
+12 -4
View File
@@ -19,6 +19,8 @@ import (
type LLMRunner struct {
// Prompts 提示词库(nil = 回退画像 SystemPrompt 单段,兼容测试与骨架)。
Prompts *prompts.Service
// Stream 提供方边界注入点(nil = 按模型路由构造 agent-native 客户端)。
Stream agentllm.StreamFunc
}
func (l *LLMRunner) Name() string { return "llm" }
@@ -115,7 +117,11 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
return fmt.Errorf("llm: 主模型未配置(当前解析:model=%q base=%q;检查 settings.models.chat 与画像 ModelRoute;演示模式可开 runtime.mock)", route.Model, route.Base)
}
client := &agentllm.Client{BaseURL: route.Base, APIKey: route.Key}
stream := l.Stream
if stream == nil {
client := &agentllm.Client{BaseURL: route.Base, APIKey: route.Key}
stream = client.Stream
}
// 知识注入:当前消息包含 @knowledge 引用时检索注入(最小实现:显式引用)。
var knowledge []core.RetrievalHit
@@ -192,7 +198,7 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
var turnThinking, turnText strings.Builder
var calls []agentllm.ToolCall
var usage agentllm.Usage
ev := client.Stream(rc.Context(), route.Model, msgs, decls, streamOpts)
ev := stream(rc.Context(), route.Model, msgs, decls, streamOpts)
phaseEmitted := false
for e := range ev {
switch e.Kind {
@@ -225,7 +231,8 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
thinkingIdx++
}
if turnText.Len() > 0 {
finalParts = append(finalParts, core.Part{Type: core.PartText, Text: turnText.String()})
// 工具轮的文本是过程解说(终态随过程折叠);终轮文本才是对用户的回复。
finalParts = append(finalParts, core.Part{Type: core.PartText, Text: turnText.String(), Process: len(calls) > 0})
textIdx++
}
@@ -318,7 +325,8 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: call.ID, Status: statusOf(err), Summary: summary}},
)
if artifactRef != nil {
finalParts = append(finalParts, core.Part{Type: core.PartArtifactRef, Artifact: artifactRef})
// 工件引用与工具调用同轮产生,同标过程(否则终态折叠时残留在回复区)。
finalParts = append(finalParts, core.Part{Type: core.PartArtifactRef, Process: true, Artifact: artifactRef})
}
// 回灌工具结果(截断保护)。
if len(out) > 16000 {
+230
View File
@@ -0,0 +1,230 @@
package runtime
import (
"context"
"fmt"
"testing"
agentllm "agent-native/runtime/llm"
core "agent-system-core"
)
type emittedEvent struct {
typ core.StreamEventType
payload any
}
// fakeRunCtx:RunCtx 最小实现——只收集事件与终态部件,其余 no-op。
type fakeRunCtx struct {
ctx context.Context
prof core.AgentProfile
events []emittedEvent
assistant []core.Part
execArtifact *core.Artifact // ExecuteTool 返回的工件(非 nil 时走工件路径)
}
func (f *fakeRunCtx) Context() context.Context { return f.ctx }
func (f *fakeRunCtx) Run() core.Run { return core.Run{ID: "run_t"} }
func (f *fakeRunCtx) Profile() core.AgentProfile { return f.prof }
func (f *fakeRunCtx) Settings() core.EffectiveSettings {
return core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"baseUrl": "http://test"}}}
}
func (f *fakeRunCtx) Emit(typ core.StreamEventType, payload any) error {
f.events = append(f.events, emittedEvent{typ, payload})
return nil
}
func (f *fakeRunCtx) SetAssistantParts(parts []core.Part) error { f.assistant = parts; return nil }
func (f *fakeRunCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) {
return core.HITLResponse{}, fmt.Errorf("not implemented")
}
func (f *fakeRunCtx) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) {
return core.AgentResultPayload{}, fmt.Errorf("not implemented")
}
func (f *fakeRunCtx) UpdatePlan(p core.PlanSnapshot) error { return nil }
func (f *fakeRunCtx) UpdateTodos(l core.TodoList) error { return nil }
func (f *fakeRunCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
return core.Artifact{}, fmt.Errorf("not implemented")
}
func (f *fakeRunCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) { return nil, nil }
func (f *fakeRunCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
res := core.ToolCallResult{Status: "ok", Output: "done"}
if f.execArtifact != nil {
res.Artifact = f.execArtifact
}
return res
}
func (f *fakeRunCtx) History() []core.Message {
return []core.Message{{ID: "m1", Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}}}
}
func (f *fakeRunCtx) Attachments() []core.ParsedDocument { return nil }
func (f *fakeRunCtx) Tools() []core.ToolDesc { return nil }
func (f *fakeRunCtx) Subtask() string { return "" }
func (f *fakeRunCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
return card, nil
}
func (f *fakeRunCtx) SetCardStatus(id string, status core.CardStatus, tone, note string) error {
return nil
}
func (f *fakeRunCtx) SetContextObservation(o core.ContextObservation) {}
// scriptStream:每次调用回放一轮(真实客户端语义:一个 message_end 后关通道)。
// 轮数耗尽后再调用返回空流(防御;正常终轮后 runner 不再调用)。
func scriptStream(turns [][]agentllm.Event) agentllm.StreamFunc {
n := 0
return func(ctx context.Context, model string, msgs []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event {
out := make(chan agentllm.Event)
var evs []agentllm.Event
if n < len(turns) {
evs = turns[n]
n++
}
go func() {
defer close(out)
for _, e := range evs {
select {
case out <- e:
case <-ctx.Done():
return
}
}
}()
return out
}
}
// TestLLMRunnerToolTurnTextIsProcess 验证:工具轮的文本是过程解说(Process=true),
// 终轮文本(无工具调用)才是对用户的回复(无标记)——中栏过程/回复分离的数据基础。
func TestLLMRunnerToolTurnTextIsProcess(t *testing.T) {
turns := [][]agentllm.Event{
{ // 第 1 轮:思考 + 解说文本 + 工具调用。
agentllm.Event{Kind: agentllm.KindReasoningDelta, ReasoningDelta: "先想想"},
agentllm.Event{Kind: agentllm.KindTextDelta, Text: "我先更新待办。"},
agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{ToolCalls: []agentllm.ToolCall{
{ID: "c1", Name: "update_todos", Arguments: "{}"},
}}},
},
{ // 第 2 轮(终轮):最终回复文本。
agentllm.Event{Kind: agentllm.KindTextDelta, Text: "完成了。"},
agentllm.Event{Kind: agentllm.KindMessageEnd},
},
}
r := &LLMRunner{Stream: scriptStream(turns)}
rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}}
if err := r.Execute(rc); err != nil {
t.Fatalf("Execute: %v", err)
}
type want struct {
typ core.PartType
text string
process bool
}
got := make([]want, 0, len(rc.assistant))
for _, p := range rc.assistant {
got = append(got, want{p.Type, p.Text, p.Process})
}
expected := []want{
{core.PartThinking, "先想想", true},
{core.PartText, "我先更新待办。", true}, // 工具轮文本 = 过程
{core.PartToolCall, "", true},
{core.PartToolResult, "", true},
{core.PartText, "完成了。", false}, // 终轮文本 = 回复
}
if len(got) != len(expected) {
t.Fatalf("parts 数量不符:got %d (%+v) want %d", len(got), got, len(expected))
}
for i, g := range got {
if g != expected[i] {
t.Errorf("parts[%d] = %+v, want %+v", i, g, expected[i])
}
}
}
// TestLLMRunnerPhaseEventsPerTurn 验证:每轮首个文本增量发 responding、工具轮开始发
// acting——前端「响应中折叠过程」的驱动信号(交错轮不粘滞)。
func TestLLMRunnerPhaseEventsPerTurn(t *testing.T) {
turns := [][]agentllm.Event{
{
agentllm.Event{Kind: agentllm.KindTextDelta, Text: "解说"},
agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{ToolCalls: []agentllm.ToolCall{
{ID: "c1", Name: "update_todos"},
}}},
},
{
agentllm.Event{Kind: agentllm.KindTextDelta, Text: "最终回复"},
agentllm.Event{Kind: agentllm.KindMessageEnd},
},
}
r := &LLMRunner{Stream: scriptStream(turns)}
rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}}
if err := r.Execute(rc); err != nil {
t.Fatalf("Execute: %v", err)
}
var phases []string
for _, e := range rc.events {
if e.typ == core.EvRunPhase {
p, ok := e.payload.(core.RunPhasePayload)
if !ok {
t.Fatalf("run.phase payload 类型错误: %T", e.payload)
}
phases = append(phases, string(p.Phase))
}
}
// 第 1 轮 responding(文本增量)→ acting(工具轮);第 2 轮再次 responding。
wantPhases := []string{"responding", "acting", "responding"}
if len(phases) != len(wantPhases) {
t.Fatalf("phase 序列 = %v, want %v", phases, wantPhases)
}
for i := range phases {
if phases[i] != wantPhases[i] {
t.Fatalf("phase 序列 = %v, want %v", phases, wantPhases)
}
}
}
// TestLLMRunnerArtifactPartIsProcess 验证:工具产生的工件引用部件同样标记为过程,
// 终态随过程折叠(不残留在回复区)。
func TestLLMRunnerArtifactPartIsProcess(t *testing.T) {
turns := [][]agentllm.Event{
{ // 第 1 轮:解说文本 + 工具调用(产生工件)。
agentllm.Event{Kind: agentllm.KindTextDelta, Text: "我生成一份报表。"},
agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{ToolCalls: []agentllm.ToolCall{
{ID: "c1", Name: "make_report", Arguments: "{}"},
}}},
},
{ // 第 2 轮(终轮):最终回复文本。
agentllm.Event{Kind: agentllm.KindTextDelta, Text: "见上方工件。"},
agentllm.Event{Kind: agentllm.KindMessageEnd},
},
}
r := &LLMRunner{Stream: scriptStream(turns)}
rc := &fakeRunCtx{
ctx: context.Background(),
prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}},
execArtifact: &core.Artifact{ID: "a1", Name: "report.csv", Kind: "csv"},
}
if err := r.Execute(rc); err != nil {
t.Fatalf("Execute: %v", err)
}
found := false
for _, p := range rc.assistant {
if p.Type != core.PartArtifactRef {
continue
}
found = true
if !p.Process {
t.Errorf("artifact_ref 部件未标记 Process: %+v", p)
}
if p.Artifact == nil || p.Artifact.ArtifactID != "a1" {
t.Errorf("artifact 引用不符: %+v", p.Artifact)
}
}
if !found {
t.Fatalf("未找到 artifact_ref 部件: %+v", rc.assistant)
}
}