Documentation
¶
Index ¶
- Constants
- Variables
- func Retryable(err error) bool
- func UnsafeNotDone(pending []PendingTool, doneIDs map[string]bool) bool
- type Config
- type DefaultRunner
- func (r *DefaultRunner) Resume(ctx context.Context, runID string, req Request, sink EventSink) (*Result, error)
- func (r *DefaultRunner) Run(ctx context.Context, req Request) (*Result, error)
- func (r *DefaultRunner) RunWithSink(ctx context.Context, req Request, sink EventSink) (*Result, error)
- func (r *DefaultRunner) Stream(ctx context.Context, req Request, sink EventSink) (*Result, error)
- type DurableExecution
- type Error
- type ErrorKind
- type Event
- func NewHeartbeatEvent() Event
- func NewJobDoneEvent(jobID string) Event
- func NewJobFailedEvent(jobID string, err error) Event
- func NewJobPreviewEvent(jobID string, preview any) Event
- func NewJobProgressEvent(jobID string, percent float64, message string) Event
- func NewJobQueuedEvent(jobID string) Event
- func NewJobRunningEvent(jobID string) Event
- type EventSink
- type EventSinkFunc
- type EventType
- type JobProgress
- type Limits
- type Memory
- type MemoryEntry
- type Middleware
- type MiddlewareFuncSet
- func (m MiddlewareFuncSet) AfterLLM(ctx context.Context, s *State, r *llm.CompletionResponse) error
- func (m MiddlewareFuncSet) AfterTool(ctx context.Context, s *State, c llm.ToolCall, r ToolResult) error
- func (m MiddlewareFuncSet) BeforeLLM(ctx context.Context, s *State) error
- func (m MiddlewareFuncSet) BeforeTool(ctx context.Context, s *State, c llm.ToolCall) error
- func (m MiddlewareFuncSet) Finalize(ctx context.Context, s *State) error
- type NoopStrategy
- func (NoopStrategy) AfterLLM(context.Context, *State) error
- func (NoopStrategy) BeforeTurn(context.Context, *State) error
- func (NoopStrategy) BuildSystemPrefix(context.Context, Request) string
- func (NoopStrategy) Finalize(context.Context, *State) error
- func (NoopStrategy) Name() string
- func (NoopStrategy) ShouldContinue(_ context.Context, state *State) bool
- type PendingTool
- type Permission
- type Provider
- type ProviderSelection
- type ProviderSelector
- type Redaction
- type Request
- type Result
- type Runner
- type RuntimeError
- type SSEEventSink
- type SessionLane
- func (l *SessionLane) AcquireSession(ctx context.Context, sessionKey string) (lease.FencingToken, func(), error)
- func (l *SessionLane) Do(sessionKey, requestID string, fn func() (*Result, error)) (*Result, error)
- func (l *SessionLane) DoCtx(ctx context.Context, sessionKey, requestID string, fn func() (*Result, error)) (*Result, error)
- func (l *SessionLane) Forget(requestID string)
- func (l *SessionLane) SetDistributedLease(backend lease.Lease, ttl time.Duration)
- type SideEffectClassifier
- type Snapshot
- type State
- type StateSummary
- type StaticProviderSelector
- type StopReason
- type Strategy
- type StreamMode
- type ToolCallRecord
- type ToolExecutor
- type ToolResult
- type ToolSideEffect
- type ToolStatus
Constants ¶
const EventVersionV1 = "runtime.event.v1"
Variables ¶
var ( // ErrNoProvider means no provider was configured or selected. ErrNoProvider = errors.New("runtime: no provider selected") // ErrNoFallback means provider fallback is unavailable. ErrNoFallback = errors.New("runtime: no fallback provider") // ErrNilStream means the provider returned no stream for a streaming request. ErrNilStream = errors.New("runtime: provider returned nil stream") // ErrNoDurable means Resume was called but no DurableExecution was configured. ErrNoDurable = errors.New("runtime: no DurableExecution configured") // ErrNoSnapshot means Resume found no persisted snapshot for the run ID. ErrNoSnapshot = errors.New("runtime: no snapshot to resume") // ErrUnsafeReplay means Resume would replay an in-flight step whose tool calls // are not declared replay-safe (default), risking duplicated side effects. // Resolve by declaring the tool idempotent/read-only (SideEffectClassifier) or // performing manual dedup before resuming. ErrUnsafeReplay = errors.New("runtime: resume would replay an in-flight unsafe tool call; refusing to re-execute") // ErrBudgetExceeded 表示预算(token/时间/成本)超限,执行被 fail-closed 终止。 ErrBudgetExceeded = errors.New("runtime budget exceeded") )
Functions ¶
func Retryable ¶
Retryable 报告错误是否值得重试。
显式 *Error 以其 Retryable 标记为准;其余按 Kind 给默认:超时/provider 可重试, 取消/预算/权限/拒绝重放不可重试,未知保守为不可重试。
func UnsafeNotDone ¶
func UnsafeNotDone(pending []PendingTool, doneIDs map[string]bool) bool
UnsafeNotDone 报告一组待确认工具中,是否存在「尚未完成且不可安全重放」的工具。
doneIDs 是快照里已完成(结果已记录)的工具调用 ID 集合。已完成的工具即使是 Unsafe 也可安全跳过(副作用已发生且已记录);只有尚未完成的 Unsafe 工具才需 fail-closed (无法确定其副作用是否已发生)。这把 fail-closed 窗口从「整步含任一 Unsafe」收窄到 「真正在途的那个 Unsafe」。
Types ¶
type Config ¶
type Config struct {
ProviderSelector ProviderSelector
ToolExecutor ToolExecutor
Middleware []Middleware
DefaultMaxTurns int
// Durable 可选:开启可持久化/可恢复执行。
//
// 非 nil 时,Runner 在每个步边界把执行快照 Save 到该端口(详见 durable.go 的
// DurableExecution 契约),并支持 Resume 从最近快照续跑。nil 时(默认)完全不
// 触碰持久化,行为与无此字段时一致。
//
// 持久化以 Request.ID 作为 RunID(命名空间);Request.ID 为空时 Save 静默跳过
// (无可供恢复的标识符)。Resume 必须由调用方提供同一 RunID。
Durable DurableExecution
}
Config configures DefaultRunner.
type DefaultRunner ¶
type DefaultRunner struct {
// contains filtered or unexported fields
}
DefaultRunner is the default runtime kernel implementation.
func (*DefaultRunner) Resume ¶
func (r *DefaultRunner) Resume(ctx context.Context, runID string, req Request, sink EventSink) (*Result, error)
Resume 从某次执行最近的持久化快照续跑(需在 Config.Durable 配置持久化端口)。
行为:
- 未配置 Durable → ErrNoDurable。
- runID 无任何快照 → ErrNoSnapshot。
- 命中的快照已是终态(Final)→ 该次执行早已完成,直接返回其最终结果,不重跑。
- 否则用快照重建 State,从"快照步号 + 1"继续状态机;req 提供 provider 选择 / 工具 / 策略 / 限额等运行配置(消息历史以快照为准)。
func (*DefaultRunner) RunWithSink ¶
func (r *DefaultRunner) RunWithSink(ctx context.Context, req Request, sink EventSink) (*Result, error)
RunWithSink executes a request and also emits events to the supplied sink.
type DurableExecution ¶
type DurableExecution interface {
// Save 在步边界持久化一个执行快照。
Save(ctx context.Context, snap Snapshot) error
// Load 取某次执行最近的快照;不存在时 ok=false。
Load(ctx context.Context, runID string) (Snapshot, bool, error)
}
DurableExecution 把"让一次长时 Agent 执行可持久化、可恢复"收敛为一个最小接口:
- 调用方决定何时/是否持久化(在步边界 Save、在恢复时 Load);
- 接口只规定状态模型(Snapshot)与存储抽象(checkpoint.Checkpointer)的关系, 不绑定具体后端(后端由 Checkpointer 实现替换)。
契约:
- Save 前置:snap.RunID 非空,否则返回错误且不写入。
- Save 后置:随后对同一 RunID 的 Load 返回该 RunID 上最近一次成功 Save 的快照。
- 幂等:以 (RunID, Step) 为持久化键,同一步重复 Save 覆盖该步快照。
- Load 后置:命名空间无任何快照时 ok=false、err=nil。
DurableExecution 面向执行、checkpoint.Checkpointer 面向存储,前者用后者落地, 二者职责分离、各自可替换。
func NewDurableExecution ¶
func NewDurableExecution(cp checkpoint.Checkpointer) DurableExecution
NewDurableExecution 用给定的 Checkpointer 构造默认 DurableExecution。
cp 为 nil 时 panic(属调用方编程错误:没有存储后端无法构造持久化执行)。
type Error ¶
type Error struct {
Kind ErrorKind
Message string
Retryable bool
// contains filtered or unexported fields
}
Error 是带分类的统一错误:携带 Kind + 可重试标记 + 包装的底层错误。
type ErrorKind ¶
type ErrorKind string
ErrorKind 是跨层错误分类。
const ( // KindUnknown 未分类错误 KindUnknown ErrorKind = "unknown" // KindCanceled 上下文取消 KindCanceled ErrorKind = "canceled" // KindTimeout 超时(含 deadline 超时) KindTimeout ErrorKind = "timeout" // KindBudget 预算超限(token/时间/成本) KindBudget ErrorKind = "budget" // KindPermission 权限拒绝 KindPermission ErrorKind = "permission" // KindProvider provider 不可用/未选择 KindProvider ErrorKind = "provider" // KindUnsafeReplay 拒绝重放有副作用的在途工具 KindUnsafeReplay ErrorKind = "unsafe_replay" // KindToolFailure 工具执行失败 KindToolFailure ErrorKind = "tool_failure" )
type Event ¶
type Event struct {
Version string
Type EventType
RunID string
RequestID string
SessionID string
Turn int
Sequence int64
Timestamp time.Time
TraceID string
SpanID string
ParentSpanID string
// State 是运行时**仍在演进的活状态指针**,仅在 emit 回调期间有效。
// emit 为同步调用:sink 在回调内读取/Snapshot 是安全的;若需在回调返回后保留或
// 异步读取,**必须先 State.Snapshot() 取独立副本**,否则与后续步边界写入构成 data race。
// 只需标量视图(Turn/Final/计数)时优先用 StateSummary(已是安全的值拷贝)。
State *State
StateSummary StateSummary
Response *llm.CompletionResponse
Chunk *llm.StreamChunk
ToolCall *llm.ToolCall
ToolResult *ToolResult
Error error
RuntimeError *RuntimeError
Metadata map[string]any
Payload any
Redaction Redaction
}
Event is emitted by the runtime state machine.
func NewJobFailedEvent ¶
NewJobFailedEvent 任务失败。
func NewJobPreviewEvent ¶
NewJobPreviewEvent 阶段性预览(缩略图 / 部分结果等)。
func NewJobProgressEvent ¶
NewJobProgressEvent 进度更新(percent 取值 [0,100])。
type EventSinkFunc ¶
EventSinkFunc adapts a function to EventSink.
type EventType ¶
type EventType string
EventType identifies a runtime event.
const ( EventRunStarted EventType = "run_started" EventProviderSelected EventType = "provider_selected" EventLLMStarted EventType = "llm_started" EventLLMChunk EventType = "llm_chunk" EventLLMCompleted EventType = "llm_completed" EventToolCallStarted EventType = "tool_call_started" EventToolCallCompleted EventType = "tool_call_completed" EventToolCallFailed EventType = "tool_call_failed" EventProviderFallback EventType = "provider_fallback" EventBudgetChecked EventType = "budget_checked" EventBudgetExceeded EventType = "budget_exceeded" EventContextCompacted EventType = "context_compacted" EventPermissionRequested EventType = "permission_requested" EventPermissionApproved EventType = "permission_approved" EventPermissionDenied EventType = "permission_denied" EventReasoningSanitized EventType = "reasoning_sanitized" EventCheckpointSaved EventType = "checkpoint_saved" EventRunFinished EventType = "run_finished" EventRunFailed EventType = "run_failed" // 长时任务进度事件(B-fw5)—— 供 hexeye 等"提交→轮询"长任务把进度作为一类 // RuntimeEvent 贯通到 SSE。载荷置于 Event.Payload(*JobProgress),SSE sink 自动透传。 EventJobQueued EventType = "job_queued" EventJobRunning EventType = "job_running" EventJobProgress EventType = "job_progress" EventJobPreview EventType = "job_preview" EventJobDone EventType = "job_done" EventJobFailed EventType = "job_failed" // EventHeartbeat 心跳事件,保持长连接活跃(无业务载荷)。 EventHeartbeat EventType = "heartbeat" )
type JobProgress ¶
type JobProgress struct {
// JobID 任务标识。
JobID string `json:"job_id,omitempty"`
// Stage 阶段名(queued/running/progress/preview/done/failed)。
Stage string `json:"stage,omitempty"`
// Percent 进度百分比 [0,100],progress 阶段有效。
Percent float64 `json:"percent,omitempty"`
// Message 人类可读进度描述。
Message string `json:"message,omitempty"`
// Preview 阶段性预览(缩略图 URL / 部分结果等,任意可序列化值)。
Preview any `json:"preview,omitempty"`
}
JobProgress 是长时任务进度事件的载荷(置于 Event.Payload,随 SSE 透传给客户端)。
用于 hexeye 短剧/媒体等长耗时"提交→轮询"任务:把 queued/running/progress(x%)/preview/done 作为统一的一类 RuntimeEvent 上报,desktop 的 /cost、前端进度条等可直接消费。
type Memory ¶
type Memory interface {
Search(ctx context.Context, query string, limit int) ([]MemoryEntry, error)
Save(ctx context.Context, entry MemoryEntry) error
}
Memory is the optional long-term memory port.
type MemoryEntry ¶
MemoryEntry is a product-neutral memory record.
type Middleware ¶
type Middleware interface {
BeforeLLM(ctx context.Context, state *State) error
AfterLLM(ctx context.Context, state *State, resp *llm.CompletionResponse) error
BeforeTool(ctx context.Context, state *State, call llm.ToolCall) error
AfterTool(ctx context.Context, state *State, call llm.ToolCall, result ToolResult) error
Finalize(ctx context.Context, state *State) error
}
Middleware observes or modifies runtime state at well-defined lifecycle points.
type MiddlewareFuncSet ¶
type MiddlewareFuncSet struct {
BeforeLLMFunc func(context.Context, *State) error
AfterLLMFunc func(context.Context, *State, *llm.CompletionResponse) error
BeforeToolFunc func(context.Context, *State, llm.ToolCall) error
AfterToolFunc func(context.Context, *State, llm.ToolCall, ToolResult) error
FinalizeFunc func(context.Context, *State) error
}
MiddlewareFuncSet is a convenience middleware implementation.
func (MiddlewareFuncSet) AfterLLM ¶
func (m MiddlewareFuncSet) AfterLLM(ctx context.Context, s *State, r *llm.CompletionResponse) error
func (MiddlewareFuncSet) AfterTool ¶
func (m MiddlewareFuncSet) AfterTool(ctx context.Context, s *State, c llm.ToolCall, r ToolResult) error
func (MiddlewareFuncSet) BeforeLLM ¶
func (m MiddlewareFuncSet) BeforeLLM(ctx context.Context, s *State) error
func (MiddlewareFuncSet) BeforeTool ¶
type NoopStrategy ¶
type NoopStrategy struct{}
NoopStrategy is the default ReAct-compatible strategy.
func (NoopStrategy) BeforeTurn ¶
func (NoopStrategy) BeforeTurn(context.Context, *State) error
func (NoopStrategy) BuildSystemPrefix ¶
func (NoopStrategy) BuildSystemPrefix(context.Context, Request) string
func (NoopStrategy) Name ¶
func (NoopStrategy) Name() string
func (NoopStrategy) ShouldContinue ¶
func (NoopStrategy) ShouldContinue(_ context.Context, state *State) bool
type PendingTool ¶
type PendingTool struct {
Call llm.ToolCall `json:"call"`
SideEffect ToolSideEffect `json:"side_effect"`
}
PendingTool 是步内意图快照里一条「已发起、待确认完成」的工具调用及其重放安全性。
保存完整 Call(含 ID/Name/Arguments),使 Resume 能**精确续跑**——按 ID 跳过已 完成的工具(其结果已在快照的 ToolCalls 里),只补跑未完成的,且无需重调 LLM。
type Permission ¶
Permission checks whether a tool call can execute.
type Provider ¶
type Provider interface {
Name() string
Complete(ctx context.Context, req llm.CompletionRequest) (*llm.CompletionResponse, error)
Stream(ctx context.Context, req llm.CompletionRequest) (*llm.Stream, error)
}
Provider is the LLM port used by the runtime.
type ProviderSelection ¶
ProviderSelection describes the selected model backend.
type ProviderSelector ¶
type ProviderSelector interface {
Select(ctx context.Context, req Request) (ProviderSelection, error)
Fallback(ctx context.Context, failed ProviderSelection, err error) (ProviderSelection, error)
}
ProviderSelector selects the primary provider and optional fallback.
type Redaction ¶
type Redaction struct {
Applied bool `json:"applied,omitempty"`
Fields []string `json:"fields,omitempty"`
}
Redaction describes whether sensitive event payload fields were redacted.
type Request ¶
type Request struct {
ID string
Messages []llm.Message
Tools []llm.ToolDefinition
ProviderName string
ModelName string
Metadata map[string]any
Limits Limits
Strategy Strategy
StreamMode StreamMode
}
Request is the product-neutral input for an agent runtime run.
type Result ¶
type Result struct {
Content string
Reasoning string
ToolCalls []ToolCallRecord
// Blocks 是本次运行的**有序内容块流**(text/tool_use/tool_result 按执行序交错),
// 修复 Content 单串 + ToolCalls 扁平数组无法表达多步 text↔tool 交错的缺陷。
// 客户端有此字段时按序渲染,否则回退 Content + ToolCalls。
Blocks template.Blocks
Usage llm.Usage
Metadata map[string]any
StopReason StopReason
}
Result is the product-neutral output of an agent runtime run.
type Runner ¶
type Runner interface {
Run(ctx context.Context, req Request) (*Result, error)
Stream(ctx context.Context, req Request, sink EventSink) (*Result, error)
}
Runner executes agent requests using a unified state machine.
type RuntimeError ¶
type RuntimeError struct {
Code string `json:"code,omitempty"`
Message string `json:"message,omitempty"`
Cause string `json:"cause,omitempty"`
}
RuntimeError is a structured error payload for event consumers.
func AsRuntimeError ¶
func AsRuntimeError(err error) *RuntimeError
AsRuntimeError 把任意错误转为可上报的 RuntimeError 载荷,Code 取其分类。
type SSEEventSink ¶
type SSEEventSink struct {
// 可选过滤器:返 false 跳过该事件不发给客户端。nil 表示全发。
Filter func(event Event) bool
// contains filtered or unexported fields
}
SSEEventSink 把 runtime.Event 串行序列化为 W3C SSE 帧并 flush 到 HTTP response。
把 http.ResponseWriter 包成 SSEEventSink 即获得:
- 自动 Content-Type / Cache-Control header
- 标准 `event: <type>\ndata: <json>\n\n` 帧格式
- 线程安全 flush(多 goroutine 同时 emit 不竞争)
- context.Done() 自动断流(客户端 abort 时停止序列化)
使用:
sink := runtime.NewSSEEventSink(w) defer sink.Close() runner.Run(ctx, req, sink)
host 也可在每个事件前后塞自己的应用层事件(cron compile progress 等)。
func NewSSEEventSink ¶
func NewSSEEventSink(w http.ResponseWriter) (*SSEEventSink, error)
NewSSEEventSink 在 ResponseWriter 上准备 SSE headers 并返回 sink。 若 ResponseWriter 不支持 http.Flusher,返 nil + error。
func (*SSEEventSink) AsIOCloser ¶
func (s *SSEEventSink) AsIOCloser() io.Closer
AsIOCloser 把 SSEEventSink 包装为 io.Closer 供 defer 使用。
func (*SSEEventSink) Close ¶
func (s *SSEEventSink) Close()
Close 标记 sink 关闭(之后的 Emit 静默 noop)。可重复调用。
func (*SSEEventSink) Emit ¶
func (s *SSEEventSink) Emit(ctx context.Context, event Event) error
Emit 实现 EventSink。
行为:
- 已 Close → 静默返 nil(不向 detached 连接写)
- ctx.Done → 返 ctx.Err
- Filter 过滤为 false → 跳过
- JSON 序列化失败 → 返 error(不中断 runtime)
- Write 失败 → 标记 closed + 返 error(避免后续 emit 继续写挂掉的连接)
func (*SSEEventSink) EmitRaw ¶
EmitRaw 发送 host 自定义的 SSE 事件,不经 runtime.Event 包装。
用途:host(如 hexclaw cron handler)想复用 SSEEventSink 的 headers 设置、线程安全 flush、Close 语义和 ctx 取消,但需要发自己业务定义的 event name + payload 形态(与 runtime.EventType 枚举无关)。
行为:
- 已 Close → 静默返 nil
- ctx.Done → 返 ctx.Err
- 不经 Filter(Filter 是 runtime.Event 语义)
- JSON 序列化失败 → 返 error
- Write 失败 → 标记 closed + 返 error
wire 格式与 Emit 一致:`event: <eventName>\ndata: <json>\n\n`
func (*SSEEventSink) WriteComment ¶
func (s *SSEEventSink) WriteComment(text string) error
WriteComment 发送 SSE 注释帧(": keepalive\n\n"),保持长连接活跃 / 调试用。 不算事件,不经 Filter。
type SessionLane ¶
type SessionLane struct {
// contains filtered or unexported fields
}
SessionLane 提供两条执行纪律(A8):
- **同 session 串行化**:同一 sessionKey 的执行互斥串行,避免多副本/多窗口并发踩踏 同一会话状态。默认单进程(per-session sync.Mutex);经 SetDistributedLease 注入 分布式租约后,串行跨副本生效(见 DoCtx / AcquireSession)。
- **同 RequestID 幂等**:同一 requestID 的重试不重复执行 fn(返回首次结果),避免重复 执行/重复计费。
跨副本串行 + fencing token 见 session_lane_distributed.go。
线程安全:所有方法可并发调用。
func (*SessionLane) AcquireSession ¶
func (l *SessionLane) AcquireSession(ctx context.Context, sessionKey string) (lease.FencingToken, func(), error)
AcquireSession 获取 sessionKey 的分布式租约,返回 fencing token 与释放函数。
若未注入分布式租约则返回 token=0、释放函数为空操作(退化为不跨副本串行)。 租约被他人持有时轮询等待,直到获取成功或 ctx 取消。fencing token 供下游受保护 资源比较、拒绝过期持有者写入。
func (*SessionLane) Do ¶
Do 在会话串行 + 请求幂等保护下执行 fn。
- sessionKey 非空 → 同 session 串行(互斥);为空 → 不串行。
- requestID 非空 → 幂等(同 ID 重试返回首次结果,不重复执行 fn);为空 → 每次执行。
⚠️ 注意:Do 只用进程内 sync.Mutex 串行,**不感知 SetDistributedLease 注入的分布式租约**。 若配置了分布式租约以求跨副本串行,所有调用方必须统一用 DoCtx——混用 Do 与 DoCtx 会让 Do 的调用绕过租约、与 DoCtx 的调用在跨副本下并发。需要跨副本串行时一律走 DoCtx。
func (*SessionLane) DoCtx ¶
func (l *SessionLane) DoCtx(ctx context.Context, sessionKey, requestID string, fn func() (*Result, error)) (*Result, error)
DoCtx 在跨副本会话串行 + 请求幂等保护下执行 fn。
与 Do 同义,但当注入了分布式租约时,sessionKey 的串行跨副本生效(先取分布式租约 再执行,结束释放);未注入租约时退化为单进程 sync.Mutex(与 Do 一致)。 requestID 幂等语义与 Do 相同。
func (*SessionLane) Forget ¶
func (l *SessionLane) Forget(requestID string)
Forget 移除某 requestID 的幂等缓存(释放内存;下次同 ID 将重新执行)。
func (*SessionLane) SetDistributedLease ¶
func (l *SessionLane) SetDistributedLease(backend lease.Lease, ttl time.Duration)
SetDistributedLease 注入分布式租约后端与租约 TTL,启用跨副本会话串行。
传 nil 关闭分布式串行(回到单进程 sync.Mutex)。进程内可用 lease.NewMemoryLease, 跨副本用 Redis 等后端实现的 lease.Lease。
type SideEffectClassifier ¶
type SideEffectClassifier interface {
SideEffectOf(call llm.ToolCall) ToolSideEffect
}
SideEffectClassifier 是 ToolExecutor 的可选附加能力:声明某次工具调用在「崩溃后 重放」语义下的安全性(只读 / 幂等 / 不安全)。
仅用于 Durable 执行的 exactly-once 保护:执行器若实现本接口,Runner 在工具执行前 的意图快照里记录各工具的重放安全性,Resume 据此判定能否安全续跑。**未实现时所有 工具按最保守的 SideEffectUnsafe 处理**——即崩溃在步内时 Resume 会 fail-closed, 宁可显式失败也不静默重复副作用。
type Snapshot ¶
type Snapshot struct {
// RunID 标识一次执行,同时作为持久化命名空间。
RunID string `json:"run_id"`
// Step 快照所在的步边界(对应 State.Turn)。
Step int `json:"step"`
// Messages 截至本步的消息历史。
Messages []llm.Message `json:"messages"`
// ToolCalls 截至本步已完成的工具调用记录。
ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"`
// Usage 截至本步的累计 token 用量(计量维度)。
Usage llm.Usage `json:"usage"`
// Final 标记执行是否已得出最终答案。
Final bool `json:"final,omitempty"`
// FinalText 最终答案文本(Final 为真时有效)。
FinalText string `json:"final_text,omitempty"`
// Pending 非空表示这是一个「步内意图快照」:该步的工具调用已发起但尚未确认全部
// 完成(崩溃窗口)。它记录这些工具的重放安全性,供 Resume 判定能否安全续跑——
// 含 Unsafe 工具时 Resume fail-closed(ErrUnsafeReplay),全部可重放安全时重跑该步。
// 步正常完成后的完成快照不带 Pending(同步号覆盖意图快照)。
Pending []PendingTool `json:"pending,omitempty"`
}
Snapshot 是一次执行在某个步边界上的持久化状态。
它只包含恢复一次 Agent 执行所必需的最小状态:步号、消息历史、已完成工具调用记录、 累计用量与终止信息。Reasoning、Attributes 等运行期派生/易变数据不入快照,以免把 易变细节固化进持久化格式、并降低与具体执行实现的耦合。
func SnapshotState ¶
SnapshotState 从运行时 State 抽取当前步的 Snapshot。runID 标识本次执行。
切片做浅拷贝快照,避免与仍在演进的 State 共享底层数组。
func (Snapshot) RestoreState ¶
RestoreState 用 Snapshot 重建一个可继续执行的 State(resume 入口)。
返回的 State 从 Snapshot 的步号继续;Attributes 重新初始化为空(provider/model 等运行期属性在恢复后的首次调用时重新填充)。
type State ¶
type State struct {
Request Request
Messages []llm.Message
ToolCalls []ToolCallRecord
Usage llm.Usage
Turn int
Final bool
FinalText string
Reasoning string
Attributes map[string]any
// contains filtered or unexported fields
}
State is the mutable state owned by the runtime state machine.
func (*State) Emit ¶
Emit lets middleware publish runtime events without knowing the runner internals.
func (*State) Snapshot ¶
Snapshot 返回 State 的独立只读副本,安全用于跨 goroutine 保留与异步读取。
并发契约:EventSink 收到的 Event.State 是运行时**仍在演进的活指针**——它由运行 循环单线程持有并在步边界 append Messages / 写 Attributes。emit 为同步调用,故 sink 在回调**内**(运行循环阻塞等待期间)读取或 Snapshot 是安全的;但若 sink 想在回调 返回后保留 State、或另起 goroutine 异步读取,**必须先在回调内 Snapshot 取独立副本**, 否则与后续步边界的写入构成 data race(Messages 切片读写 / Attributes map 并发读写 panic)。
复制 Messages/ToolCalls/Attributes 这三处可变部分到全新底层存储;emit 句柄不复制 (快照是只读视图,不应触发事件)。
type StateSummary ¶
type StateSummary struct {
Turn int `json:"turn"`
Final bool `json:"final"`
Messages int `json:"messages"`
ToolCalls int `json:"tool_calls"`
}
StateSummary is a stable, low-cardinality view of runtime state for event consumers.
type StaticProviderSelector ¶
StaticProviderSelector is a single-provider selector.
func (StaticProviderSelector) Fallback ¶
func (s StaticProviderSelector) Fallback(context.Context, ProviderSelection, error) (ProviderSelection, error)
Fallback returns no fallback.
func (StaticProviderSelector) Select ¶
func (s StaticProviderSelector) Select(context.Context, Request) (ProviderSelection, error)
Select returns the configured provider.
type StopReason ¶
type StopReason string
StopReason 是运行终止的一等原因,对齐 Anthropic/OpenAI 的 stop_reason 语义:到达 limit 是正常终止而非错误。它是表达「为什么停」的**唯一**机制——始终随 Result 返回,调用方据此 决定如何呈现(如 max_turns 时提示「可继续」),无需 errors.Is 反查。达到轮次上限不再产生 任何 error(Run/Stream 返回 nil error + StopReason=max_turns)。
const ( // StopReasonEndTurn 模型给出了最终答案,运行正常结束。 StopReasonEndTurn StopReason = "end_turn" // StopReasonMaxTurns 达到工具循环轮次上限仍无终态——结果携带已产出的部分内容/用量, // 调用方可据此呈现部分结果并提示用户继续。 StopReasonMaxTurns StopReason = "max_turns" )
type Strategy ¶
type Strategy interface {
Name() string
BuildSystemPrefix(ctx context.Context, req Request) string
BeforeTurn(ctx context.Context, state *State) error
AfterLLM(ctx context.Context, state *State) error
ShouldContinue(ctx context.Context, state *State) bool
Finalize(ctx context.Context, state *State) error
}
Strategy controls product-neutral agent behavior.
type StreamMode ¶
type StreamMode string
StreamMode controls how the runtime invokes the provider and projects output.
const ( // StreamModeOff uses provider.Complete and only emits lifecycle events. StreamModeOff StreamMode = "off" // StreamModeEvents uses provider.Complete while still emitting runtime events. StreamModeEvents StreamMode = "events" // StreamModeTokens uses provider.Stream and emits LLMChunk events. StreamModeTokens StreamMode = "tokens" )
type ToolCallRecord ¶
type ToolCallRecord struct {
ID string
Name string
Arguments string
Result ToolResult
}
ToolCallRecord records a completed tool call.
type ToolExecutor ¶
ToolExecutor executes tool calls.
type ToolResult ¶
type ToolResult struct {
Content string
Raw any
Error string
// Status 是执行结果状态(success / error),由框架在执行点据 execErr 与
// tool.Result.Success 判定。零值(空串)表示未填充——老快照/老路径向后兼容。
Status ToolStatus `json:"status,omitempty"`
// DurationMs 是工具执行耗时(毫秒),由框架在执行点 time.Since 测量。
DurationMs int64 `json:"duration_ms,omitempty"`
}
ToolResult is a product-neutral tool result.
type ToolSideEffect ¶
type ToolSideEffect int
ToolSideEffect 声明一次工具调用在「崩溃后重放」语义下的安全性。
用于 Durable 执行的 exactly-once 保护:当一步的工具已发起但崩溃在快照落盘前, Resume 需据此判定能否安全重跑——只读/幂等可重放,否则 fail-closed。
const ( // SideEffectUnsafe 默认:重放可能重复副作用(发邮件 / 扣款 / 写库等)。最保守默认, // 未显式声明的工具一律按此处理。 SideEffectUnsafe ToolSideEffect = iota // SideEffectIdempotent 幂等:重放产生相同结果、不产生额外副作用。 SideEffectIdempotent // SideEffectReadOnly 只读:无任何副作用。 SideEffectReadOnly )
func (ToolSideEffect) ReplaySafe ¶
func (s ToolSideEffect) ReplaySafe() bool
ReplaySafe 报告该副作用级别是否可安全重放(幂等或只读)。
type ToolStatus ¶
type ToolStatus string
ToolStatus 是工具单次调用的执行结果状态——随 ToolResult 一等返回,客户端据此渲染 (成功 / 失败),无需对结果正文做字符串嗅探。对齐 StopReason 的「执行真相一等化」范式: 框架在执行点拥有完整上下文,应直接产出状态,而非让上层反查。仅表达**已完成**的两态 (执行中是流式概念,不属批量结果)。
const ( // ToolStatusSuccess 表示工具执行成功(无 Go 级 execErr 且 tool.Result.Success 为真)。 ToolStatusSuccess ToolStatus = "success" // ToolStatusError 表示工具执行失败(Go 级 execErr,或 tool.Result.Success 为假的软失败)。 ToolStatusError ToolStatus = "error" )
工具执行结果状态常量:成功 / 失败两态。
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package strategy 提供统一 agent loop 的可选执行策略。
|
Package strategy 提供统一 agent loop 的可选执行策略。 |