feat(loop): agent loop（8 轮 + checkpoint + 流式 + 记忆注入/提取 + trace + 工具超时回喂）
This commit is contained in:
Aska committed
1 parent
01a4e29f84
commit
71697ca4f8
12 files changed
+1292
No files matched your search
@@ -0,0 +1,46 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
|
||||
)
|
||||
|
||||
// checkpointState is the durable snapshot persisted at every step. Carrying the
|
||||
// message window and the step index lets a long-running task be resumed from the
|
||||
// latest checkpoint without losing context (design §1 durable execution).
|
||||
type checkpointState struct {
|
||||
Step int `json:"step"`
|
||||
Messages []model.Message `json:"messages"`
|
||||
StopReason string `json:"stop_reason,omitempty"`
|
||||
}
|
||||
|
||||
// saveCheckpoint persists a snapshot at step. It never fails the run: a nil store
|
||||
// is a no-op and a marshal/save error is logged.
|
||||
func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, messages []model.Message, stopReason string) {
|
||||
if l.checkpoint == nil {
|
||||
return
|
||||
}
|
||||
state := checkpointState{
|
||||
Step: step,
|
||||
Messages: messages,
|
||||
StopReason: stopReason,
|
||||
}
|
||||
buf, err := json.Marshal(state)
|
||||
if err != nil {
|
||||
l.log.Warn("checkpoint marshal", "step", step, "error", err)
|
||||
return
|
||||
}
|
||||
cp := &persist.Checkpoint{
|
||||
ID: fmt.Sprintf("%s:%d", sessionID, step),
|
||||
AgentID: sessionID,
|
||||
Step: step,
|
||||
State: buf,
|
||||
}
|
||||
if err := l.checkpoint.Save(ctx, cp); err != nil {
|
||||
l.log.Warn("checkpoint save", "step", step, "error", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
// Package loop implements the default agent execution engine of the Cyrene v2
|
||||
// runtime (DevDocs/design/04-运行时.md §1). It owns the loop that package agent
|
||||
// deliberately leaves out: assemble context from memory, drive the model for up
|
||||
// to MaxRounds steps, execute any requested tool calls under a per-invoke
|
||||
// timeout, feed tool results back to the model, persist a checkpoint at every
|
||||
// step, emit structured traces, deliver streamed deltas and run the
|
||||
// memory-extraction hook.
|
||||
//
|
||||
// The loop is configured with injected dependencies rather than concrete wiring,
|
||||
// so it stays leaf-testable with mocks (a mock adapter, mock guardrail and mock
|
||||
// tool, plus real in-memory memory/persist stores and a buffered trace recorder).
|
||||
package loop
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// Constants that bound the loop. These are compile-time so no magic number
|
||||
// appears in the run path.
|
||||
const (
|
||||
// DefaultMaxRounds is the tool-loop upper bound used when Options.MaxRounds
|
||||
// is non-positive. It matches the agent package's own default.
|
||||
DefaultMaxRounds = 8
|
||||
|
||||
// StopReasonMaxRounds reports that the loop exhausted its round budget
|
||||
// before reaching a final answer.
|
||||
StopReasonMaxRounds = "max_rounds"
|
||||
)
|
||||
|
||||
// Sentinel errors returned by the Loop. Use errors.Is to test for them; they are
|
||||
// returned directly or wrapped with %w so the underlying cause stays available.
|
||||
var (
|
||||
// ErrNilAgent is returned by Run when no *agent.Agent was configured.
|
||||
ErrNilAgent = errors.New("loop: nil agent")
|
||||
|
||||
// ErrNoAdapter is returned by Run (streaming path) when the agent was built
|
||||
// without a model Adapter.
|
||||
ErrNoAdapter = errors.New("loop: no model adapter")
|
||||
)
|
||||
|
||||
// Result is the outcome of a completed Run.
|
||||
type Result struct {
|
||||
// Text is the final assistant text, or empty when the loop exhausted its
|
||||
// budget without a final answer.
|
||||
Text string
|
||||
// Steps is the number of model steps executed.
|
||||
Steps int
|
||||
// StopReason classifies how the loop ended: an agent StopReason* value or
|
||||
// StopReasonMaxRounds.
|
||||
StopReason string
|
||||
// ToolCallsExecuted is the number of tool calls the loop invoked, including
|
||||
// those that failed.
|
||||
ToolCallsExecuted int
|
||||
}
|
||||
|
||||
// Loop is the default agent execution engine. It is configured once via New and
|
||||
// safe for concurrent Run calls as long as the injected dependencies (agent,
|
||||
// dispatcher, stores, recorder, tracer) are themselves concurrency-safe, which
|
||||
// they are by contract. Run keeps all loop state on the stack.
|
||||
type Loop struct {
|
||||
agent *agent.Agent
|
||||
tools *tool.Dispatcher
|
||||
memory *memory.Store
|
||||
retriever *memory.Retriever
|
||||
working *memory.WorkingMemory
|
||||
extractor memory.Extractor
|
||||
checkpoint *persist.Store
|
||||
tracer *trace.Tracer
|
||||
guards *guardrail.Pipeline
|
||||
personaID string
|
||||
maxRounds int
|
||||
onStream func(delta string)
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// New builds a Loop from opts, applying defaults. A non-positive MaxRounds
|
||||
// selects DefaultMaxRounds; a nil Log is replaced by a discard logger tagged with
|
||||
// source "runtime/loop". Every dependency may be nil and is tolerated until the
|
||||
// step that needs it (mirroring the run-time nil-tolerance of the underlying
|
||||
// packages); Run returns ErrNilAgent when no agent is configured.
|
||||
func New(opts Options) *Loop {
|
||||
maxRounds := opts.MaxRounds
|
||||
if maxRounds <= 0 {
|
||||
maxRounds = DefaultMaxRounds
|
||||
}
|
||||
return &Loop{
|
||||
agent: opts.Agent,
|
||||
tools: opts.Tools,
|
||||
memory: opts.Memory,
|
||||
retriever: opts.Retriever,
|
||||
working: opts.WorkingMemory,
|
||||
extractor: opts.Extractor,
|
||||
checkpoint: opts.Checkpoint,
|
||||
tracer: opts.Tracer,
|
||||
guards: opts.Guards,
|
||||
personaID: opts.PersonaID,
|
||||
maxRounds: maxRounds,
|
||||
onStream: opts.OnStream,
|
||||
log: sourcedLog(opts.Log),
|
||||
}
|
||||
}
|
||||
|
||||
// MaxRounds returns the tool-loop upper bound configured for the loop.
|
||||
func (l *Loop) MaxRounds() int { return l.maxRounds }
|
||||
|
||||
// Agent returns the injected agent.
|
||||
func (l *Loop) Agent() *agent.Agent { return l.agent }
|
||||
|
||||
// sourcedLog returns a logger tagged with the module source "runtime/loop",
|
||||
// replacing a nil logger with a discard one so the loop can always log safely.
|
||||
func sourcedLog(l *slog.Logger) *slog.Logger {
|
||||
if l == nil {
|
||||
l = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
}
|
||||
return logger.WithSource(l, "runtime/loop")
|
||||
}
|
||||
@@ -0,0 +1,375 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
func TestRun_NoTools_FinalText(t *testing.T) {
|
||||
t.Parallel()
|
||||
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Model: "m", Content: "Hello!"}}}}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess-1", []model.Message{model.TextMessage(model.RoleUser, "hi")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "Hello!" {
|
||||
t.Fatalf("Text = %q, want Hello!", res.Text)
|
||||
}
|
||||
if res.Steps != 1 {
|
||||
t.Fatalf("Steps = %d, want 1", res.Steps)
|
||||
}
|
||||
if res.StopReason != agent.StopReasonStop {
|
||||
t.Fatalf("StopReason = %q, want %q", res.StopReason, agent.StopReasonStop)
|
||||
}
|
||||
if res.ToolCallsExecuted != 0 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 0", res.ToolCallsExecuted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_ToolCallThenFinal_FeedsBack(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "Echoed!"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "echo this")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "Echoed!" {
|
||||
t.Fatalf("Text = %q, want Echoed!", res.Text)
|
||||
}
|
||||
if res.Steps != 2 {
|
||||
t.Fatalf("Steps = %d, want 2", res.Steps)
|
||||
}
|
||||
if res.StopReason != agent.StopReasonStop {
|
||||
t.Fatalf("StopReason = %q, want stop", res.StopReason)
|
||||
}
|
||||
if res.ToolCallsExecuted != 1 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 1", res.ToolCallsExecuted)
|
||||
}
|
||||
if len(ad.reqs) < 2 {
|
||||
t.Fatalf("adapter requests = %d, want >= 2", len(ad.reqs))
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1]
|
||||
got := last.Messages[len(last.Messages)-1]
|
||||
if got.Role != model.RoleTool {
|
||||
t.Fatalf("last message role = %q, want tool", got.Role)
|
||||
}
|
||||
if !strings.Contains(got.Content[0].Text, "hi") {
|
||||
t.Fatalf("tool result text = %q, want to contain 'hi'", got.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_MaxRounds(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: call}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "x"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 2})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Steps != 2 {
|
||||
t.Fatalf("Steps = %d, want 2", res.Steps)
|
||||
}
|
||||
if res.StopReason != StopReasonMaxRounds {
|
||||
t.Fatalf("StopReason = %q, want %q", res.StopReason, StopReasonMaxRounds)
|
||||
}
|
||||
if res.ToolCallsExecuted != 2 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 2", res.ToolCallsExecuted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_ToolError_FeedsBack(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"boom","args":{}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "recovered"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "boom", err: errors.New("kaboom")})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "boom now")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "recovered" {
|
||||
t.Fatalf("Text = %q, want recovered", res.Text)
|
||||
}
|
||||
if res.ToolCallsExecuted != 1 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 1", res.ToolCallsExecuted)
|
||||
}
|
||||
got := ad.reqs[len(ad.reqs)-1].Messages[len(ad.reqs[len(ad.reqs)-1].Messages)-1]
|
||||
if !strings.Contains(got.Content[0].Text, "kaboom") {
|
||||
t.Fatalf("tool result should carry the error, got %q", got.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_ToolTimeout_FeedsBack(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"slow","args":{}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "continued after timeout"}},
|
||||
}}
|
||||
td := dispatcherWithTimeout(t, 30*time.Millisecond, &mockTool{name: "slow", delay: 500 * time.Millisecond})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "slow it")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "continued after timeout" {
|
||||
t.Fatalf("Text = %q, want recovery text", res.Text)
|
||||
}
|
||||
got := ad.reqs[len(ad.reqs)-1].Messages[len(ad.reqs[len(ad.reqs)-1].Messages)-1]
|
||||
if !strings.Contains(got.Content[0].Text, "timed out") {
|
||||
t.Fatalf("tool result should report the timeout, got %q", got.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_CheckpointPerStep(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "done"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
cp := newTestCheckpointStore(t)
|
||||
l := New(Options{Agent: a, Checkpoint: cp, MaxRounds: 8})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
got, err := cp.List(context.Background(), "sess")
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("checkpoints = %d, want 2", len(got))
|
||||
}
|
||||
if got[0].Step != 1 || got[1].Step != 2 {
|
||||
t.Fatalf("checkpoint steps = [%d,%d], want [1,2]", got[0].Step, got[1].Step)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_TraceSpans(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"hi"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "done"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
var buf bytes.Buffer
|
||||
tr := trace.New(trace.NewJSONRecorder(&buf))
|
||||
l := New(Options{Agent: a, Tracer: tr, MaxRounds: 8})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
saw := map[trace.SpanKind]bool{}
|
||||
for _, r := range decodeSpanLines(t, buf.Bytes()) {
|
||||
saw[r.Kind] = true
|
||||
}
|
||||
for _, k := range []trace.SpanKind{trace.SpanKindLLM, trace.SpanKindTool, trace.SpanKindStateTransition} {
|
||||
if !saw[k] {
|
||||
t.Fatalf("missing span kind %q in %+v", k, saw)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_MemoryInject(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestMemoryStore(t)
|
||||
if _, err := store.Add(context.Background(), memory.MemoryItem{
|
||||
PersonaID: "persona-a", Content: "她喜欢喝冰美式",
|
||||
Scope: memory.ScopePersona, Kind: memory.KindRelationship,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed memory: %v", err)
|
||||
}
|
||||
ret := memory.NewRetriever(store)
|
||||
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "喜欢冰美式"}}}}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{Agent: a, Retriever: ret, PersonaID: "persona-a", MaxRounds: 8})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", nil); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
// An empty conversation still routes through injectMemory; the retriever
|
||||
// returns the persona's items by recency when the query is blank.
|
||||
if len(ad.reqs) == 0 {
|
||||
t.Fatal("no request captured")
|
||||
}
|
||||
found := false
|
||||
for _, m := range ad.reqs[0].Messages {
|
||||
if m.Role == model.RoleSystem && strings.Contains(m.Content[0].Text, "冰美式") {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("injected memory not found in request messages: %+v", ad.reqs[0].Messages)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_MemoryExtract(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestMemoryStore(t)
|
||||
ext := memory.NewManualExtractor("persona-a", "role-1")
|
||||
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "记住了"}}}}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{Agent: a, Memory: store, Extractor: ext, PersonaID: "persona-a", MaxRounds: 8})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "记住 我爱喝冰美式")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
items, err := store.ListByPersona(context.Background(), "persona-a")
|
||||
if err != nil {
|
||||
t.Fatalf("ListByPersona: %v", err)
|
||||
}
|
||||
if len(items) != 1 {
|
||||
t.Fatalf("items = %d, want 1", len(items))
|
||||
}
|
||||
if items[0].Content != "我爱喝冰美式" {
|
||||
t.Fatalf("item content = %q, want 我爱喝冰美式", items[0].Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_StreamDeltas(t *testing.T) {
|
||||
t.Parallel()
|
||||
ad := &mockAdapter{
|
||||
stream: []mgResp{{resp: model.Response{Content: "Hello world"}}},
|
||||
chunks: [][]string{{"Hel", "lo ", "world"}},
|
||||
}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
var mu sync.Mutex
|
||||
var deltas []string
|
||||
l := New(Options{
|
||||
Agent: a,
|
||||
OnStream: func(d string) { mu.Lock(); deltas = append(deltas, d); mu.Unlock() },
|
||||
MaxRounds: 8,
|
||||
})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "Hello world" {
|
||||
t.Fatalf("Text = %q, want Hello world", res.Text)
|
||||
}
|
||||
mu.Lock()
|
||||
got := append([]string(nil), deltas...)
|
||||
mu.Unlock()
|
||||
if len(got) != 3 || got[0] != "Hel" || got[1] != "lo " || got[2] != "world" {
|
||||
t.Fatalf("deltas = %+v, want [Hel lo world]", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_StreamGuardrailBlocks(t *testing.T) {
|
||||
t.Parallel()
|
||||
g := &mockGuard{inputVerdict: &guardrail.Verdict{Allow: false, Reason: "blocked in"}}
|
||||
pipe := guardrail.New(guardrail.WithGuard(g))
|
||||
ad := &mockAdapter{}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{Agent: a, Guards: pipe, OnStream: func(string) {}, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.StopReason != agent.StopReasonGuardrailInput {
|
||||
t.Fatalf("StopReason = %q, want %q", res.StopReason, agent.StopReasonGuardrailInput)
|
||||
}
|
||||
if res.Text != "blocked in" {
|
||||
t.Fatalf("Text = %q, want blocked in", res.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_ContextCanceled(t *testing.T) {
|
||||
t.Parallel()
|
||||
ad := &mockAdapter{}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
if _, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")}); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("Run error = %v, want context.Canceled", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_NilAgent(t *testing.T) {
|
||||
t.Parallel()
|
||||
l := New(Options{})
|
||||
if _, err := l.Run(context.Background(), "sess", nil); !errors.Is(err, ErrNilAgent) {
|
||||
t.Fatalf("Run error = %v, want ErrNilAgent", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNew_DefaultMaxRounds(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := New(Options{}).MaxRounds(); got != DefaultMaxRounds {
|
||||
t.Fatalf("MaxRounds = %d, want %d", got, DefaultMaxRounds)
|
||||
}
|
||||
if got := New(Options{MaxRounds: 3}).MaxRounds(); got != 3 {
|
||||
t.Fatalf("MaxRounds = %d, want 3", got)
|
||||
}
|
||||
}
|
||||
|
||||
// decodeSpanLines reads every JSON span emitted as JSONL into recs, failing the
|
||||
// test if any line is malformed.
|
||||
func decodeSpanLines(t *testing.T, data []byte) []trace.SpanRecord {
|
||||
t.Helper()
|
||||
sc := bufio.NewScanner(bytes.NewReader(data))
|
||||
var recs []trace.SpanRecord
|
||||
for sc.Scan() {
|
||||
line := sc.Bytes()
|
||||
if len(bytes.TrimSpace(line)) == 0 {
|
||||
continue
|
||||
}
|
||||
var r trace.SpanRecord
|
||||
if err := json.Unmarshal(line, &r); err != nil {
|
||||
t.Fatalf("decode span: %v", err)
|
||||
}
|
||||
recs = append(recs, r)
|
||||
}
|
||||
if err := sc.Err(); err != nil {
|
||||
t.Fatalf("scan spans: %v", err)
|
||||
}
|
||||
return recs
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// injectMemory retrieves the persona's relevant long-term memories and injects
|
||||
// them into the first-turn context as a leading system message, so the model can
|
||||
// ground its answer in recalled context (design §6 / §2). It returns the input
|
||||
// messages unchanged when no retriever/persona is configured, retrieval fails, or
|
||||
// nothing relevant is found.
|
||||
func (l *Loop) injectMemory(ctx context.Context, messages []model.Message) []model.Message {
|
||||
if l.retriever == nil || l.personaID == "" {
|
||||
return messages
|
||||
}
|
||||
block, err := l.retriever.Inject(ctx, l.personaID, queryFromMessages(messages), 0)
|
||||
if err != nil {
|
||||
l.log.Warn("memory inject", "error", err)
|
||||
return messages
|
||||
}
|
||||
if block == "" {
|
||||
return messages
|
||||
}
|
||||
out := make([]model.Message, 0, len(messages)+1)
|
||||
out = append(out, model.TextMessage(model.RoleSystem, block))
|
||||
out = append(out, messages...)
|
||||
return out
|
||||
}
|
||||
|
||||
// extractMemory runs the extraction hook (design §6) over the current
|
||||
// conversation and persists any derived items. Errors are logged, never fatal, so
|
||||
// a bad extraction cannot abort the run.
|
||||
func (l *Loop) extractMemory(ctx context.Context, messages []model.Message, step int) {
|
||||
if l.extractor == nil || l.memory == nil {
|
||||
return
|
||||
}
|
||||
items, err := l.extractor.Extract(ctx, messages)
|
||||
if err != nil {
|
||||
l.log.Warn("memory extract", "step", step, "error", err)
|
||||
return
|
||||
}
|
||||
for _, it := range items {
|
||||
if it.PersonaID == "" && l.personaID != "" {
|
||||
it.PersonaID = l.personaID
|
||||
}
|
||||
if _, err := l.memory.Add(ctx, it); err != nil {
|
||||
l.log.Warn("memory store", "step", step, "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// queryFromMessages derives a retrieval query from the conversation: the last
|
||||
// user turn's text, falling back to the last message's text, then "".
|
||||
func queryFromMessages(messages []model.Message) string {
|
||||
for i := len(messages) - 1; i >= 0; i-- {
|
||||
if messages[i].Role == model.RoleUser {
|
||||
return messageText(messages[i])
|
||||
}
|
||||
}
|
||||
if n := len(messages); n > 0 {
|
||||
return messageText(messages[n-1])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// messageText returns the concatenated plain-text payload of a message's text
|
||||
// parts, which is the shared view used to build a retrieval query.
|
||||
func messageText(m model.Message) string {
|
||||
var b strings.Builder
|
||||
for _, part := range m.Content {
|
||||
if part.Type != model.ContentTypeText {
|
||||
continue
|
||||
}
|
||||
if b.Len() > 0 {
|
||||
b.WriteByte(' ')
|
||||
}
|
||||
b.WriteString(part.Text)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
@@ -0,0 +1,204 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
// mgResp is one scripted adapter result: a response plus an optional error.
|
||||
type mgResp struct {
|
||||
resp model.Response
|
||||
err error
|
||||
}
|
||||
|
||||
// mockAdapter is a configurable model.Adapter for loop tests. It keeps two
|
||||
// separate scripts and call counters (one for Chat, one for ChatStream) so a
|
||||
// single instance serves both the non-streaming and streaming paths. Every
|
||||
// request it is handed is recorded so tests can assert on the messages that were
|
||||
// fed to the model (e.g. the injected memory or a fed-back tool result).
|
||||
type mockAdapter struct {
|
||||
name string
|
||||
|
||||
chat []mgResp
|
||||
chatCalls int
|
||||
reqs []model.Request
|
||||
|
||||
stream []mgResp
|
||||
chunks [][]string
|
||||
streamCalls int
|
||||
streamReqs []model.Request
|
||||
}
|
||||
|
||||
func (m *mockAdapter) Name() string {
|
||||
if m.name == "" {
|
||||
return "mock"
|
||||
}
|
||||
return m.name
|
||||
}
|
||||
|
||||
func (m *mockAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) {
|
||||
m.reqs = append(m.reqs, req)
|
||||
if m.chatCalls >= len(m.chat) {
|
||||
return model.Response{Content: "done"}, nil
|
||||
}
|
||||
r := m.chat[m.chatCalls]
|
||||
m.chatCalls++
|
||||
return r.resp, r.err
|
||||
}
|
||||
|
||||
func (m *mockAdapter) ChatStream(_ context.Context, req model.Request, on func(string)) (model.Response, error) {
|
||||
m.streamReqs = append(m.streamReqs, req)
|
||||
i := m.streamCalls
|
||||
m.streamCalls++
|
||||
if i < len(m.chunks) {
|
||||
for _, c := range m.chunks[i] {
|
||||
if on != nil {
|
||||
on(c)
|
||||
}
|
||||
}
|
||||
}
|
||||
if i >= len(m.stream) {
|
||||
return model.Response{Content: "done"}, nil
|
||||
}
|
||||
return m.stream[i].resp, m.stream[i].err
|
||||
}
|
||||
|
||||
// mockGuard is a configurable guardrail.Guardrail for loop tests.
|
||||
type mockGuard struct {
|
||||
inputVerdict *guardrail.Verdict
|
||||
outputVerdict *guardrail.Verdict
|
||||
inputErr error
|
||||
outputErr error
|
||||
}
|
||||
|
||||
func (m *mockGuard) CheckInput(_ context.Context, _ *model.Request) (*guardrail.Verdict, error) {
|
||||
return m.inputVerdict, m.inputErr
|
||||
}
|
||||
|
||||
func (m *mockGuard) CheckOutput(_ context.Context, _ *model.Response) (*guardrail.Verdict, error) {
|
||||
return m.outputVerdict, m.outputErr
|
||||
}
|
||||
|
||||
// mockTool is a minimal sdk.Tool for loop tests. It supports a delay so a test
|
||||
// can exercise the per-invoke timeout, and an error so a test can verify that a
|
||||
// failed call is fed back to the model.
|
||||
type mockTool struct {
|
||||
name string
|
||||
desc string
|
||||
params map[string]any
|
||||
result any
|
||||
err error
|
||||
delay time.Duration
|
||||
}
|
||||
|
||||
func (m *mockTool) Name() string { return m.name }
|
||||
|
||||
func (m *mockTool) Description() string {
|
||||
if m.desc == "" {
|
||||
return "mock tool " + m.name
|
||||
}
|
||||
return m.desc
|
||||
}
|
||||
|
||||
func (m *mockTool) Parameters() (map[string]any, error) { return m.params, nil }
|
||||
|
||||
func (m *mockTool) Execute(ctx context.Context, _ any) (any, error) {
|
||||
if m.delay > 0 {
|
||||
select {
|
||||
case <-time.After(m.delay):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
return m.result, m.err
|
||||
}
|
||||
|
||||
// testLogger returns a logger that writes nowhere, keeping test output clean.
|
||||
func testLogger() *slog.Logger {
|
||||
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
}
|
||||
|
||||
// newTestAgent builds an Agent over ad with the given tools and no guardrails.
|
||||
func newTestAgent(t *testing.T, ad model.Adapter, tools *tool.Dispatcher) *agent.Agent {
|
||||
t.Helper()
|
||||
a, err := agent.New("You are a helpful assistant.", tools, nil, agent.WithModel(ad, nil))
|
||||
if err != nil {
|
||||
t.Fatalf("agent.New: %v", err)
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// dispatcherWith builds a Dispatcher with the default tool timeout.
|
||||
func dispatcherWith(t *testing.T, tools ...sdk.Tool) *tool.Dispatcher {
|
||||
t.Helper()
|
||||
return dispatcherWithTimeout(t, 0, tools...)
|
||||
}
|
||||
|
||||
// dispatcherWithTimeout builds a Dispatcher applying a custom default timeout
|
||||
// (a non-positive value selects the package default) and registering tools.
|
||||
func dispatcherWithTimeout(t *testing.T, timeout time.Duration, tools ...sdk.Tool) *tool.Dispatcher {
|
||||
t.Helper()
|
||||
reg := registry.New()
|
||||
var opts []tool.Option
|
||||
if timeout > 0 {
|
||||
opts = append(opts, tool.WithDefaultTimeout(timeout))
|
||||
}
|
||||
d, err := tool.New(reg, opts...)
|
||||
if err != nil {
|
||||
t.Fatalf("tool.New: %v", err)
|
||||
}
|
||||
for _, tk := range tools {
|
||||
if err := d.Register(tk); err != nil {
|
||||
t.Fatalf("Register %q: %v", tk.Name(), err)
|
||||
}
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// newTestMemoryStore opens an isolated in-memory SQLite database and returns a
|
||||
// memory.Store over it. Each call owns a private DB, so parallel tests are
|
||||
// isolated by construction.
|
||||
func newTestMemoryStore(t *testing.T) *memory.Store {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
d, err := db.Open(ctx, ":memory:", testLogger())
|
||||
if err != nil {
|
||||
t.Fatalf("db.Open: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.Close() })
|
||||
s, err := memory.NewStore(ctx, d, testLogger())
|
||||
if err != nil {
|
||||
t.Fatalf("memory.NewStore: %v", err)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// newTestCheckpointStore opens an isolated in-memory SQLite database and returns
|
||||
// a persist.Store over it.
|
||||
func newTestCheckpointStore(t *testing.T) *persist.Store {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
d, err := db.Open(ctx, ":memory:", testLogger())
|
||||
if err != nil {
|
||||
t.Fatalf("db.Open: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.Close() })
|
||||
s, err := persist.New(ctx, d, testLogger())
|
||||
if err != nil {
|
||||
t.Fatalf("persist.New: %v", err)
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// Options configures a Loop. All fields are optional and nil is tolerated until
|
||||
// the step that actually needs the field: the agent is required for Run (Run
|
||||
// returns ErrNilAgent otherwise); a missing tool dispatcher causes a requested
|
||||
// call to be fed back as an error; a missing retriever/persona skips memory
|
||||
// injection; a missing extractor/store skips the memory hook; a missing
|
||||
// checkpoint skips persistence; a missing tracer skips tracing.
|
||||
type Options struct {
|
||||
// Agent is the loop's execution unit: an instruction string plus a backend
|
||||
// adapter, an optional tool dispatcher and optional guardrails.
|
||||
Agent *agent.Agent
|
||||
// Tools is the tool dispatcher used to execute model-requested calls. When
|
||||
// nil the loop falls back to the agent's own dispatcher (if any).
|
||||
Tools *tool.Dispatcher
|
||||
// Memory is the long-term memory store used to persist extracted items.
|
||||
Memory *memory.Store
|
||||
// Retriever injects relevant memories into the first-turn context.
|
||||
Retriever *memory.Retriever
|
||||
// WorkingMemory is the per-session sliding window, provided for the
|
||||
// orchestrator to inspect; the loop does not mutate it.
|
||||
WorkingMemory *memory.WorkingMemory
|
||||
// Extractor derives memory items to persist after every step.
|
||||
Extractor memory.Extractor
|
||||
// Checkpoint persists a durable snapshot of the loop state at every step.
|
||||
Checkpoint *persist.Store
|
||||
// Tracer emits a structured span for every step, LLM call and tool call.
|
||||
Tracer *trace.Tracer
|
||||
// Guards validates request/response payloads on the streaming path, where
|
||||
// RunStep is bypassed. On the non-streaming path the agent runs its own
|
||||
// guardrails inside RunStep.
|
||||
Guards *guardrail.Pipeline
|
||||
// PersonaID scopes memory retrieval and extraction. Blank skips both.
|
||||
PersonaID string
|
||||
// MaxRounds is the tool-loop upper bound. Non-positive selects
|
||||
// DefaultMaxRounds (8).
|
||||
MaxRounds int
|
||||
// OnStream, when non-nil, switches the loop to the streaming path: the model
|
||||
// adapter's ChatStream is driven directly and each token delta is forwarded
|
||||
// to this callback.
|
||||
OnStream func(delta string)
|
||||
// Log is the base logger. Nil is replaced with a discard logger tagged with
|
||||
// source "runtime/loop".
|
||||
Log *slog.Logger
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// buildRequest assembles a model.Request from the agent's public surface,
|
||||
// mirroring agent.buildRequest so the streaming path behaves identically to
|
||||
// RunStep (leading instruction message, tool schemas, model routing).
|
||||
func (l *Loop) buildRequest(messages []model.Message) model.Request {
|
||||
req := model.Request{
|
||||
Model: l.agent.Model(),
|
||||
Messages: l.assembleMessages(messages),
|
||||
Temperature: l.temperature(),
|
||||
MaxTokens: l.agent.MaxTokens(),
|
||||
Purpose: l.agent.Purpose(),
|
||||
Tools: l.toolSchemas(),
|
||||
}
|
||||
if l.agent.Router() != nil {
|
||||
if m, _ := l.agent.Router().RouteRequest(req); m != "" {
|
||||
req.Model = m
|
||||
}
|
||||
}
|
||||
return req
|
||||
}
|
||||
|
||||
// assembleMessages prepends the agent's instructions as a leading system message
|
||||
// when they are non-empty, otherwise returns the input slice unchanged.
|
||||
func (l *Loop) assembleMessages(messages []model.Message) []model.Message {
|
||||
instr := l.agent.Instructions()
|
||||
if instr == "" {
|
||||
return messages
|
||||
}
|
||||
out := make([]model.Message, 0, len(messages)+1)
|
||||
out = append(out, model.TextMessage(model.RoleSystem, instr))
|
||||
out = append(out, messages...)
|
||||
return out
|
||||
}
|
||||
|
||||
// temperature returns a pointer to the configured temperature, mapping zero to
|
||||
// nil so the provider default applies (matching the agent package).
|
||||
func (l *Loop) temperature() *float64 {
|
||||
t := l.agent.Temperature()
|
||||
if t == 0 {
|
||||
return nil
|
||||
}
|
||||
return &t
|
||||
}
|
||||
|
||||
// toolSchemas advertises the registered tool schemas to the model, or nil when no
|
||||
// dispatcher is configured.
|
||||
func (l *Loop) toolSchemas() []model.ToolSpec {
|
||||
d := l.dispatcher()
|
||||
if d == nil {
|
||||
return nil
|
||||
}
|
||||
return d.ToolSchemas()
|
||||
}
|
||||
|
||||
// decodeToolCalls decodes tool calls from a model.Response using the same two
|
||||
// conventions as the agent package: the normalized RoleTool placeholder, then a
|
||||
// JSON envelope of the form {"tool_calls":[...]} in Content. It reports ok=true
|
||||
// (with a non-empty slice) when a call is decoded.
|
||||
func decodeToolCalls(resp model.Response) ([]agent.ToolCall, bool) {
|
||||
if resp.Message.Role == model.RoleAssistant && resp.Message.ToolCallName != "" {
|
||||
return []agent.ToolCall{{Name: resp.Message.ToolCallName, Args: map[string]any{}}}, true
|
||||
}
|
||||
var payload struct {
|
||||
ToolCalls []agent.ToolCall `json:"tool_calls"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(resp.Content), &payload); err != nil {
|
||||
return nil, false
|
||||
}
|
||||
if len(payload.ToolCalls) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
return payload.ToolCalls, true
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// Run drives the agent loop: inject relevant memory into the first-turn context,
|
||||
// then run up to MaxRounds steps. Each step runs the model (streaming through
|
||||
// OnStream when configured) and any requested tool calls are executed under a
|
||||
// per-invoke timeout and fed back as RoleTool messages so the model can continue.
|
||||
// A checkpoint is persisted at every step, every step is traced as a
|
||||
// state_transition span (with an llm_call child and tool_call spans for each
|
||||
// invocation), and after each step the memory Extractor runs with errors logged
|
||||
// rather than fatal.
|
||||
//
|
||||
// It returns the final answer, or a partial Result with StopReasonMaxRounds when
|
||||
// the budget is exhausted without a final answer. Step errors are wrapped with %w
|
||||
// so callers can classify them with errors.Is.
|
||||
func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Message) (*Result, error) {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if l.agent == nil {
|
||||
return nil, fmt.Errorf("loop: run: %w", ErrNilAgent)
|
||||
}
|
||||
|
||||
msgs := l.injectMemory(ctx, messages)
|
||||
res := &Result{}
|
||||
for step := 1; step <= l.maxRounds; step++ {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, fmt.Errorf("loop: run: step %d: %w", step, err)
|
||||
}
|
||||
|
||||
rc, roundSpan := l.span(ctx, "loop.step", trace.SpanKindStateTransition)
|
||||
outcome, err := l.step(rc, sessionID, msgs)
|
||||
if err != nil {
|
||||
l.spanProp(roundSpan, "error", err.Error())
|
||||
l.finishSpan(roundSpan, trace.SpanStatusError)
|
||||
return nil, fmt.Errorf("loop: run: step %d: %w", step, err)
|
||||
}
|
||||
l.spanProp(roundSpan, "step", step)
|
||||
l.spanProp(roundSpan, "stop_reason", outcome.stopReason)
|
||||
l.finishSpan(roundSpan, trace.SpanStatusOK)
|
||||
|
||||
res.Steps = step
|
||||
if outcome.final {
|
||||
res.Text = outcome.text
|
||||
res.StopReason = outcome.stopReason
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, outcome.stopReason)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
return res, nil
|
||||
}
|
||||
|
||||
newMsgs, executed, err := l.runToolCalls(rc, msgs, outcome.toolCalls)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loop: run: step %d: tools: %w", step, err)
|
||||
}
|
||||
msgs = newMsgs
|
||||
res.ToolCallsExecuted += executed
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, agent.StopReasonToolCalls)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
}
|
||||
|
||||
// Budget exhausted without a final answer.
|
||||
res.Steps = l.maxRounds
|
||||
res.StopReason = StopReasonMaxRounds
|
||||
l.saveCheckpoint(ctx, sessionID, l.maxRounds, msgs, StopReasonMaxRounds)
|
||||
l.extractMemory(ctx, msgs, l.maxRounds)
|
||||
return res, nil
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// stepOutcome is the loop's internal view of one model step, normalized from the
|
||||
// agent's *agent.StepResult (non-streaming) or the equivalent built by the
|
||||
// streaming path.
|
||||
type stepOutcome struct {
|
||||
final bool
|
||||
text string
|
||||
toolCalls []agent.ToolCall
|
||||
stopReason string
|
||||
}
|
||||
|
||||
// step runs a single model call. When an OnStream callback is configured it
|
||||
// drives the adapter's ChatStream directly, threading deltas to OnStream and
|
||||
// applying the loop's guardrails; otherwise it delegates to the agent's RunStep,
|
||||
// which runs the agent's own guardrails and records its own CallRecord. The whole
|
||||
// call is wrapped in an llm_call span so the M9 panel sees each LLM invocation.
|
||||
func (l *Loop) step(ctx context.Context, sessionID string, messages []model.Message) (*stepOutcome, error) {
|
||||
_, llmSpan := l.span(ctx, "llm.call", trace.SpanKindLLM)
|
||||
|
||||
var (
|
||||
out *stepOutcome
|
||||
err error
|
||||
)
|
||||
if l.onStream != nil {
|
||||
out, err = l.streamStep(ctx, messages)
|
||||
} else {
|
||||
out, err = l.agentStep(ctx, sessionID, messages)
|
||||
}
|
||||
if err != nil {
|
||||
l.spanProp(llmSpan, "error", err.Error())
|
||||
l.finishSpan(llmSpan, trace.SpanStatusError)
|
||||
return nil, fmt.Errorf("loop: model step: %w", err)
|
||||
}
|
||||
|
||||
l.spanProp(llmSpan, "stop_reason", out.stopReason)
|
||||
l.spanProp(llmSpan, "tool_calls", len(out.toolCalls))
|
||||
l.finishSpan(llmSpan, trace.SpanStatusOK)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// agentStep delegates a non-streaming step to the agent, which runs input and
|
||||
// output guardrails, calls the model and decodes any tool calls.
|
||||
func (l *Loop) agentStep(ctx context.Context, sessionID string, messages []model.Message) (*stepOutcome, error) {
|
||||
sr, err := l.agent.RunStep(ctx, sessionID, messages)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loop: agent step: %w", err)
|
||||
}
|
||||
return &stepOutcome{
|
||||
final: sr.Final,
|
||||
text: sr.Text,
|
||||
toolCalls: sr.ToolCalls,
|
||||
stopReason: sr.StopReason,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// streamStep drives a streaming model call. It is the streaming counterpart of
|
||||
// agent.RunStep: it builds the request from the agent's public surface, runs the
|
||||
// loop's input guardrail, streams deltas through OnStream, runs the output
|
||||
// guardrail and decodes any tool calls. Because RunStep does not expose a stream
|
||||
// path, this method is owned by the loop; it mirrors the agent's semantics so the
|
||||
// two behave identically apart from the delta callback.
|
||||
func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepOutcome, error) {
|
||||
adapter := l.agent.Adapter()
|
||||
if adapter == nil {
|
||||
return nil, fmt.Errorf("loop: stream step: %w", ErrNoAdapter)
|
||||
}
|
||||
req := l.buildRequest(messages)
|
||||
|
||||
if l.guards != nil {
|
||||
verdict, err := l.guards.CheckInput(ctx, &req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loop: check input: %w", err)
|
||||
}
|
||||
if verdict != nil && !verdict.Allow {
|
||||
return &stepOutcome{final: true, text: verdict.Reason, stopReason: agent.StopReasonGuardrailInput}, nil
|
||||
}
|
||||
}
|
||||
|
||||
resp, err := adapter.ChatStream(ctx, req, func(delta string) {
|
||||
if l.onStream != nil {
|
||||
l.onStream(delta)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loop: chat stream: %w", err)
|
||||
}
|
||||
|
||||
if l.guards != nil {
|
||||
verdict, err := l.guards.CheckOutput(ctx, &resp)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loop: check output: %w", err)
|
||||
}
|
||||
if verdict != nil && !verdict.Allow {
|
||||
return &stepOutcome{final: true, text: verdict.Reason, stopReason: agent.StopReasonGuardrailOutput}, nil
|
||||
}
|
||||
}
|
||||
|
||||
if calls, ok := decodeToolCalls(resp); ok {
|
||||
return &stepOutcome{final: false, toolCalls: calls, stopReason: agent.StopReasonToolCalls}, nil
|
||||
}
|
||||
return &stepOutcome{final: true, text: resp.Content, stopReason: agent.StopReasonStop}, nil
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// runToolCalls executes each requested tool call, appending a RoleTool result
|
||||
// message (success or failure) so the model can continue. It returns the updated
|
||||
// message slice and the number of tool calls invoked. Invocation is delegated to
|
||||
// the dispatched tool, which applies the per-tool deadline; a missing dispatcher
|
||||
// is fed back as an error so the loop never panics or crashes on an unhandled
|
||||
// call.
|
||||
func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls []agent.ToolCall) ([]model.Message, int, error) {
|
||||
d := l.dispatcher()
|
||||
out := messages
|
||||
executed := 0
|
||||
for _, tc := range calls {
|
||||
var (
|
||||
result any
|
||||
invokeErr error
|
||||
)
|
||||
if d == nil {
|
||||
invokeErr = fmt.Errorf("no tool dispatcher configured for %q", tc.Name)
|
||||
} else {
|
||||
_, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool)
|
||||
result, invokeErr = d.Invoke(ctx, tc.Name, tc.Args)
|
||||
if invokeErr != nil {
|
||||
l.spanProp(toolSpan, "error", invokeErr.Error())
|
||||
l.finishSpan(toolSpan, trace.SpanStatusError)
|
||||
} else {
|
||||
l.spanProp(toolSpan, "result", result)
|
||||
l.finishSpan(toolSpan, trace.SpanStatusOK)
|
||||
}
|
||||
}
|
||||
executed++
|
||||
out = append(out, toolResultMessage(tc, result, invokeErr))
|
||||
}
|
||||
return out, executed, nil
|
||||
}
|
||||
|
||||
// dispatcher returns the loop's tool dispatcher, falling back to the agent's own
|
||||
// when the loop was not given one.
|
||||
func (l *Loop) dispatcher() *tool.Dispatcher {
|
||||
if l.tools != nil {
|
||||
return l.tools
|
||||
}
|
||||
if l.agent != nil {
|
||||
return l.agent.Tools()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// toolResultMessage renders a tool call's outcome as a RoleTool message ready to
|
||||
// be fed back to the model. Errors are included in the text so the LLM can
|
||||
// react to a failed or timed-out call.
|
||||
func toolResultMessage(tc agent.ToolCall, result any, invokeErr error) model.Message {
|
||||
text := fmt.Sprintf("tool %s result: %s", tc.Name, formatToolResult(result))
|
||||
if invokeErr != nil {
|
||||
text = fmt.Sprintf("tool %s failed: %v", tc.Name, invokeErr)
|
||||
}
|
||||
return model.Message{Role: model.RoleTool, ToolCallName: tc.Name,
|
||||
Content: []model.Content{{Type: model.ContentTypeText, Text: text}}}
|
||||
}
|
||||
|
||||
// formatToolResult renders a tool return value as text for the model. Strings are
|
||||
// passed through (an empty string becomes "empty"); everything else is JSON
|
||||
// encoded, falling back to a plain %v render on an encoding error.
|
||||
func formatToolResult(result any) string {
|
||||
switch v := result.(type) {
|
||||
case nil:
|
||||
return "null"
|
||||
case string:
|
||||
if v == "" {
|
||||
return "empty"
|
||||
}
|
||||
return v
|
||||
default:
|
||||
b, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("%v", v)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// span starts a named span of the given kind when a Tracer is configured, else
|
||||
// returns a nil no-op span. The returned context carries the span as the current
|
||||
// one so nested spans chain as children.
|
||||
func (l *Loop) span(ctx context.Context, name string, kind trace.SpanKind) (context.Context, *trace.Span) {
|
||||
if l.tracer == nil {
|
||||
return ctx, nil
|
||||
}
|
||||
return l.tracer.StartSpan(ctx, name, kind)
|
||||
}
|
||||
|
||||
// finishSpan closes a span with the given status, ignoring a nil span.
|
||||
func (l *Loop) finishSpan(sp *trace.Span, status trace.SpanStatus) {
|
||||
if sp == nil {
|
||||
return
|
||||
}
|
||||
sp.Finish(status)
|
||||
}
|
||||
|
||||
// spanProp attaches a structured property to a span, ignoring a nil span and any
|
||||
// mutation-after-close error.
|
||||
func (l *Loop) spanProp(sp *trace.Span, key string, value any) {
|
||||
if sp == nil {
|
||||
return
|
||||
}
|
||||
_ = sp.SetProp(key, value)
|
||||
}
|
||||
Reference in new issue
Block a user