Documentation
¶
Overview ¶
Package agent 提供 AI Agent 核心接口和实现
本包实现了多种 Agent 类型,包括:
- BaseAgent: 基础 Agent,提供通用能力
- ReActAgent: 实现 ReAct (Reasoning + Acting) 推理模式
- Team: 多 Agent 协作团队,支持顺序/层级/协作/轮询四种模式
- SwarmRunner: 模仿 OpenAI Swarm 的 Agent 交接运行器
状态管理:
- TurnState: 单轮对话状态
- SessionState: 会话级状态
- AgentState: Agent 持久状态
- GlobalState: 跨 Agent 共享状态
使用示例:
agent := NewReAct(
WithName("assistant"),
WithLLM(llmProvider),
WithTools(searchTool, calculatorTool),
)
output, err := agent.Run(ctx, Input{Query: "Hello"})
conversation.go 提供多轮对话 Agent 封装
ConversationAgent 在普通 Agent 之上增加对话历史管理, 自动将历史上下文注入到每次请求中。
使用示例:
conv := agent.NewConversation(myAgent,
agent.WithConvMaxTurns(20),
agent.WithConvMaxTokens(4096),
)
output, err := conv.Chat(ctx, "你好")
output, err = conv.Chat(ctx, "继续上面的话题")
deep.go 实现深度 Agent 模式
DeepAgent 支持递归子任务分解:当遇到复杂任务时,通过内置的 "create_subtask" 工具自动创建子 Agent 处理子任务,将结果汇总返回。
实现递归子任务分解与层级执行。
使用示例:
deep := NewDeepAgent("deep-researcher",
baseAgent,
WithSubAgentFactory(func(task string) Agent {
return NewBaseAgent(
WithName("sub-"+task[:10]),
WithLLM(myLLM),
WithSystemPrompt("Focus on: "+task),
)
}),
WithMaxDepth(3),
)
output, err := deep.Run(ctx, Input{Query: "对比分析三大云厂商"})
Package agent 提供 AI Agent 核心接口和实现 ¶
Package agent 提供 AI Agent 核心功能 ¶
本文件实现 Agent 能力协商功能:
能力声明:Agent 声明自己的能力
能力查询:查询 Agent 的能力
能力匹配:匹配任务需求与 Agent 能力
能力协商:多 Agent 能力协商
A2A Protocol: Agent 能力协商
OpenAPI: 接口描述规范
WSDL: 服务描述语言
Package agent 提供 AI Agent 核心接口和实现 ¶
primitives.go 实现高级 Agent 编排原语:
- SequentialAgent: 按顺序依次执行多个子 Agent
- ParallelAgent: 并行执行多个子 Agent,合并结果
- LoopAgent: 循环执行子 Agent 直到满足条件
这些原语提供比 Graph 更简单直观的编排方式,适合常见场景。
使用示例:
// 顺序执行:研究 → 撰写 → 审核
pipeline := NewSequentialAgent("pipeline",
researchAgent,
writerAgent,
reviewerAgent,
)
output, err := pipeline.Run(ctx, Input{Query: "写一篇关于 AI 的文章"})
// 并行执行:同时搜索多个来源
searcher := NewParallelAgent("searcher",
webSearchAgent,
dbSearchAgent,
docSearchAgent,
)
output, err := searcher.Run(ctx, Input{Query: "Go 并发模式"})
Package agent 提供 AI Agent 核心功能 ¶
本文件实现 Agent 发现与注册功能:
Agent 注册表:中央注册和管理 Agent
服务发现:动态发现可用 Agent
健康检查:监控 Agent 状态
负载均衡:智能路由请求
Consul: 服务注册与发现
Kubernetes: Service Discovery
gRPC: 服务发现机制
Package agent 提供 AI Agent 接口和实现 ¶
supervisor.go 实现监督者 Agent 模式
SupervisorAgent 由一个 Manager Agent 动态决定将任务分派给哪个 Worker Agent。 Supervisor 模式:主 Agent 调度子 Agent 并汇总结果。
执行流程:
- 将所有 worker 通过 AgentAsTool 注册为 manager 的工具
- manager 通过 tool call 选择 worker
- 执行被选中的 worker
- 将 worker 结果返回给 manager
- manager 决定继续分派或返回最终结果
使用示例:
supervisor := NewSupervisor("coordinator",
managerAgent,
[]Agent{researcher, writer, reviewer},
WithSupervisorRounds(5),
)
output, err := supervisor.Run(ctx, Input{Query: "写一篇关于 AI 的文章"})
Index ¶
- Constants
- Variables
- func AgentAsTool(ag Agent) tool.Tool
- func ContextWithSafeVariables(ctx context.Context, vars *SafeContextVariables) context.Context
- func ContextWithStateManager(ctx context.Context, sm StateManager) context.Context
- func ContextWithVariables(ctx context.Context, vars ContextVariables) context.Context
- func NewGlobalState() *defaultGlobalState
- func RegisterGlobal(agent Agent, opts ...RegisterOption) error
- func TransferTo(target Agent) tool.Tool
- func UpdateContextVariables(ctx context.Context, updates ContextVariables) context.Context
- type AdaptedModule
- type Agent
- type AgentHandler
- type AgentInfo
- type AgentMiddleware
- func DefaultMiddlewares() []AgentMiddleware
- func LoggingMiddleware(logger *log.Logger) AgentMiddleware
- func MetricsMiddleware(collector MetricsCollector) AgentMiddleware
- func ProductionMiddlewares(serviceName string, collector MetricsCollector) []AgentMiddleware
- func RateLimitMiddleware(limiter RateLimiter) AgentMiddleware
- func RecoverMiddleware() AgentMiddleware
- func RetryMiddleware(maxRetries int, backoff time.Duration) AgentMiddleware
- func TimeoutMiddleware(timeout time.Duration) AgentMiddleware
- func TracingMiddleware(serviceName string) AgentMiddleware
- type AgentNetwork
- func (n *AgentNetwork) Broadcast(ctx context.Context, from string, content any) error
- func (n *AgentNetwork) BroadcastToNeighbors(ctx context.Context, from string, content any) error
- func (n *AgentNetwork) Connect(agent1ID, agent2ID string) error
- func (n *AgentNetwork) Disconnect(agent1ID, agent2ID string) error
- func (n *AgentNetwork) GetAgent(agentID string) (Agent, bool)
- func (n *AgentNetwork) GetNeighbors(agentID string) ([]Agent, error)
- func (n *AgentNetwork) GetNode(agentID string) (*NetworkNode, bool)
- func (n *AgentNetwork) HandleMessage(ctx context.Context, msg *NetworkMessage) (*NetworkMessage, error)
- func (n *AgentNetwork) ID() string
- func (n *AgentNetwork) ListAgents() []Agent
- func (n *AgentNetwork) ListOnlineAgents() []Agent
- func (n *AgentNetwork) Multicast(ctx context.Context, from string, to []string, content any) error
- func (n *AgentNetwork) Name() string
- func (n *AgentNetwork) Register(agent Agent) error
- func (n *AgentNetwork) RegisterHandler(topic string, handler MessageHandler)
- func (n *AgentNetwork) Request(ctx context.Context, from, to string, content any) (*NetworkMessage, error)
- func (n *AgentNetwork) Send(ctx context.Context, msg *NetworkMessage) error
- func (n *AgentNetwork) SendTo(ctx context.Context, from, to string, content any) error
- func (n *AgentNetwork) Start(ctx context.Context) error
- func (n *AgentNetwork) Stats() NetworkStats
- func (n *AgentNetwork) Stop()
- func (n *AgentNetwork) Topology() NetworkTopology
- func (n *AgentNetwork) Unregister(agentID string) error
- type AgentState
- type AgentStats
- type AgentStatus
- type AgentTool
- type AgentToolInput
- type AgentToolOption
- type BaseAgent
- func (a *BaseAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *BaseAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *BaseAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *BaseAgent) Config() Config
- func (a *BaseAgent) Description() string
- func (a *BaseAgent) ID() string
- func (a *BaseAgent) InputSchema() *core.Schema
- func (a *BaseAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *BaseAgent) LLM() llm.Provider
- func (a *BaseAgent) Memory() memory.Memory
- func (a *BaseAgent) Name() string
- func (a *BaseAgent) OutputSchema() *core.Schema
- func (a *BaseAgent) Role() Role
- func (a *BaseAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *BaseAgent) SetMemory(mem memory.Memory)
- func (a *BaseAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *BaseAgent) Tools() []tool.Tool
- func (a *BaseAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type Capability
- type CapabilityAssignment
- type CapabilityMatcher
- type CapabilityRequirement
- type CapabilityScorer
- type CapabilitySpec
- type Config
- type ConsensusConfig
- type ConsensusOption
- func WithAgentWeights(weights map[string]float64) ConsensusOption
- func WithConsensusStrategy(strategy ConsensusStrategy) ConsensusOption
- func WithConsensusThreshold(threshold float64) ConsensusOption
- func WithConsensusTimeout(timeout time.Duration) ConsensusOption
- func WithMinParticipation(rate float64) ConsensusOption
- func WithScorer(scorer func(vote Vote) float64) ConsensusOption
- type ConsensusProtocol
- type ConsensusResult
- type ConsensusStrategy
- type Constraint
- type ConstraintType
- type ContextVariables
- type ConvMessage
- type ConversationAgent
- type ConversationOption
- type CostEstimate
- type DeepAgent
- func (d *DeepAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (d *DeepAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (d *DeepAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (d *DeepAgent) Description() string
- func (d *DeepAgent) ID() string
- func (b *DeepAgent) InputSchema() *core.Schema
- func (d *DeepAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (d *DeepAgent) LLM() llm.Provider
- func (d *DeepAgent) Memory() memory.Memory
- func (d *DeepAgent) Name() string
- func (b *DeepAgent) OutputSchema() *core.Schema
- func (d *DeepAgent) Role() Role
- func (d *DeepAgent) Run(ctx context.Context, input Input) (Output, error)
- func (d *DeepAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (d *DeepAgent) Tools() []tool.Tool
- func (d *DeepAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type DeepOption
- type DefaultCapabilityMatcher
- type DefaultCapabilityScorer
- type DefaultStateManager
- func (m *DefaultStateManager) Agent() AgentState
- func (m *DefaultStateManager) Global() GlobalState
- func (m *DefaultStateManager) NewTurn() TurnState
- func (m *DefaultStateManager) Restore(snapshot StateSnapshot) error
- func (m *DefaultStateManager) Session() SessionState
- func (m *DefaultStateManager) Snapshot() StateSnapshot
- func (m *DefaultStateManager) Turn() TurnState
- type DiscoverOption
- type DiscoverQuery
- type GlobalState
- type Handoff
- type HandoffHandler
- type HealthCheckConfig
- type HealthChecker
- type Input
- type LLMReflector
- type LeastConnectionsBalancer
- type LoadBalancer
- type LoopAgent
- func (a *LoopAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *LoopAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *LoopAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *LoopAgent) Description() string
- func (a *LoopAgent) ID() string
- func (b *LoopAgent) InputSchema() *core.Schema
- func (a *LoopAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *LoopAgent) LLM() llm.Provider
- func (a *LoopAgent) Memory() memory.Memory
- func (a *LoopAgent) Name() string
- func (b *LoopAgent) OutputSchema() *core.Schema
- func (a *LoopAgent) Role() Role
- func (a *LoopAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *LoopAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *LoopAgent) Tools() []tool.Tool
- func (a *LoopAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type LoopOption
- type MemorySetter
- type Message
- type MessageHandler
- type MessageRouter
- func (r *MessageRouter) Broadcast(ctx context.Context, msg *NetworkMessage) error
- func (r *MessageRouter) BroadcastToNeighbors(ctx context.Context, msg *NetworkMessage) error
- func (r *MessageRouter) Multicast(ctx context.Context, msg *NetworkMessage, targets []string) error
- func (r *MessageRouter) RequestResponse(ctx context.Context, msg *NetworkMessage) (*NetworkMessage, error)
- func (r *MessageRouter) Route(ctx context.Context, msg *NetworkMessage) error
- func (r *MessageRouter) Start(ctx context.Context)
- func (r *MessageRouter) Stop()
- type MessageType
- type MetricsCollector
- type MiddlewareChain
- func (c *MiddlewareChain) Len() int
- func (c *MiddlewareChain) Prepend(middlewares ...AgentMiddleware) *MiddlewareChain
- func (c *MiddlewareChain) Use(middlewares ...AgentMiddleware) *MiddlewareChain
- func (c *MiddlewareChain) Wrap(handler AgentHandler) AgentHandler
- func (c *MiddlewareChain) WrapAgent(agent Agent) AgentHandler
- type NegotiationPreferences
- type NegotiationRequest
- type NegotiationResult
- type Negotiator
- func (n *Negotiator) DeclareCapability(agentID string, capability CapabilitySpec) error
- func (n *Negotiator) ExportCapabilities(agentID string) ([]byte, error)
- func (n *Negotiator) ImportCapabilities(agentID string, data []byte) error
- func (n *Negotiator) Negotiate(ctx context.Context, req *NegotiationRequest) (*NegotiationResult, error)
- func (n *Negotiator) QueryCapabilities(agentID string) ([]CapabilitySpec, error)
- type NegotiatorOption
- type NetworkMessage
- type NetworkNode
- type NetworkOption
- type NetworkStats
- type NetworkTopology
- type NodeStatus
- type Option
- func WithCheckpointer(cp checkpoint.Checkpointer) Option
- func WithDescription(desc string) Option
- func WithID(id string) Option
- func WithLLM(provider llm.Provider) Option
- func WithMaxIterations(n int) Option
- func WithMemory(mem memory.Memory) Option
- func WithMiddleware(mws ...agentruntime.Middleware) Option
- func WithName(name string) Option
- func WithRole(role Role) Option
- func WithStrategy(s agentruntime.Strategy) Option
- func WithSystemPrompt(prompt string) Option
- func WithTools(tools ...tool.Tool) Option
- func WithVerbose(v bool) Option
- type Output
- type ParallelAgent
- func (a *ParallelAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *ParallelAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *ParallelAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *ParallelAgent) Description() string
- func (a *ParallelAgent) ID() string
- func (b *ParallelAgent) InputSchema() *core.Schema
- func (a *ParallelAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *ParallelAgent) LLM() llm.Provider
- func (a *ParallelAgent) Memory() memory.Memory
- func (a *ParallelAgent) Name() string
- func (b *ParallelAgent) OutputSchema() *core.Schema
- func (a *ParallelAgent) Role() Role
- func (a *ParallelAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *ParallelAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *ParallelAgent) Tools() []tool.Tool
- func (a *ParallelAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type ParallelOption
- type ParameterSpec
- type PlanExecuteAgent
- func (a *PlanExecuteAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *PlanExecuteAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *PlanExecuteAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *PlanExecuteAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *PlanExecuteAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *PlanExecuteAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *PlanExecuteAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type PlanExecuteOption
- type Poll
- type ProxyConfig
- type ProxyOption
- type RateLimiter
- type ReActAgent
- func (a *ReActAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *ReActAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *ReActAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *ReActAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *ReActAgent) Resume(ctx context.Context, runID string, input Input) (Output, error)
- func (a *ReActAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *ReActAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *ReActAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type ReasoningModule
- type Reflection
- type ReflectionAgent
- func (a *ReflectionAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *ReflectionAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *ReflectionAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *ReflectionAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *ReflectionAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *ReflectionAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *ReflectionAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type ReflectionOption
- type Reflector
- type RegisterOption
- type Registry
- func (r *Registry) Deregister(id string) error
- func (r *Registry) Discover(opts ...DiscoverOption) []*AgentInfo
- func (r *Registry) DiscoverOne(opts ...DiscoverOption) (*AgentInfo, error)
- func (r *Registry) Get(id string) (*AgentInfo, bool)
- func (r *Registry) GetAgent(id string) (Agent, bool)
- func (r *Registry) Heartbeat(id string) error
- func (r *Registry) List() []*AgentInfo
- func (r *Registry) Register(info *AgentInfo) error
- func (r *Registry) RegisterAgent(agent Agent, opts ...RegisterOption) error
- func (r *Registry) SetLoadBalancer(lb LoadBalancer)
- func (r *Registry) Unwatch(id string)
- func (r *Registry) Watch(callback WatchCallback) string
- type RegistryConfig
- type ReplaySafety
- type ReplaySafetyAware
- type Role
- type RoleBuilder
- func (b *RoleBuilder) AllowDelegation(allow bool) *RoleBuilder
- func (b *RoleBuilder) Backstory(backstory string) *RoleBuilder
- func (b *RoleBuilder) Build() Role
- func (b *RoleBuilder) Constraints(constraints ...string) *RoleBuilder
- func (b *RoleBuilder) DelegateTo(agents ...string) *RoleBuilder
- func (b *RoleBuilder) Expertise(areas ...string) *RoleBuilder
- func (b *RoleBuilder) Goal(goal string) *RoleBuilder
- func (b *RoleBuilder) Personality(personality string) *RoleBuilder
- func (b *RoleBuilder) Title(title string) *RoleBuilder
- func (b *RoleBuilder) Tools(tools ...string) *RoleBuilder
- type RoundRobinBalancer
- type SLASpec
- type SafeContextVariables
- func (s *SafeContextVariables) Clone() ContextVariables
- func (s *SafeContextVariables) CloneSafe() *SafeContextVariables
- func (s *SafeContextVariables) Delete(key string)
- func (s *SafeContextVariables) Get(key string) (any, bool)
- func (s *SafeContextVariables) Keys() []string
- func (s *SafeContextVariables) Len() int
- func (s *SafeContextVariables) Merge(other ContextVariables)
- func (s *SafeContextVariables) MergeSafe(other *SafeContextVariables)
- func (s *SafeContextVariables) Set(key string, value any)
- type ScoreWeights
- type SelfDiscoveryAgent
- func (a *SelfDiscoveryAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *SelfDiscoveryAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *SelfDiscoveryAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *SelfDiscoveryAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *SelfDiscoveryAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *SelfDiscoveryAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *SelfDiscoveryAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type SelfDiscoveryOption
- type SequentialAgent
- func (a *SequentialAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (a *SequentialAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *SequentialAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (a *SequentialAgent) Description() string
- func (a *SequentialAgent) ID() string
- func (b *SequentialAgent) InputSchema() *core.Schema
- func (a *SequentialAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (a *SequentialAgent) LLM() llm.Provider
- func (a *SequentialAgent) Memory() memory.Memory
- func (a *SequentialAgent) Name() string
- func (b *SequentialAgent) OutputSchema() *core.Schema
- func (a *SequentialAgent) Role() Role
- func (a *SequentialAgent) Run(ctx context.Context, input Input) (Output, error)
- func (a *SequentialAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (a *SequentialAgent) Tools() []tool.Tool
- func (a *SequentialAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type SessionState
- type SharedMemory
- func (sm *SharedMemory) Clear(ctx context.Context) error
- func (sm *SharedMemory) Delete(ctx context.Context, id string) error
- func (sm *SharedMemory) Get(ctx context.Context, id string) (*memory.Entry, error)
- func (sm *SharedMemory) Save(ctx context.Context, entry memory.Entry) error
- func (sm *SharedMemory) SaveBatch(ctx context.Context, entries []memory.Entry) error
- func (sm *SharedMemory) Search(ctx context.Context, query memory.SearchQuery) ([]memory.Entry, error)
- func (sm *SharedMemory) Stats() memory.MemoryStats
- type SharedMemoryConfig
- type SharedMemoryOption
- type SharedMemoryProxy
- func (p *SharedMemoryProxy) Clear(ctx context.Context) error
- func (p *SharedMemoryProxy) Delete(ctx context.Context, id string) error
- func (p *SharedMemoryProxy) Get(ctx context.Context, id string) (*memory.Entry, error)
- func (p *SharedMemoryProxy) Local() memory.Memory
- func (p *SharedMemoryProxy) Save(ctx context.Context, entry memory.Entry) error
- func (p *SharedMemoryProxy) SaveBatch(ctx context.Context, entries []memory.Entry) error
- func (p *SharedMemoryProxy) Search(ctx context.Context, query memory.SearchQuery) ([]memory.Entry, error)
- func (p *SharedMemoryProxy) Stats() memory.MemoryStats
- type StateManager
- type StateSnapshot
- type SubtaskInput
- type Summarizer
- type SupervisorAgent
- func (s *SupervisorAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (s *SupervisorAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (s *SupervisorAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (s *SupervisorAgent) Description() string
- func (s *SupervisorAgent) ID() string
- func (b *SupervisorAgent) InputSchema() *core.Schema
- func (s *SupervisorAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (s *SupervisorAgent) LLM() llm.Provider
- func (s *SupervisorAgent) Memory() memory.Memory
- func (s *SupervisorAgent) Name() string
- func (b *SupervisorAgent) OutputSchema() *core.Schema
- func (s *SupervisorAgent) Role() Role
- func (s *SupervisorAgent) Run(ctx context.Context, input Input) (Output, error)
- func (s *SupervisorAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (s *SupervisorAgent) Tools() []tool.Tool
- func (s *SupervisorAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type SupervisorOption
- type SwarmRunner
- type Team
- func (t *Team) AddAgent(agent Agent)
- func (t *Team) Agents() []Agent
- func (t *Team) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
- func (t *Team) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (t *Team) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
- func (t *Team) Description() string
- func (t *Team) ID() string
- func (t *Team) InputSchema() *core.Schema
- func (t *Team) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
- func (t *Team) Mode() TeamMode
- func (t *Team) Name() string
- func (t *Team) OutputSchema() *core.Schema
- func (t *Team) RemoveAgent(agentID string)
- func (t *Team) Run(ctx context.Context, input Input) (Output, error)
- func (t *Team) SharedMemory() *SharedMemory
- func (t *Team) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
- func (t *Team) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
- type TeamMode
- type TeamOption
- func WithAgents(agents ...Agent) TeamOption
- func WithGlobalState(state GlobalState) TeamOption
- func WithManager(manager Agent) TeamOption
- func WithMaxRounds(rounds int) TeamOption
- func WithMode(mode TeamMode) TeamOption
- func WithSharedMemory(sm *SharedMemory) TeamOption
- func WithTeamDescription(desc string) TeamOption
- func WithTeamVerbose(verbose bool) TeamOption
- type ToolCall
- type ToolCallRecord
- type TransferToInput
- type TurnState
- type Vote
- type WatchCallback
- type WatchEvent
- type WatchEventType
- type WeightedBalancer
Constants ¶
const ( // ReplayUnsafe 默认:重放可能重复副作用(发邮件 / 扣款 / 写库等)。 ReplayUnsafe = agentruntime.SideEffectUnsafe // ReplayIdempotent 幂等:重放产生相同结果、无额外副作用。 ReplayIdempotent = agentruntime.SideEffectIdempotent // ReplayReadOnly 只读:无任何副作用。 ReplayReadOnly = agentruntime.SideEffectReadOnly )
重放安全性等级(用于工具声明 ReplaySafety;语义见 runtime.ToolSideEffect)。
用于 Durable exactly-once:当一个工具步在崩溃后被 Resume 时,框架据此判定能否安全 重放——只读 / 幂等可重放,否则 fail-closed 拒绝重放、避免重复副作用。
Variables ¶
var ( // CriticalThinkingModule 批判性思维模块 CriticalThinkingModule = ReasoningModule{ Name: "批判性思维", Description: "评估论点的有效性、识别假设和偏见、分析证据的可靠性", Template: "1. 识别核心论点\n2. 找出隐含假设\n3. 评估证据质量\n4. 考虑反对意见\n5. 得出结论", } // StepByStepModule 逐步推理模块 StepByStepModule = ReasoningModule{ Name: "逐步推理", Description: "将复杂问题分解为更小的可管理步骤,按顺序解决", Template: "1. 理解问题\n2. 分解为子问题\n3. 逐个解决\n4. 整合结果", } // CreativeThinkingModule 创造性思维模块 CreativeThinkingModule = ReasoningModule{ Name: "创造性思维", Description: "生成新颖的想法和解决方案,打破常规思维", Template: "1. 重新定义问题\n2. 头脑风暴多种方案\n3. 组合不同想法\n4. 评估可行性", } // SystemAnalysisModule 系统分析模块 SystemAnalysisModule = ReasoningModule{ Name: "系统分析", Description: "理解系统组件之间的关系、因果链和反馈循环", Template: "1. 识别系统组件\n2. 分析组件关系\n3. 找出关键节点\n4. 预测系统行为", } // AnalogicalReasoningModule 类比推理模块 AnalogicalReasoningModule = ReasoningModule{ Name: "类比推理", Description: "利用已知领域的知识来理解新领域", Template: "1. 找到相似的已知情况\n2. 映射相似性\n3. 转移解决方案\n4. 验证适用性", } // InductiveReasoningModule 归纳推理模块 InductiveReasoningModule = ReasoningModule{ Name: "归纳推理", Description: "从具体观察中推导一般规律", Template: "1. 收集具体案例\n2. 识别模式\n3. 形成假设\n4. 验证推广", } // DeductiveReasoningModule 演绎推理模块 DeductiveReasoningModule = ReasoningModule{ Name: "演绎推理", Description: "从一般原则推导具体结论", Template: "1. 确定前提\n2. 应用逻辑规则\n3. 推导结论\n4. 验证有效性", } // DecompositionModule 问题分解模块 DecompositionModule = ReasoningModule{ Name: "问题分解", Description: "将大问题分解为更小、更容易处理的子问题", Template: "1. 识别问题边界\n2. 划分子问题\n3. 确定依赖关系\n4. 制定解决顺序", } // DefaultReasoningModules 默认推理模块集 DefaultReasoningModules = []ReasoningModule{ CriticalThinkingModule, StepByStepModule, CreativeThinkingModule, SystemAnalysisModule, AnalogicalReasoningModule, InductiveReasoningModule, DeductiveReasoningModule, DecompositionModule, } )
预定义的推理模块
var DefaultRegistryConfig = RegistryConfig{ HealthCheckInterval: 10 * time.Second, HeartbeatTimeout: 30 * time.Second, DeregisterAfter: 60 * time.Second, EnableHealthCheck: true, }
DefaultRegistryConfig 默认配置
var DefaultScoreWeights = ScoreWeights{
VersionMatch: 0.2,
ConstraintMatch: 0.3,
CostWeight: 0.2,
SLAWeight: 0.3,
}
DefaultScoreWeights 默认权重
var DeveloperRole = NewRole("developer").
Title("Senior Software Developer").
Goal("Design and implement high-quality software solutions").
Backstory("You are an experienced developer with expertise in multiple programming languages and best practices.").
Expertise("software development", "code review", "debugging", "system design").
Personality("logical, detail-oriented, problem-solver").
Constraints("Write clean, maintainable code", "Follow security best practices").
Build()
DeveloperRole 开发者角色
var ResearcherRole = NewRole("researcher").
Title("Senior Research Analyst").
Goal("Uncover cutting-edge developments and provide insightful analysis").
Backstory("You are an expert analyst with 10 years of experience in research. You have a keen eye for detail and are excellent at synthesizing complex information.").
Expertise("research", "analysis", "data interpretation", "report writing").
Personality("thorough, analytical, curious, objective").
Constraints("Always cite sources", "Verify information from multiple sources").
Build()
ResearcherRole 研究员角色
var WriterRole = NewRole("writer").
Title("Content Writer").
Goal("Create engaging and informative content").
Backstory("You are a skilled writer with expertise in creating compelling narratives and clear explanations.").
Expertise("writing", "editing", "storytelling", "content creation").
Personality("creative, articulate, adaptable").
Constraints("Maintain consistent tone", "Follow style guidelines").
Build()
WriterRole 作家角色
Functions ¶
func AgentAsTool ¶
AgentAsTool 将 Agent 包装为 Tool
使 Agent 可以被其他 Agent 作为工具调用。与 TransferTo 不同, AgentAsTool 直接执行目标 Agent 并返回结果,而不是创建交接。
使用场景:
- SupervisorAgent 将 worker Agent 注册为工具
- 让一个 Agent 调用另一个 Agent 的能力
工具名称:agent_<name> 输入参数:message (必填) + context (可选)
func ContextWithSafeVariables ¶
func ContextWithSafeVariables(ctx context.Context, vars *SafeContextVariables) context.Context
ContextWithSafeVariables 将线程安全的上下文变量添加到 context
func ContextWithStateManager ¶
func ContextWithStateManager(ctx context.Context, sm StateManager) context.Context
ContextWithStateManager 将 StateManager 添加到 context
func ContextWithVariables ¶
func ContextWithVariables(ctx context.Context, vars ContextVariables) context.Context
ContextWithVariables 将上下文变量添加到 context
func RegisterGlobal ¶
func RegisterGlobal(agent Agent, opts ...RegisterOption) error
RegisterGlobal 注册到全局注册表
func TransferTo ¶
TransferTool 创建转交工具 通过工具调用触发的 Agent 交接(handoff)机制。
func UpdateContextVariables ¶
func UpdateContextVariables(ctx context.Context, updates ContextVariables) context.Context
UpdateContextVariables 更新 context 中的变量
Types ¶
type AdaptedModule ¶
AdaptedModule 适配后的模块
type Agent ¶
type Agent interface {
core.Runnable[Input, Output]
// ID 返回 Agent 唯一标识
ID() string
// Role 返回 Agent 的角色定义
Role() Role
// Tools 返回 Agent 可用的工具列表
Tools() []tool.Tool
// Memory 返回 Agent 的记忆系统
Memory() memory.Memory
// LLM 返回 Agent 使用的 LLM Provider
LLM() llm.Provider
// Run 执行 Agent(向后兼容方法)
// Deprecated: 请使用 Invoke
Run(ctx context.Context, input Input) (Output, error)
}
Agent 是 AI Agent 的核心接口 继承 Runnable 接口,添加 Agent 特有的方法
type AgentInfo ¶
type AgentInfo struct {
// ID Agent 唯一标识
ID string `json:"id"`
// Name Agent 名称
Name string `json:"name"`
// Description Agent 描述
Description string `json:"description,omitempty"`
// Version 版本号
Version string `json:"version,omitempty"`
// Tags 标签列表
Tags []string `json:"tags,omitempty"`
// Capabilities 能力列表
Capabilities []Capability `json:"capabilities,omitempty"`
// Endpoint 服务端点
Endpoint string `json:"endpoint,omitempty"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
// Status 状态
Status AgentStatus `json:"status"`
// RegisteredAt 注册时间
RegisteredAt time.Time `json:"registered_at"`
// LastHeartbeat 最后心跳时间
LastHeartbeat time.Time `json:"last_heartbeat"`
// HealthCheck 健康检查配置
HealthCheck *HealthCheckConfig `json:"health_check,omitempty"`
// Weight 权重(用于负载均衡)
Weight int `json:"weight,omitempty"`
// MaxConcurrency 最大并发数
MaxConcurrency int `json:"max_concurrency,omitempty"`
// CurrentLoad 当前负载
CurrentLoad int `json:"current_load,omitempty"`
// contains filtered or unexported fields
}
AgentInfo Agent 信息
func DiscoverGlobal ¶
func DiscoverGlobal(opts ...DiscoverOption) []*AgentInfo
DiscoverGlobal 从全局注册表发现
func DiscoverOneGlobal ¶
func DiscoverOneGlobal(opts ...DiscoverOption) (*AgentInfo, error)
DiscoverOneGlobal 从全局注册表发现单个
type AgentMiddleware ¶
type AgentMiddleware func(next AgentHandler) AgentHandler
AgentMiddleware Agent 中间件函数
接收下一个处理器,返回一个包装后的处理器 中间件可以:
- 在调用前后执行逻辑
- 修改输入或输出
- 拦截请求
- 处理错误
func DefaultMiddlewares ¶
func DefaultMiddlewares() []AgentMiddleware
DefaultMiddlewares 返回默认的中间件组合
包含:
- RecoverMiddleware: panic 恢复
- LoggingMiddleware: 日志记录
- TimeoutMiddleware: 超时控制(默认 60 秒)
使用示例:
chain := NewMiddlewareChain(DefaultMiddlewares()...)
func LoggingMiddleware ¶
func LoggingMiddleware(logger *log.Logger) AgentMiddleware
LoggingMiddleware 日志记录中间件
记录请求开始、结束、耗时和错误信息
参数:
- logger: 可选的日志记录器,nil 时使用标准库 log
使用示例:
chain.Use(LoggingMiddleware(nil))
func MetricsMiddleware ¶
func MetricsMiddleware(collector MetricsCollector) AgentMiddleware
MetricsMiddleware 指标采集中间件
收集 Agent 执行的各种指标
参数:
- collector: 指标收集器
使用示例:
collector := NewMetricsCollector() chain.Use(MetricsMiddleware(collector))
func ProductionMiddlewares ¶
func ProductionMiddlewares(serviceName string, collector MetricsCollector) []AgentMiddleware
ProductionMiddlewares 返回生产环境推荐的中间件组合
包含:
- RecoverMiddleware: panic 恢复
- TracingMiddleware: 追踪
- MetricsMiddleware: 指标采集
- TimeoutMiddleware: 超时控制
- RetryMiddleware: 重试
参数:
- serviceName: 服务名称
- collector: 指标收集器
使用示例:
chain := NewMiddlewareChain(ProductionMiddlewares("my-service", collector)...)
func RateLimitMiddleware ¶
func RateLimitMiddleware(limiter RateLimiter) AgentMiddleware
RateLimitMiddleware 限流中间件
限制 Agent 的调用频率
参数:
- limiter: 限流器实例
使用示例:
limiter := rate.NewLimiter(10, 100) // 10 QPS,突发 100 chain.Use(RateLimitMiddleware(limiter))
func RecoverMiddleware ¶
func RecoverMiddleware() AgentMiddleware
RecoverMiddleware panic 恢复中间件
捕获处理过程中的 panic,转换为错误返回 防止单个请求的 panic 导致整个服务崩溃
使用示例:
chain.Use(RecoverMiddleware())
func RetryMiddleware ¶
func RetryMiddleware(maxRetries int, backoff time.Duration) AgentMiddleware
RetryMiddleware 重试中间件
在失败时自动重试,使用指数退避策略。 退避时间 = min(backoff * 2^attempt, maxBackoffDuration)
参数:
- maxRetries: 最大重试次数
- backoff: 基础重试间隔
使用示例:
chain.Use(RetryMiddleware(3, 1*time.Second))
func TimeoutMiddleware ¶
func TimeoutMiddleware(timeout time.Duration) AgentMiddleware
TimeoutMiddleware 超时控制中间件
设置请求处理的最大时间
参数:
- timeout: 超时时间
使用示例:
chain.Use(TimeoutMiddleware(30 * time.Second))
func TracingMiddleware ¶
func TracingMiddleware(serviceName string) AgentMiddleware
TracingMiddleware 追踪中间件
添加追踪信息到上下文和输出
参数:
- serviceName: 服务名称
使用示例:
chain.Use(TracingMiddleware("my-agent"))
type AgentNetwork ¶
type AgentNetwork struct {
// contains filtered or unexported fields
}
AgentNetwork 多 Agent 网络
func NewAgentNetwork ¶
func NewAgentNetwork(name string, opts ...NetworkOption) *AgentNetwork
NewAgentNetwork 创建 Agent 网络
func (*AgentNetwork) BroadcastToNeighbors ¶
BroadcastToNeighbors 广播消息给邻居节点
func (*AgentNetwork) Connect ¶
func (n *AgentNetwork) Connect(agent1ID, agent2ID string) error
Connect 手动连接两个节点(用于 CustomTopology)
func (*AgentNetwork) Disconnect ¶
func (n *AgentNetwork) Disconnect(agent1ID, agent2ID string) error
Disconnect 断开两个节点的连接
func (*AgentNetwork) GetAgent ¶
func (n *AgentNetwork) GetAgent(agentID string) (Agent, bool)
GetAgent 获取 Agent
func (*AgentNetwork) GetNeighbors ¶
func (n *AgentNetwork) GetNeighbors(agentID string) ([]Agent, error)
GetNeighbors 获取邻居节点
func (*AgentNetwork) GetNode ¶
func (n *AgentNetwork) GetNode(agentID string) (*NetworkNode, bool)
GetNode 获取节点
func (*AgentNetwork) HandleMessage ¶
func (n *AgentNetwork) HandleMessage(ctx context.Context, msg *NetworkMessage) (*NetworkMessage, error)
HandleMessage 处理消息
func (*AgentNetwork) ListOnlineAgents ¶
func (n *AgentNetwork) ListOnlineAgents() []Agent
ListOnlineAgents 列出在线 Agent
func (*AgentNetwork) Register ¶
func (n *AgentNetwork) Register(agent Agent) error
Register 注册 Agent 到网络
func (*AgentNetwork) RegisterHandler ¶
func (n *AgentNetwork) RegisterHandler(topic string, handler MessageHandler)
RegisterHandler 注册消息处理器
func (*AgentNetwork) Request ¶
func (n *AgentNetwork) Request(ctx context.Context, from, to string, content any) (*NetworkMessage, error)
Request 发送请求并等待响应
func (*AgentNetwork) Send ¶
func (n *AgentNetwork) Send(ctx context.Context, msg *NetworkMessage) error
Send 发送消息给指定 Agent
func (*AgentNetwork) Stop ¶
func (n *AgentNetwork) Stop()
Stop 停止网络
线程安全:此方法使用安全的 channel 关闭机制,不会因为重复关闭或并发发送而 panic。
func (*AgentNetwork) Unregister ¶
func (n *AgentNetwork) Unregister(agentID string) error
Unregister 从网络注销 Agent
线程安全:此方法使用安全的 channel 关闭机制,不会因为重复关闭或并发发送而 panic。
type AgentState ¶
type AgentState interface {
// Get 获取值
Get(key string) (any, bool)
// Set 设置值
Set(key string, value any)
// Delete 删除值
Delete(key string)
// All 获取所有键值对
All() map[string]any
// Stats 获取 Agent 统计信息
Stats() AgentStats
// UpdateStats 更新统计信息
UpdateStats(fn func(*AgentStats))
}
AgentState Agent 持久状态 生命周期:Agent 实例
type AgentStats ¶
type AgentStats struct {
TotalRuns int64 `json:"total_runs"`
SuccessfulRuns int64 `json:"successful_runs"`
FailedRuns int64 `json:"failed_runs"`
TotalTokens int64 `json:"total_tokens"`
PromptTokens int64 `json:"prompt_tokens"`
CompletionTokens int64 `json:"completion_tokens"`
TotalToolCalls int64 `json:"total_tool_calls"`
TotalDuration time.Duration `json:"total_duration"`
LastRunAt time.Time `json:"last_run_at"`
}
AgentStats Agent 统计信息
type AgentStatus ¶
type AgentStatus string
AgentStatus Agent 状态
const ( // StatusUnknown 未知状态 StatusUnknown AgentStatus = "unknown" // StatusHealthy 健康状态 StatusHealthy AgentStatus = "healthy" // StatusUnhealthy 不健康状态 StatusUnhealthy AgentStatus = "unhealthy" // StatusDraining 正在排空 StatusDraining AgentStatus = "draining" // StatusOffline 离线 StatusOffline AgentStatus = "offline" )
type AgentTool ¶
type AgentTool struct {
// contains filtered or unexported fields
}
AgentTool 把一个 Agent 适配为 tool.Tool,使任意 agent loop(ReAct / PlanExecute / Reflection 等)都能把另一个 Agent 当作工具来调用,从而构成「递归子链」—— agent 调 agent、loop 套 loop。
设计意图(A10 统一 agent loop 的组合落点):与其把三种 agent 强行合并进同一个 回合循环(会丢失 PlanExecute 的规划/重计划、Reflection 的自检等各自特性),不如 让它们各自保留循环、通过统一的「agent 即工具」基元相互嵌套。这样:
- DeepAgent 的递归分解、Swarm 的 handoff 等专用组合可统一表达为「子 agent 作工具」;
- 任意 agent 都能作为子链节点被另一 agent 调度,无需为每种组合各写一套适配。
入参约定为单字段对象 {"query": string},即交给子 Agent 的子任务文本。
func NewAgentTool ¶
func NewAgentTool(a Agent, opts ...AgentToolOption) *AgentTool
NewAgentTool 把 agent 包装为可被其它 agent 调用的工具。
默认工具名取 agent.Name(),描述取 agent.Description()(为空则用通用说明), 入参字段为 "query"。
func (*AgentTool) Execute ¶
Execute 运行子 Agent 并把其最终回复作为工具结果返回。
子 Agent 出错时以失败 Result(而非 error)返回,避免单个子链节点失败直接中断 上层 agent 的循环——上层可据此决定重试/重计划/换路。
type AgentToolInput ¶
type AgentToolInput struct {
// Message 传递给 Agent 的消息
Message string `json:"message" desc:"Message to send to the agent" required:"true"`
// Context 额外上下文
Context map[string]any `json:"context,omitempty" desc:"Additional context to pass to the agent"`
}
AgentToolInput AgentAsTool 的输入参数
type AgentToolOption ¶
type AgentToolOption func(*AgentTool)
AgentToolOption 配置 AgentTool。
func WithAgentToolDescription ¶
func WithAgentToolDescription(desc string) AgentToolOption
WithAgentToolDescription 覆盖工具描述(默认取 agent.Description() 或通用说明)。
func WithAgentToolName ¶
func WithAgentToolName(name string) AgentToolOption
WithAgentToolName 覆盖工具名(默认取 agent.Name())。
func WithAgentToolQueryKey ¶
func WithAgentToolQueryKey(key string) AgentToolOption
WithAgentToolQueryKey 覆盖入参字段名(默认 "query")。
type BaseAgent ¶
type BaseAgent struct {
// contains filtered or unexported fields
}
BaseAgent 提供 Agent 的基础实现
func (*BaseAgent) Batch ¶
func (a *BaseAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*BaseAgent) BatchStream ¶
func (a *BaseAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*BaseAgent) Collect ¶
func (a *BaseAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*BaseAgent) InputSchema ¶
InputSchema 返回输入 Schema
func (*BaseAgent) OutputSchema ¶
OutputSchema 返回输出 Schema
type Capability ¶
type Capability struct {
// Name 能力名称
Name string `json:"name"`
// Description 能力描述
Description string `json:"description,omitempty"`
// Version 能力版本
Version string `json:"version,omitempty"`
// InputSchema 输入 Schema
InputSchema map[string]any `json:"input_schema,omitempty"`
// OutputSchema 输出 Schema
OutputSchema map[string]any `json:"output_schema,omitempty"`
// Constraints 约束条件
Constraints map[string]any `json:"constraints,omitempty"`
}
Capability Agent 能力定义
type CapabilityAssignment ¶
type CapabilityAssignment struct {
// Requirement 需求
Requirement *CapabilityRequirement `json:"requirement"`
// Agent 分配的 Agent
Agent *AgentInfo `json:"agent"`
// Capability 匹配的能力
Capability *CapabilitySpec `json:"capability"`
// Score 匹配分数
Score float64 `json:"score"`
}
CapabilityAssignment 能力分配
type CapabilityMatcher ¶
type CapabilityMatcher interface {
// Match 检查 Agent 能力是否满足需求
Match(requirement *CapabilityRequirement, capability *CapabilitySpec) bool
}
CapabilityMatcher 能力匹配器接口
type CapabilityRequirement ¶
type CapabilityRequirement struct {
// Name 需要的能力名称
Name string `json:"name"`
// Version 需要的版本(可选)
Version string `json:"version,omitempty"`
// Parameters 参数要求
Parameters map[string]any `json:"parameters,omitempty"`
// Constraints 约束要求
Constraints []Constraint `json:"constraints,omitempty"`
// Priority 优先级(1-10)
Priority int `json:"priority,omitempty"`
// Optional 是否可选
Optional bool `json:"optional,omitempty"`
}
CapabilityRequirement 能力需求
type CapabilityScorer ¶
type CapabilityScorer interface {
// Score 计算能力匹配分数
Score(requirement *CapabilityRequirement, capability *CapabilitySpec) float64
}
CapabilityScorer 能力评分器接口
type CapabilitySpec ¶
type CapabilitySpec struct {
// Name 能力名称
Name string `json:"name"`
// Version 能力版本
Version string `json:"version,omitempty"`
// Description 能力描述
Description string `json:"description,omitempty"`
// Category 能力分类
Category string `json:"category,omitempty"`
// InputSchema 输入 Schema
InputSchema map[string]any `json:"input_schema,omitempty"`
// OutputSchema 输出 Schema
OutputSchema map[string]any `json:"output_schema,omitempty"`
// Parameters 参数定义
Parameters []ParameterSpec `json:"parameters,omitempty"`
// Constraints 约束条件
Constraints []Constraint `json:"constraints,omitempty"`
// Dependencies 依赖的能力
Dependencies []string `json:"dependencies,omitempty"`
// Cost 能力成本估算
Cost *CostEstimate `json:"cost,omitempty"`
// SLA 服务级别协议
SLA *SLASpec `json:"sla,omitempty"`
}
CapabilitySpec 能力规格
type Config ¶
type Config struct {
// ID Agent 唯一标识
ID string
// Name Agent 名称
Name string
// Description Agent 描述
Description string
// Role Agent 角色定义
Role Role
// SystemPrompt 系统提示词
SystemPrompt string
// LLM LLM 提供者
LLM llm.Provider
// Tools 可用工具列表
Tools []tool.Tool
// Memory 记忆系统
Memory memory.Memory
// MaxIterations 最大迭代次数(防止无限循环)
MaxIterations int
// Verbose 是否输出详细日志
Verbose bool
// Middleware 注入到底层 runtime 的中间件链(按序在每步 BeforeLLM/AfterLLM/
// BeforeTool/AfterTool/Finalize 触发)。这是 runtime 中间件扩展点在 Agent 层的出口。
//
// 典型用途——共享同一个 cost.Controller 接入统一预算控制:
//
// ctrl, err := cost.NewController(cost.WithBudget(10.0))
// if err != nil {
// return nil, err
// }
// // 在 provider/调用方每次真正发起 LLM 请求前执行:
// if err := ctrl.CheckRequest(ctx, estimatedTokens); err != nil {
// return nil, err
// }
// agent := NewReAct(WithLLM(p), WithMiddleware(middleware.NewBudgetControl(
// middleware.BudgetControlConfig{
// Limits: middleware.BudgetLimits{MaxCostUSD: 10.0, MaxTokens: 100000},
// Cost: ctrl.BudgetCostFunc(),
// Record: ctrl.RecordUsageFunc(),
// },
// )))
//
// 为空时(默认)底层 runner 不挂任何中间件,行为不变。
//
// NewBudgetControl 是首选入口,内部组合两层语义:Budget 按每次 runtime run
// 检查 token/墙钟/成本,CostControl 通过 RecordUsageFunc 把每次 LLM 响应写入
// 共享 Controller,因而对 PlanExecute/Reflection 等多 run agent 也提供全程累计
// 封顶。CheckRequest 是发请求前的 token/频率预检,不由中间件代替;三者
// 应共享同一 Controller。直接挂 Budget 或 CostControl 仅适合明确只需底层
// per-run 或 cross-run 单层语义的高级用法。
Middleware []agentruntime.Middleware
// Durable 可选:开启可持久化/可恢复执行(经 WithCheckpointer 设置)。
//
// 非 nil 时,Agent 的底层 runtime run 会在每步边界持久化快照,并支持经 Resume
// 从最近快照续跑。nil 时(默认)不触碰持久化、行为不变。
//
// 语义按 agent 类型:仅 ReActAgent(单次多轮 run)有干净的"整次执行可恢复"语义;
// 多 run agent(PlanExecute/Reflection)每次内部调用是独立 run,Durable 不适用其整体恢复。
Durable agentruntime.DurableExecution
// Strategy 可选:选择统一 agent loop 的执行策略(经 WithStrategy 设置)。
//
// 统一 runtime 的回合循环由 Strategy 定制(系统前缀 / 是否继续 / 收尾)。nil 时
// 默认 NoopStrategy(等价 ReAct)。借助 runtime/strategy 包可让同一个 ReActAgent
// 以 ReAct / PlanExecute / Reflection 三种策略在**同一个统一回合循环**上运行
// (提示词引导式),无需各自独立的 loop 实现。
//
// 注:独立的 PlanExecuteAgent / ReflectionAgent 是功能更丰富的多调用编排实现,
// 与本"统一 loop + 策略"轻量路径并存,互不影响。
Strategy agentruntime.Strategy
}
Config 是 Agent 的配置
type ConsensusConfig ¶
type ConsensusConfig struct {
// Strategy 共识策略
Strategy ConsensusStrategy
// Threshold 阈值(用于多数投票,默认 0.5)
Threshold float64
// Timeout 超时时间
Timeout time.Duration
// MinParticipation 最小参与率(默认 0.5)
MinParticipation float64
// Weights Agent 权重(用于加权投票)
Weights map[string]float64
// Scorer 评分函数(用于最佳选择)
Scorer func(vote Vote) float64
// Validator 投票验证函数
Validator func(vote Vote) bool
// AllowAbstain 允许弃权
AllowAbstain bool
}
ConsensusConfig 共识配置
func DefaultConsensusConfig ¶
func DefaultConsensusConfig() ConsensusConfig
DefaultConsensusConfig 返回默认配置
type ConsensusOption ¶
type ConsensusOption func(*ConsensusConfig)
ConsensusOption 共识配置选项
func WithAgentWeights ¶
func WithAgentWeights(weights map[string]float64) ConsensusOption
WithAgentWeights 设置 Agent 权重
func WithConsensusStrategy ¶
func WithConsensusStrategy(strategy ConsensusStrategy) ConsensusOption
WithConsensusStrategy 设置共识策略
func WithConsensusThreshold ¶
func WithConsensusThreshold(threshold float64) ConsensusOption
WithConsensusThreshold 设置阈值
func WithConsensusTimeout ¶
func WithConsensusTimeout(timeout time.Duration) ConsensusOption
WithConsensusTimeout 设置超时
func WithMinParticipation ¶
func WithMinParticipation(rate float64) ConsensusOption
WithMinParticipation 设置最小参与率
type ConsensusProtocol ¶
type ConsensusProtocol struct {
// contains filtered or unexported fields
}
ConsensusProtocol 共识协议
func NewConsensusProtocol ¶
func NewConsensusProtocol(network *AgentNetwork, opts ...ConsensusOption) *ConsensusProtocol
NewConsensusProtocol 创建共识协议
func (*ConsensusProtocol) Propose ¶
func (p *ConsensusProtocol) Propose(ctx context.Context, question string, options []any) (*ConsensusResult, error)
Propose 发起提案
func (*ConsensusProtocol) ProposeToAgents ¶
func (p *ConsensusProtocol) ProposeToAgents(ctx context.Context, question string, options []any, agentIDs []string) (*ConsensusResult, error)
ProposeToAgents 向指定 Agent 发起提案
type ConsensusResult ¶
type ConsensusResult struct {
// ID 结果 ID
ID string `json:"id"`
// Strategy 使用的策略
Strategy ConsensusStrategy `json:"strategy"`
// Decision 最终决策
Decision any `json:"decision"`
// Confidence 置信度(0-1)
Confidence float64 `json:"confidence"`
// Votes 所有投票
Votes []Vote `json:"votes"`
// VoteCount 投票统计
VoteCount map[string]int `json:"vote_count"`
// Participation 参与率
Participation float64 `json:"participation"`
// Reached 是否达成共识
Reached bool `json:"reached"`
// Reason 结果说明
Reason string `json:"reason"`
// Duration 耗时
Duration time.Duration `json:"duration"`
// Timestamp 时间戳
Timestamp time.Time `json:"timestamp"`
}
ConsensusResult 共识结果
func AggregateOutputs ¶
func AggregateOutputs(outputs []Output, strategy ConsensusStrategy) (*ConsensusResult, error)
AggregateOutputs 聚合多个 Agent 输出
type ConsensusStrategy ¶
type ConsensusStrategy int
ConsensusStrategy 共识策略
const ( // ConsensusMajority 多数投票 ConsensusMajority ConsensusStrategy = iota // ConsensusUnanimous 全票通过 ConsensusUnanimous // ConsensusWeighted 加权投票 ConsensusWeighted // ConsensusAverage 平均值(用于数值决策) ConsensusAverage // ConsensusBorda Borda 计数法(用于排序) ConsensusBorda // ConsensusFirst 采用第一个响应 ConsensusFirst // ConsensusBest 采用最佳响应(根据评分) ConsensusBest )
type Constraint ¶
type Constraint struct {
// Type 约束类型
Type ConstraintType `json:"type"`
// Expression 约束表达式
Expression string `json:"expression,omitempty"`
// Value 约束值
Value any `json:"value,omitempty"`
// Description 约束描述
Description string `json:"description,omitempty"`
}
Constraint 约束条件
type ConstraintType ¶
type ConstraintType string
ConstraintType 约束类型
const ( // ConstraintMaxTokens 最大 token 数 ConstraintMaxTokens ConstraintType = "max_tokens" // ConstraintMaxConcurrency 最大并发数 ConstraintMaxConcurrency ConstraintType = "max_concurrency" // ConstraintRateLimit 速率限制 ConstraintRateLimit ConstraintType = "rate_limit" // ConstraintTimeout 超时限制 ConstraintTimeout ConstraintType = "timeout" // ConstraintLanguage 语言限制 ConstraintLanguage ConstraintType = "language" // ConstraintRegion 地区限制 ConstraintRegion ConstraintType = "region" // ConstraintCustom 自定义约束 ConstraintCustom ConstraintType = "custom" )
type ContextVariables ¶
ContextVariables 上下文变量 用于在 Agent 之间传递状态
注意:此类型(普通 map)不是线程安全的。 如果需要并发访问,请使用 SafeContextVariables。
func VariablesFromContext ¶
func VariablesFromContext(ctx context.Context) ContextVariables
VariablesFromContext 从 context 中获取上下文变量
type ConvMessage ¶
type ConvMessage struct {
// Role 角色(user/assistant)
Role string `json:"role"`
// Content 内容
Content string `json:"content"`
// Timestamp 时间戳
Timestamp time.Time `json:"timestamp"`
}
ConvMessage 对话消息记录
注意:与 llm/conversation.TimedMessage 结构相似,但属于不同包。 agent 包不直接依赖 llm/conversation 包,以避免循环依赖。
type ConversationAgent ¶
type ConversationAgent struct {
// contains filtered or unexported fields
}
ConversationAgent 多轮对话 Agent
封装一个 Agent,自动维护对话历史并将上下文注入到每次调用。 线程安全。
func NewConversation ¶
func NewConversation(agent Agent, opts ...ConversationOption) *ConversationAgent
NewConversation 创建多轮对话 Agent
func (*ConversationAgent) Chat ¶
Chat 发送用户消息并获取回复
流程:
- 添加用户消息到历史
- 构建上下文字符串(在 Token 预算内)
- 调用内部 Agent
- 添加助手回复到历史
- 返回结果
func (*ConversationAgent) ClearHistory ¶
func (c *ConversationAgent) ClearHistory()
ClearHistory 清空对话历史
func (*ConversationAgent) History ¶
func (c *ConversationAgent) History() []ConvMessage
History 返回对话历史副本
type ConversationOption ¶
type ConversationOption func(*ConversationAgent)
ConversationOption 对话 Agent 配置选项
func WithConvMaxTokens ¶
func WithConvMaxTokens(n int) ConversationOption
WithConvMaxTokens 设置 Token 预算上限
type CostEstimate ¶
type CostEstimate struct {
// PerRequest 每请求成本
PerRequest float64 `json:"per_request,omitempty"`
// PerToken 每 token 成本
PerToken float64 `json:"per_token,omitempty"`
// PerMinute 每分钟成本
PerMinute float64 `json:"per_minute,omitempty"`
// Currency 货币单位
Currency string `json:"currency,omitempty"`
}
CostEstimate 成本估算
type DeepAgent ¶
type DeepAgent struct {
// contains filtered or unexported fields
}
DeepAgent 深度 Agent
支持递归子任务分解。当主 Agent 判断任务需要拆分时, 通过 "create_subtask" 工具创建子任务,由子 Agent 递归处理。
关键参数:
- agent: 基础 Agent(需要配置 LLM)
- subAgentFn: 子 Agent 工厂函数
- maxDepth: 最大递归深度(防止无限递归)
线程安全:DeepAgent 是不可变的,创建后可安全并发使用。
func NewDeepAgent ¶
func NewDeepAgent(name string, agent Agent, opts ...DeepOption) *DeepAgent
NewDeepAgent 创建深度 Agent
参数:
- name: Agent 名称
- agent: 基础 Agent(需要配置 LLM 用于任务分解决策)
- opts: 可选配置(子 Agent 工厂、最大深度等)
func (*DeepAgent) Batch ¶
func (d *DeepAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*DeepAgent) BatchStream ¶
func (d *DeepAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*DeepAgent) Collect ¶
func (d *DeepAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*DeepAgent) InputSchema ¶
InputSchema 返回输入 Schema
func (*DeepAgent) OutputSchema ¶
OutputSchema 返回输出 Schema
func (*DeepAgent) Run ¶
Run 执行深度 Agent
在指定深度内递归执行。如果 Agent 通过工具调用 create_subtask, 将创建子 Agent 处理子任务并汇总结果。
type DeepOption ¶
type DeepOption func(*DeepAgent)
DeepOption DeepAgent 专用选项
func WithMaxDepth ¶
func WithMaxDepth(n int) DeepOption
WithMaxDepth 设置最大递归深度
超过最大深度时,子任务将直接由当前 Agent 处理而不再递归分解。 默认值: 3
func WithSubAgentFactory ¶
func WithSubAgentFactory(fn func(task string) Agent) DeepOption
WithSubAgentFactory 设置子 Agent 工厂函数
工厂函数接收子任务描述,返回处理该子任务的 Agent。 如果未设置,DeepAgent 将使用主 Agent 处理子任务。
type DefaultCapabilityMatcher ¶
type DefaultCapabilityMatcher struct{}
DefaultCapabilityMatcher 默认能力匹配器
func (*DefaultCapabilityMatcher) Match ¶
func (m *DefaultCapabilityMatcher) Match(req *CapabilityRequirement, cap *CapabilitySpec) bool
Match 检查能力匹配
type DefaultCapabilityScorer ¶
type DefaultCapabilityScorer struct {
// Weights 各维度权重
Weights ScoreWeights
}
DefaultCapabilityScorer 默认评分器
func (*DefaultCapabilityScorer) Score ¶
func (s *DefaultCapabilityScorer) Score(req *CapabilityRequirement, cap *CapabilitySpec) float64
Score 计算分数
type DefaultStateManager ¶
type DefaultStateManager struct {
// contains filtered or unexported fields
}
DefaultStateManager 默认状态管理器实现
func NewStateManager ¶
func NewStateManager(sessionID string, global GlobalState) *DefaultStateManager
NewStateManager 创建默认状态管理器
func (*DefaultStateManager) Agent ¶
func (m *DefaultStateManager) Agent() AgentState
func (*DefaultStateManager) Global ¶
func (m *DefaultStateManager) Global() GlobalState
func (*DefaultStateManager) NewTurn ¶
func (m *DefaultStateManager) NewTurn() TurnState
func (*DefaultStateManager) Restore ¶
func (m *DefaultStateManager) Restore(snapshot StateSnapshot) error
func (*DefaultStateManager) Session ¶
func (m *DefaultStateManager) Session() SessionState
func (*DefaultStateManager) Snapshot ¶
func (m *DefaultStateManager) Snapshot() StateSnapshot
func (*DefaultStateManager) Turn ¶
func (m *DefaultStateManager) Turn() TurnState
type DiscoverQuery ¶
type DiscoverQuery struct {
Tags []string
Capabilities []string
Status AgentStatus
Metadata map[string]any
}
DiscoverQuery 发现查询
type GlobalState ¶
type GlobalState interface {
// Get 获取值
Get(key string) (any, bool)
// Set 设置值
Set(key string, value any)
// Delete 删除值
Delete(key string)
// All 获取所有键值对
All() map[string]any
// RegisterAgent 注册 Agent
RegisterAgent(agentID string, agent Agent)
// GetAgent 获取已注册的 Agent
GetAgent(agentID string) (Agent, bool)
// ListAgents 列出所有已注册的 Agent
ListAgents() []string
}
GlobalState 全局共享状态 生命周期:应用程序 多个 Agent 之间共享
type Handoff ¶
type Handoff struct {
// TargetAgent 目标 Agent
TargetAgent Agent
// Message 交接消息
Message string
// Context 交接上下文
Context map[string]any
// Reason 交接原因
Reason string
}
Handoff 交接结果 当 Agent 需要将任务交接给另一个 Agent 时使用
type HandoffHandler ¶
type HandoffHandler struct {
// OnHandoff 交接回调
OnHandoff func(ctx context.Context, handoff Handoff) error
}
HandoffHandler 交接处理器 用于在外层处理 Agent 交接
func (*HandoffHandler) ProcessToolResult ¶
func (h *HandoffHandler) ProcessToolResult(ctx context.Context, result tool.Result) (*Handoff, error)
ProcessToolResult 处理工具结果,检测是否有交接
type HealthCheckConfig ¶
type HealthCheckConfig struct {
// Interval 检查间隔
Interval time.Duration `json:"interval"`
// Timeout 超时时间
Timeout time.Duration `json:"timeout"`
// HealthyThreshold 健康阈值
HealthyThreshold int `json:"healthy_threshold"`
// UnhealthyThreshold 不健康阈值
UnhealthyThreshold int `json:"unhealthy_threshold"`
// HTTPPath HTTP 检查路径
HTTPPath string `json:"http_path,omitempty"`
}
HealthCheckConfig 健康检查配置
type HealthChecker ¶
type HealthChecker struct {
// contains filtered or unexported fields
}
HealthChecker 健康检查器
func NewHealthChecker ¶
func NewHealthChecker(r *Registry, interval time.Duration) *HealthChecker
NewHealthChecker 创建健康检查器
type Input ¶
type Input struct {
// Query 用户查询
Query string `json:"query"`
// Context 额外上下文
Context map[string]any `json:"context,omitempty"`
}
Input 是 Agent 的输入
type LLMReflector ¶
type LLMReflector struct {
// contains filtered or unexported fields
}
LLMReflector 基于 LLM 的反思器
func NewLLMReflector ¶
func NewLLMReflector(provider llm.Provider) *LLMReflector
NewLLMReflector 创建 LLM 反思器
func (*LLMReflector) Reflect ¶
func (r *LLMReflector) Reflect(ctx context.Context, input Input, output Output) (*Reflection, error)
Reflect 执行反思
func (*LLMReflector) ScoreQuality ¶
func (r *LLMReflector) ScoreQuality(ctx context.Context, input Input, output Output) (float32, error)
ScoreQuality 评估质量分数
type LeastConnectionsBalancer ¶
type LeastConnectionsBalancer struct{}
LeastConnectionsBalancer 最少连接负载均衡器
func (*LeastConnectionsBalancer) Select ¶
func (b *LeastConnectionsBalancer) Select(agents []*AgentInfo) *AgentInfo
Select 选择连接数最少的
type LoadBalancer ¶
LoadBalancer 负载均衡器接口
type LoopAgent ¶
type LoopAgent struct {
// contains filtered or unexported fields
}
LoopAgent 循环执行 Agent 反复执行子 Agent 直到满足终止条件
func NewLoopAgent ¶
func NewLoopAgent(name string, agent Agent, lopts ...LoopOption) *LoopAgent
NewLoopAgent 创建循环执行 Agent
子 Agent 将反复执行,直到 condition 返回 true 或达到 maxLoops。 默认最大循环 10 次。
func (*LoopAgent) Batch ¶
func (a *LoopAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*LoopAgent) BatchStream ¶
func (a *LoopAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*LoopAgent) Collect ¶
func (a *LoopAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*LoopAgent) InputSchema ¶
InputSchema 返回输入 Schema
func (*LoopAgent) OutputSchema ¶
OutputSchema 返回输出 Schema
type LoopOption ¶
type LoopOption func(*LoopAgent)
LoopOption LoopAgent 专用选项
func WithLoopCondition ¶
func WithLoopCondition(condition func(Output, int) bool) LoopOption
WithLoopCondition 设置循环终止条件 condition 函数接收当前输出和循环次数,返回 true 时停止循环
type MemorySetter ¶
MemorySetter 允许外部替换 Agent 的记忆系统
用于共享记忆场景:Team 通过此接口将 Agent 原始记忆包装为 SharedMemoryProxy, 实现跨 Agent 记忆自动共享。BaseAgent 和 ReActAgent 均实现此接口。
type Message ¶
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
ToolCalls []ToolCall `json:"tool_calls,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
Message 消息结构(用于状态存储)
type MessageHandler ¶
type MessageHandler func(ctx context.Context, msg *NetworkMessage) (*NetworkMessage, error)
MessageHandler 消息处理器
type MessageRouter ¶
type MessageRouter struct {
// contains filtered or unexported fields
}
MessageRouter 消息路由器
func NewMessageRouter ¶
func NewMessageRouter(network *AgentNetwork) *MessageRouter
NewMessageRouter 创建消息路由器 队列大小从 network.routerQueueSize 获取,默认 10000。
func (*MessageRouter) Broadcast ¶
func (r *MessageRouter) Broadcast(ctx context.Context, msg *NetworkMessage) error
Broadcast 广播消息
func (*MessageRouter) BroadcastToNeighbors ¶
func (r *MessageRouter) BroadcastToNeighbors(ctx context.Context, msg *NetworkMessage) error
BroadcastToNeighbors 广播给邻居
func (*MessageRouter) Multicast ¶
func (r *MessageRouter) Multicast(ctx context.Context, msg *NetworkMessage, targets []string) error
Multicast 多播消息
func (*MessageRouter) RequestResponse ¶
func (r *MessageRouter) RequestResponse(ctx context.Context, msg *NetworkMessage) (*NetworkMessage, error)
RequestResponse 请求-响应模式
func (*MessageRouter) Route ¶
func (r *MessageRouter) Route(ctx context.Context, msg *NetworkMessage) error
Route 路由消息
如果队列未启动则直接投递,否则放入队列异步处理。 线程安全:会检查队列是否已关闭,避免向已关闭的 channel 发送消息。
func (*MessageRouter) Stop ¶
func (r *MessageRouter) Stop()
Stop 停止路由器
使用 sync.Once 确保队列只关闭一次,避免重复关闭导致 panic。
type MessageType ¶
type MessageType int
MessageType 消息类型
const ( // MessageTypeRequest 请求消息 MessageTypeRequest MessageType = iota // MessageTypeResponse 响应消息 MessageTypeResponse // MessageTypeBroadcast 广播消息 MessageTypeBroadcast // MessageTypeEvent 事件消息 MessageTypeEvent // MessageTypeHeartbeat 心跳消息 MessageTypeHeartbeat )
type MetricsCollector ¶
type MetricsCollector interface {
// RecordDuration 记录执行时长
RecordDuration(duration time.Duration)
// RecordCall 记录调用(成功或失败)
RecordCall(success bool)
// RecordToolCalls 记录工具调用次数
RecordToolCalls(count int)
// RecordTokens 记录 Token 使用量
RecordTokens(count int)
}
MetricsCollector 指标收集器接口
type MiddlewareChain ¶
type MiddlewareChain struct {
// contains filtered or unexported fields
}
MiddlewareChain 中间件链
管理一组中间件,按顺序执行 中间件执行顺序:外层 -> 内层 -> 核心处理 -> 内层 -> 外层
线程安全:所有方法都是并发安全的
func NewMiddlewareChain ¶
func NewMiddlewareChain(middlewares ...AgentMiddleware) *MiddlewareChain
NewMiddlewareChain 创建中间件链
参数:
- middlewares: 要添加的中间件列表
返回:
- 新的中间件链实例
使用示例:
chain := NewMiddlewareChain(
RecoverMiddleware(),
LoggingMiddleware(),
TimeoutMiddleware(30*time.Second),
)
func (*MiddlewareChain) Prepend ¶
func (c *MiddlewareChain) Prepend(middlewares ...AgentMiddleware) *MiddlewareChain
Prepend 在链头部添加中间件
这些中间件会最先执行(最后返回)
线程安全:此方法是并发安全的
func (*MiddlewareChain) Use ¶
func (c *MiddlewareChain) Use(middlewares ...AgentMiddleware) *MiddlewareChain
Use 添加中间件到链
参数:
- middlewares: 要添加的中间件
返回:
- 返回自身,支持链式调用
线程安全:此方法是并发安全的
func (*MiddlewareChain) Wrap ¶
func (c *MiddlewareChain) Wrap(handler AgentHandler) AgentHandler
Wrap 用中间件链包装处理器
参数:
- handler: 核心处理函数
返回:
- 包装后的处理函数
使用示例:
handler := chain.Wrap(func(ctx context.Context, input Input) (Output, error) {
return agent.Run(ctx, input)
})
output, err := handler(ctx, input)
线程安全:此方法是并发安全的
func (*MiddlewareChain) WrapAgent ¶
func (c *MiddlewareChain) WrapAgent(agent Agent) AgentHandler
WrapAgent 用中间件链包装 Agent
返回一个带中间件的 AgentHandler
type NegotiationPreferences ¶
type NegotiationPreferences struct {
// PreferredAgents 偏好的 Agent ID
PreferredAgents []string `json:"preferred_agents,omitempty"`
// ExcludedAgents 排除的 Agent ID
ExcludedAgents []string `json:"excluded_agents,omitempty"`
// MaxCost 最大成本
MaxCost *float64 `json:"max_cost,omitempty"`
// MinAvailability 最低可用性
MinAvailability *float64 `json:"min_availability,omitempty"`
// MaxResponseTime 最大响应时间(毫秒)
MaxResponseTime *int `json:"max_response_time,omitempty"`
}
NegotiationPreferences 协商偏好
type NegotiationRequest ¶
type NegotiationRequest struct {
// Requirements 能力需求列表
Requirements []*CapabilityRequirement `json:"requirements"`
// Preferences 偏好设置
Preferences *NegotiationPreferences `json:"preferences,omitempty"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
}
NegotiationRequest 协商请求
type NegotiationResult ¶
type NegotiationResult struct {
// Success 是否成功
Success bool `json:"success"`
// Assignments 能力分配
Assignments []*CapabilityAssignment `json:"assignments"`
// UnmetRequirements 未满足的需求
UnmetRequirements []*CapabilityRequirement `json:"unmet_requirements,omitempty"`
// Score 总分
Score float64 `json:"score"`
// Error 错误信息
Error string `json:"error,omitempty"`
}
NegotiationResult 协商结果
type Negotiator ¶
type Negotiator struct {
// contains filtered or unexported fields
}
Negotiator 能力协商器
func NewNegotiator ¶
func NewNegotiator(registry *Registry, opts ...NegotiatorOption) *Negotiator
NewNegotiator 创建协商器
func (*Negotiator) DeclareCapability ¶
func (n *Negotiator) DeclareCapability(agentID string, capability CapabilitySpec) error
DeclareCapability 声明能力
func (*Negotiator) ExportCapabilities ¶
func (n *Negotiator) ExportCapabilities(agentID string) ([]byte, error)
ExportCapabilities 导出能力描述(JSON 格式)
func (*Negotiator) ImportCapabilities ¶
func (n *Negotiator) ImportCapabilities(agentID string, data []byte) error
ImportCapabilities 导入能力描述
func (*Negotiator) Negotiate ¶
func (n *Negotiator) Negotiate(ctx context.Context, req *NegotiationRequest) (*NegotiationResult, error)
Negotiate 执行能力协商
func (*Negotiator) QueryCapabilities ¶
func (n *Negotiator) QueryCapabilities(agentID string) ([]CapabilitySpec, error)
QueryCapabilities 查询能力
type NegotiatorOption ¶
type NegotiatorOption func(*Negotiator)
NegotiatorOption 协商器选项
func WithCapabilityScorer ¶
func WithCapabilityScorer(scorer CapabilityScorer) NegotiatorOption
WithCapabilityScorer 设置评分器
type NetworkMessage ¶
type NetworkMessage struct {
// ID 消息 ID
ID string `json:"id"`
// Type 消息类型
Type MessageType `json:"type"`
// From 发送者 Agent ID
From string `json:"from"`
// To 接收者 Agent ID(空表示广播)
To string `json:"to,omitempty"`
// Topic 消息主题
Topic string `json:"topic,omitempty"`
// Content 消息内容
Content any `json:"content"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
// Timestamp 时间戳
Timestamp time.Time `json:"timestamp"`
// ReplyTo 回复的消息 ID
ReplyTo string `json:"reply_to,omitempty"`
// TTL 消息生存时间
TTL time.Duration `json:"ttl,omitempty"`
// Priority 优先级(0-9,越大越高)
Priority int `json:"priority,omitempty"`
}
NetworkMessage 网络消息
func NewMessage ¶
func NewMessage(from, to string, msgType MessageType, content any) *NetworkMessage
NewMessage 创建消息
type NetworkNode ¶
type NetworkNode struct {
// Agent 关联的 Agent
Agent Agent
// Neighbors 相邻节点 ID
Neighbors []string
// Inbox 收件箱
Inbox chan *NetworkMessage
// Status 节点状态
Status NodeStatus
// LastHeartbeat 最后心跳时间
LastHeartbeat time.Time
// Metadata 节点元数据
Metadata map[string]any
// contains filtered or unexported fields
}
NetworkNode 网络节点
func (*NetworkNode) CloseInbox ¶
func (n *NetworkNode) CloseInbox()
CloseInbox 安全关闭收件箱
使用 sync.Once 确保只关闭一次,避免 panic。 此方法是并发安全的。
type NetworkOption ¶
type NetworkOption func(*AgentNetwork)
NetworkOption 网络配置选项
func WithHeartbeatInterval ¶
func WithHeartbeatInterval(interval time.Duration) NetworkOption
WithHeartbeatInterval 设置心跳间隔
func WithNetworkInboxSize ¶
func WithNetworkInboxSize(size int) NetworkOption
WithNetworkInboxSize 设置收件箱大小
func WithNetworkTopology ¶
func WithNetworkTopology(topology NetworkTopology) NetworkOption
WithNetworkTopology 设置拓扑
func WithRouterQueueSize ¶
func WithRouterQueueSize(size int) NetworkOption
WithRouterQueueSize 设置路由器消息队列大小 默认 10000,高负载场景可适当增大。
type NetworkStats ¶
type NetworkStats struct {
TotalNodes int `json:"total_nodes"`
OnlineNodes int `json:"online_nodes"`
OfflineNodes int `json:"offline_nodes"`
BusyNodes int `json:"busy_nodes"`
TotalEdges int `json:"total_edges"`
Topology string `json:"topology"`
MessagesSent int64 `json:"messages_sent"`
MessagesRecv int64 `json:"messages_recv"`
MessagesFailed int64 `json:"messages_failed"`
}
NetworkStats 网络统计
type NetworkTopology ¶
type NetworkTopology int
NetworkTopology 网络拓扑类型
const ( // TopologyMesh 全连接网格 TopologyMesh NetworkTopology = iota // TopologyHub 中心辐射 TopologyHub // TopologyRing 环形 TopologyRing // TopologyTree 树形 TopologyTree // TopologyCustom 自定义 TopologyCustom )
type NodeStatus ¶
type NodeStatus int
NodeStatus 节点状态
const ( // NodeStatusOnline 在线 NodeStatusOnline NodeStatus = iota // NodeStatusOffline 离线 NodeStatusOffline // NodeStatusBusy 忙碌 NodeStatusBusy // NodeStatusError 错误 NodeStatusError )
type Option ¶
type Option func(*Config)
Option 是 Agent 配置选项
func WithCheckpointer ¶
func WithCheckpointer(cp checkpoint.Checkpointer) Option
WithCheckpointer 开启 Agent 的可持久化/可恢复执行,以给定 Checkpointer 作为存储后端。
设置后,ReActAgent.Run 会在每步边界持久化执行快照(命名空间为本次 run 的 run_id, 经 Output.Metadata["run_id"] 返回),并可用该 run_id 调 ReActAgent.Resume 续跑或 取回已完成结果。cp 为 nil 时本选项 no-op。
仅 ReActAgent(单次多轮 run)有干净的整次可恢复语义;多 run agent 不适用。
func WithMiddleware ¶
func WithMiddleware(mws ...agentruntime.Middleware) Option
WithMiddleware 追加注入到底层 runtime 的中间件(runtime 中间件扩展点在 Agent 层的出口)。
预算控制首选 middleware.NewBudgetControl:用 cost.Controller.BudgetCostFunc() 提供单 run 成本估算,用 RecordUsageFunc() 提供跨 run 累计记账,并在 provider/调用方的每次 LLM 外呼前调用 CheckRequest() 完成 token/频率预检。 NewBudgetControl 底层组合 per-run Budget 与 cross-run CostControl;三者共享同一 Controller 时,多 run agent 无需额外累加器。Agent 仍不引入对 security/cost 的硬依赖。
func WithStrategy ¶
func WithStrategy(s agentruntime.Strategy) Option
WithStrategy 选择统一 agent loop 的执行策略。
传入 runtime/strategy 包提供的策略(strategy.ReAct{} / strategy.PlanExecute{} / strategy.Reflection{}),即可让 ReActAgent 以对应策略(提示词引导式)在同一个 统一回合循环上运行。nil 时默认 ReAct 行为。
type Output ¶
type Output struct {
// Content 最终回复内容
Content string `json:"content"`
// ToolCalls 执行的工具调用记录
ToolCalls []ToolCallRecord `json:"tool_calls,omitempty"`
// Blocks 本次运行的有序内容块流(text↔tool 交错序),与 runtime.Result.Blocks 对齐。
// SDK 消费者(如 hexeye)据此保真展示多步 ReAct,而非把 Content 压平。
Blocks template.Blocks `json:"blocks,omitempty"`
// Usage Token 使用统计
Usage llm.Usage `json:"usage,omitempty"`
// StopReason 运行终止原因(end_turn / max_turns),与运行时 Result.StopReason 对齐,
// 让调用方据此决定如何呈现(如轮次耗尽时提示「可继续」),无需 errors.Is 反查。
StopReason agentruntime.StopReason `json:"stop_reason,omitempty"`
// Metadata 额外元数据
Metadata map[string]any `json:"metadata,omitempty"`
}
Output 是 Agent 的输出
type ParallelAgent ¶
type ParallelAgent struct {
// contains filtered or unexported fields
}
ParallelAgent 并行执行 Agent 同时执行多个子 Agent,合并所有结果
func NewParallelAgent ¶
func NewParallelAgent(name string, agents []Agent, popts ...ParallelOption) *ParallelAgent
NewParallelAgent 创建并行执行 Agent
所有子 Agent 将同时执行,结果通过 mergeFunc 合并。 默认合并策略:拼接所有输出内容。
func (*ParallelAgent) Batch ¶
func (a *ParallelAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*ParallelAgent) BatchStream ¶
func (a *ParallelAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*ParallelAgent) Collect ¶
func (a *ParallelAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*ParallelAgent) InputSchema ¶
InputSchema 返回输入 Schema
func (*ParallelAgent) Invoke ¶
func (a *ParallelAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
Invoke 执行 Agent
func (*ParallelAgent) OutputSchema ¶
OutputSchema 返回输出 Schema
func (*ParallelAgent) Stream ¶
func (a *ParallelAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 Agent
func (*ParallelAgent) Tools ¶
func (a *ParallelAgent) Tools() []tool.Tool
Tools 返回工具列表(聚合所有子 Agent 的工具,按名称去重)
func (*ParallelAgent) Transform ¶
func (a *ParallelAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type ParallelOption ¶
type ParallelOption func(*ParallelAgent)
ParallelOption ParallelAgent 专用选项
func WithMergeFunc ¶
func WithMergeFunc(fn func([]Output) Output) ParallelOption
WithMergeFunc 设置结果合并函数
type ParameterSpec ¶
type ParameterSpec struct {
// Name 参数名称
Name string `json:"name"`
// Type 参数类型
Type string `json:"type"`
// Required 是否必需
Required bool `json:"required,omitempty"`
// Description 参数描述
Description string `json:"description,omitempty"`
// Default 默认值
Default any `json:"default,omitempty"`
// Enum 枚举值
Enum []any `json:"enum,omitempty"`
// Min 最小值
Min *float64 `json:"min,omitempty"`
// Max 最大值
Max *float64 `json:"max,omitempty"`
}
ParameterSpec 参数规格
type PlanExecuteAgent ¶
type PlanExecuteAgent struct {
*BaseAgent
// contains filtered or unexported fields
}
PlanExecuteAgent 计划执行分离的 Agent
Plan-and-Execute 模式将任务规划和执行分离,让 Agent 能够:
- 先生成完整的执行计划
- 按步骤执行计划
- 根据执行结果动态调整计划
- 在失败时进行重计划
与 ReAct Agent 的区别:
- ReAct: 边思考边行动,适合简单任务
- Plan-Execute: 先规划后执行,适合复杂多步任务
使用示例:
agent := NewPlanExecute(
WithLLM(llmProvider),
WithTools(searchTool, calculatorTool),
WithPlanExecutePlanner(planner.NewSequentialPlanner(...)),
)
output, err := agent.Run(ctx, Input{Query: "完成复杂的多步任务"})
func NewPlanExecute ¶
func NewPlanExecute(opts []Option, peOpts ...PlanExecuteOption) *PlanExecuteAgent
NewPlanExecute 创建 Plan-Execute Agent
参数:
- opts: Agent 基础配置选项
- peOpts: Plan-Execute 特有配置选项
使用示例:
agent := NewPlanExecute(
WithLLM(llm),
WithTools(tools...),
WithPlanExecutePlanner(planner),
)
func (*PlanExecuteAgent) Batch ¶
func (a *PlanExecuteAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*PlanExecuteAgent) BatchStream ¶
func (a *PlanExecuteAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*PlanExecuteAgent) Collect ¶
func (a *PlanExecuteAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*PlanExecuteAgent) Invoke ¶
func (a *PlanExecuteAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
Invoke 执行 PlanExecute Agent(实现 Runnable 接口)
func (*PlanExecuteAgent) Run ¶
Run 执行 Plan-Execute Agent
执行流程:
- 使用 Planner 生成执行计划
- 按顺序执行每个步骤
- 收集执行结果
- 步骤失败时进行重计划(如果启用)
- 汇总所有结果生成最终回复
func (*PlanExecuteAgent) Stream ¶
func (a *PlanExecuteAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 Agent
func (*PlanExecuteAgent) Transform ¶
func (a *PlanExecuteAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type PlanExecuteOption ¶
type PlanExecuteOption func(*PlanExecuteAgent)
PlanExecuteOption Plan-Execute Agent 配置选项
func WithPlanExecuteMaxReplans ¶
func WithPlanExecuteMaxReplans(n int) PlanExecuteOption
WithPlanExecuteMaxReplans 设置最大重计划次数 默认值: 3
func WithPlanExecutePlanner ¶
func WithPlanExecutePlanner(p planner.Planner) PlanExecuteOption
WithPlanExecutePlanner 设置规划器
func WithPlanExecuteReplanOnFailure ¶
func WithPlanExecuteReplanOnFailure(enabled bool) PlanExecuteOption
WithPlanExecuteReplanOnFailure 设置步骤失败时是否自动重计划 默认值: true
func WithPlanExecuteSummarizer ¶
func WithPlanExecuteSummarizer(s Summarizer) PlanExecuteOption
WithPlanExecuteSummarizer 设置结果汇总器
type Poll ¶
type Poll struct {
// ID 投票 ID
ID string
// Question 问题
Question string
// Options 选项(可选)
Options []any
// Voters 投票者列表
Voters []string
// Votes 收到的投票
Votes []Vote
// StartedAt 开始时间
StartedAt time.Time
// Closed 是否关闭
Closed bool
// contains filtered or unexported fields
}
Poll 投票会话
type ProxyConfig ¶
ProxyConfig 代理配置
type ProxyOption ¶
type ProxyOption func(*ProxyConfig)
ProxyOption 代理配置选项
func WithReadFromShared ¶
func WithReadFromShared(enabled bool) ProxyOption
WithReadFromShared 设置是否从共享记忆读取
func WithSharedSearchLimit ¶
func WithSharedSearchLimit(limit int) ProxyOption
WithSharedSearchLimit 设置共享记忆搜索结果限制
func WithWriteToShared ¶
func WithWriteToShared(enabled bool) ProxyOption
WithWriteToShared 设置是否同步写入到共享记忆
type RateLimiter ¶
RateLimiter 限流器接口
type ReActAgent ¶
type ReActAgent struct {
*BaseAgent
}
ReActAgent 实现 ReAct (Reasoning + Acting) 模式的 Agent ReAct 模式让 Agent 交替进行推理和行动,直到完成任务
func (*ReActAgent) Batch ¶
func (a *ReActAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 ReAct Agent
func (*ReActAgent) BatchStream ¶
func (a *ReActAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*ReActAgent) Collect ¶
func (a *ReActAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*ReActAgent) Resume ¶
Resume 从某次执行最近的持久化快照续跑(需经 WithCheckpointer 开启持久化)。
runID 取自先前 Run 返回的 Output.Metadata["run_id"]。若该 run 已完成,Resume 直接 返回其最终结果(不重跑);若中断在中间步,则从该步之后继续。
func (*ReActAgent) Run ¶
Run 执行 ReAct Agent。
若经 WithCheckpointer 开启了持久化,本次执行会在每步边界持久化快照;返回的 Output.Metadata["run_id"] 即本次 run 的标识符,可用于后续 Resume。
func (*ReActAgent) Stream ¶
func (a *ReActAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 ReAct Agent
func (*ReActAgent) Transform ¶
func (a *ReActAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type ReasoningModule ¶
type ReasoningModule struct {
// Name 模块名称
Name string `json:"name"`
// Description 模块描述
Description string `json:"description"`
// Template 推理模板(用于 ADAPT 阶段)
Template string `json:"template"`
// Examples 示例(可选)
Examples []string `json:"examples,omitempty"`
}
ReasoningModule 推理模块 每个模块代表一种推理策略
type Reflection ¶
type Reflection struct {
// Quality 质量评分 (0.0 - 1.0)
Quality float32 `json:"quality"`
// Strengths 优点列表
Strengths []string `json:"strengths,omitempty"`
// Weaknesses 缺点列表
Weaknesses []string `json:"weaknesses,omitempty"`
// Suggestions 改进建议列表
Suggestions []string `json:"suggestions,omitempty"`
// ShouldRetry 是否需要重试
ShouldRetry bool `json:"should_retry"`
// Feedback 给下次执行的反馈(用于改进)
Feedback string `json:"feedback,omitempty"`
}
Reflection 反思结果
type ReflectionAgent ¶
type ReflectionAgent struct {
*BaseAgent
// contains filtered or unexported fields
}
ReflectionAgent 自我反思 Agent
Reflection Agent 在执行任务后进行自我反思,评估输出质量, 并在质量不满足要求时自动重试。
反思流程:
- 执行任务生成初始输出
- 使用 Reflector 评估输出质量
- 识别优缺点和改进建议
- 如果质量未达标,根据反馈重新执行
- 重复直到达标或达到最大迭代次数
使用示例:
agent := NewReflection(
WithLLM(llmProvider),
WithReflectionMaxIterations(5),
WithReflectionQualityTarget(0.8),
)
output, err := agent.Run(ctx, Input{Query: "撰写一篇高质量文章"})
func NewReflection ¶
func NewReflection(opts []Option, rOpts ...ReflectionOption) *ReflectionAgent
NewReflection 创建 Reflection Agent
参数:
- opts: Agent 基础配置选项
- rOpts: Reflection 特有配置选项
使用示例:
agent := NewReflection(
[]Option{WithLLM(llm)},
WithReflectionQualityTarget(0.85),
)
func (*ReflectionAgent) Batch ¶
func (a *ReflectionAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*ReflectionAgent) BatchStream ¶
func (a *ReflectionAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*ReflectionAgent) Collect ¶
func (a *ReflectionAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*ReflectionAgent) Invoke ¶
func (a *ReflectionAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
Invoke 执行 Reflection Agent(实现 Runnable 接口)
func (*ReflectionAgent) Run ¶
Run 执行 Reflection Agent
执行流程:
- 执行任务生成初始输出
- 反思评估输出质量
- 如果质量未达标且未达到最大迭代次数,重试
- 返回最终输出
func (*ReflectionAgent) Stream ¶
func (a *ReflectionAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 Agent
func (*ReflectionAgent) Transform ¶
func (a *ReflectionAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type ReflectionOption ¶
type ReflectionOption func(*ReflectionAgent)
ReflectionOption Reflection Agent 配置选项
func WithReflectionMaxIterations ¶
func WithReflectionMaxIterations(n int) ReflectionOption
WithReflectionMaxIterations 设置最大反思迭代次数 默认值: 3
func WithReflectionMinIterations ¶
func WithReflectionMinIterations(n int) ReflectionOption
WithReflectionMinIterations 设置最小迭代次数 默认值: 1
func WithReflectionQualityTarget ¶
func WithReflectionQualityTarget(target float32) ReflectionOption
WithReflectionQualityTarget 设置目标质量分数 默认值: 0.8
type Reflector ¶
type Reflector interface {
// Reflect 对输出进行反思
// 返回反思结果,包括质量评分、优缺点和改进建议
Reflect(ctx context.Context, input Input, output Output) (*Reflection, error)
// ScoreQuality 仅评估输出质量分数
// 返回 0.0-1.0 之间的分数
ScoreQuality(ctx context.Context, input Input, output Output) (float32, error)
}
Reflector 反思器接口 负责评估输出质量并提供改进建议
type RegisterOption ¶
type RegisterOption func(*AgentInfo)
RegisterOption 注册选项
func WithCapabilities ¶
func WithCapabilities(caps ...Capability) RegisterOption
WithCapabilities 设置能力
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry Agent 注册表
功能:
- 注册和注销 Agent
- 按标签/能力查询 Agent
- 健康检查和心跳
- 负载均衡选择
使用示例:
registry := NewRegistry()
// 注册 Agent
registry.Register(&AgentInfo{
ID: "agent-1",
Name: "assistant",
Tags: []string{"chat", "qa"},
})
// 发现 Agent
agents := registry.Discover(WithTag("chat"))
func (*Registry) Discover ¶
func (r *Registry) Discover(opts ...DiscoverOption) []*AgentInfo
Discover 发现 Agent
func (*Registry) DiscoverOne ¶
func (r *Registry) DiscoverOne(opts ...DiscoverOption) (*AgentInfo, error)
DiscoverOne 发现单个 Agent(负载均衡)
func (*Registry) RegisterAgent ¶
func (r *Registry) RegisterAgent(agent Agent, opts ...RegisterOption) error
RegisterAgent 注册 Agent 实例
func (*Registry) SetLoadBalancer ¶
func (r *Registry) SetLoadBalancer(lb LoadBalancer)
SetLoadBalancer 设置负载均衡器
type RegistryConfig ¶
type RegistryConfig struct {
// HealthCheckInterval 健康检查间隔
HealthCheckInterval time.Duration
// HeartbeatTimeout 心跳超时时间
HeartbeatTimeout time.Duration
// DeregisterAfter 超时后自动注销时间
DeregisterAfter time.Duration
// EnableHealthCheck 启用健康检查
EnableHealthCheck bool
}
RegistryConfig 注册表配置
type ReplaySafety ¶
type ReplaySafety = agentruntime.ToolSideEffect
ReplaySafety 是工具声明的重放安全性等级类型别名(= runtime.ToolSideEffect)。
type ReplaySafetyAware ¶
type ReplaySafetyAware interface {
ReplaySafety() ReplaySafety
}
ReplaySafetyAware 是工具**可选**实现的接口,用于声明自身在 Durable 崩溃重放下的 安全性。未实现的工具一律按最保守的 ReplayUnsafe 处理(Durable 续跑时对其 fail-closed)。
示例:一个只读的检索工具可声明可安全重放,从而崩溃后自动续跑而非 fail-closed:
func (t *SearchTool) ReplaySafety() agent.ReplaySafety { return agent.ReplayReadOnly }
type Role ¶
type Role struct {
// Name 角色名称
Name string `yaml:"name" json:"name"`
// Title 角色头衔 (e.g., "Senior Researcher", "Lead Developer")
Title string `yaml:"title" json:"title"`
// Goal 角色目标
Goal string `yaml:"goal" json:"goal"`
// Backstory 背景故事,帮助 LLM 更好地扮演角色
Backstory string `yaml:"backstory" json:"backstory"`
// Expertise 专长领域
Expertise []string `yaml:"expertise" json:"expertise"`
// Tools 可用工具名称列表
Tools []string `yaml:"tools" json:"tools"`
// Personality 性格特点
Personality string `yaml:"personality" json:"personality"`
// Constraints 行为约束
Constraints []string `yaml:"constraints" json:"constraints"`
// AllowDelegation 是否允许委托任务给其他 Agent
AllowDelegation bool `yaml:"allow_delegation" json:"allow_delegation"`
// DelegateTo 可以委托给的 Agent 名称列表
DelegateTo []string `yaml:"delegate_to" json:"delegate_to"`
}
Role 角色定义 角色系统:定义 Agent 的 Name/Goal/Backstory。
type RoleBuilder ¶
type RoleBuilder struct {
// contains filtered or unexported fields
}
RoleBuilder 角色构建器
func (*RoleBuilder) AllowDelegation ¶
func (b *RoleBuilder) AllowDelegation(allow bool) *RoleBuilder
AllowDelegation 设置是否允许委托
func (*RoleBuilder) Backstory ¶
func (b *RoleBuilder) Backstory(backstory string) *RoleBuilder
Backstory 设置背景故事
func (*RoleBuilder) Constraints ¶
func (b *RoleBuilder) Constraints(constraints ...string) *RoleBuilder
Constraints 设置行为约束
func (*RoleBuilder) DelegateTo ¶
func (b *RoleBuilder) DelegateTo(agents ...string) *RoleBuilder
DelegateTo 设置可委托的 Agent
func (*RoleBuilder) Expertise ¶
func (b *RoleBuilder) Expertise(areas ...string) *RoleBuilder
Expertise 设置专长领域
func (*RoleBuilder) Personality ¶
func (b *RoleBuilder) Personality(personality string) *RoleBuilder
Personality 设置性格特点
type RoundRobinBalancer ¶
type RoundRobinBalancer struct {
// contains filtered or unexported fields
}
RoundRobinBalancer 轮询负载均衡器
func (*RoundRobinBalancer) Select ¶
func (b *RoundRobinBalancer) Select(agents []*AgentInfo) *AgentInfo
Select 轮询选择
type SLASpec ¶
type SLASpec struct {
// Availability 可用性(百分比)
Availability float64 `json:"availability,omitempty"`
// ResponseTime 响应时间(毫秒)
ResponseTime int `json:"response_time,omitempty"`
// Throughput 吞吐量(请求/秒)
Throughput int `json:"throughput,omitempty"`
// ErrorRate 错误率(百分比)
ErrorRate float64 `json:"error_rate,omitempty"`
}
SLASpec 服务级别协议
type SafeContextVariables ¶
type SafeContextVariables struct {
// contains filtered or unexported fields
}
SafeContextVariables 线程安全的上下文变量 用于在多个 goroutine 之间安全地传递和修改状态
func NewSafeContextVariables ¶
func NewSafeContextVariables() *SafeContextVariables
NewSafeContextVariables 创建线程安全的上下文变量
func SafeVariablesFromContext ¶
func SafeVariablesFromContext(ctx context.Context) *SafeContextVariables
SafeVariablesFromContext 从 context 中获取线程安全的上下文变量
func (*SafeContextVariables) Clone ¶
func (s *SafeContextVariables) Clone() ContextVariables
Clone 克隆变量(线程安全,返回普通 ContextVariables)
func (*SafeContextVariables) CloneSafe ¶
func (s *SafeContextVariables) CloneSafe() *SafeContextVariables
CloneSafe 克隆为新的 SafeContextVariables(线程安全)
func (*SafeContextVariables) Delete ¶
func (s *SafeContextVariables) Delete(key string)
Delete 删除值(线程安全)
func (*SafeContextVariables) Get ¶
func (s *SafeContextVariables) Get(key string) (any, bool)
Get 获取值(线程安全)
func (*SafeContextVariables) Merge ¶
func (s *SafeContextVariables) Merge(other ContextVariables)
Merge 合并变量(线程安全)
func (*SafeContextVariables) MergeSafe ¶
func (s *SafeContextVariables) MergeSafe(other *SafeContextVariables)
MergeSafe 合并另一个 SafeContextVariables(线程安全)
func (*SafeContextVariables) Set ¶
func (s *SafeContextVariables) Set(key string, value any)
Set 设置值(线程安全)
type ScoreWeights ¶
type ScoreWeights struct {
// VersionMatch 版本匹配权重
VersionMatch float64
// ConstraintMatch 约束匹配权重
ConstraintMatch float64
// CostWeight 成本权重
CostWeight float64
// SLAWeight SLA 权重
SLAWeight float64
}
ScoreWeights 评分权重
type SelfDiscoveryAgent ¶
type SelfDiscoveryAgent struct {
*BaseAgent
// contains filtered or unexported fields
}
SelfDiscoveryAgent 自我发现 Agent
Self-Discovery Agent 参考 Google 的 SELF-DISCOVER 论文实现, 通过以下四个阶段解决复杂推理任务:
- SELECT(选择): 从推理模块库中选择与任务相关的推理模块
- ADAPT(适配): 将选中的模块适配到具体任务
- IMPLEMENT(实现): 生成任务特定的推理结构
- EXECUTE(执行): 使用生成的结构执行推理
内置推理模块包括:
- 批判性思维: 评估论点和证据
- 逐步推理: 分解复杂问题
- 创造性思维: 生成新颖解决方案
- 系统分析: 理解组件间关系
- 类比推理: 利用相似性解决问题
- 归纳推理: 从具体到一般
- 演绎推理: 从一般到具体
使用示例:
agent := NewSelfDiscovery(
WithLLM(llmProvider),
WithSelfDiscoveryModules(
CriticalThinkingModule,
StepByStepModule,
),
)
output, err := agent.Run(ctx, Input{Query: "解决复杂推理问题"})
func NewSelfDiscovery ¶
func NewSelfDiscovery(opts []Option, sdOpts ...SelfDiscoveryOption) *SelfDiscoveryAgent
NewSelfDiscovery 创建 Self-Discovery Agent
参数:
- opts: Agent 基础配置选项
- sdOpts: Self-Discovery 特有配置选项
使用示例:
agent := NewSelfDiscovery(
[]Option{WithLLM(llm)},
WithSelfDiscoveryMaxModules(4),
)
func (*SelfDiscoveryAgent) Batch ¶
func (a *SelfDiscoveryAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*SelfDiscoveryAgent) BatchStream ¶
func (a *SelfDiscoveryAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*SelfDiscoveryAgent) Collect ¶
func (a *SelfDiscoveryAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*SelfDiscoveryAgent) Invoke ¶
func (a *SelfDiscoveryAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
Invoke 执行 SelfDiscovery Agent(实现 Runnable 接口)
func (*SelfDiscoveryAgent) Run ¶
Run 执行 Self-Discovery Agent
执行流程:
- SELECT: 选择相关的推理模块
- ADAPT: 将模块适配到当前任务
- IMPLEMENT: 生成任务特定的推理结构
- EXECUTE: 使用结构执行推理
func (*SelfDiscoveryAgent) Stream ¶
func (a *SelfDiscoveryAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 Agent
func (*SelfDiscoveryAgent) Transform ¶
func (a *SelfDiscoveryAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type SelfDiscoveryOption ¶
type SelfDiscoveryOption func(*SelfDiscoveryAgent)
SelfDiscoveryOption Self-Discovery Agent 配置选项
func WithSelfDiscoveryMaxModules ¶
func WithSelfDiscoveryMaxModules(n int) SelfDiscoveryOption
WithSelfDiscoveryMaxModules 设置最多选择的模块数 默认值: 3
func WithSelfDiscoveryModules ¶
func WithSelfDiscoveryModules(modules ...ReasoningModule) SelfDiscoveryOption
WithSelfDiscoveryModules 设置可用的推理模块
type SequentialAgent ¶
type SequentialAgent struct {
// contains filtered or unexported fields
}
SequentialAgent 顺序执行 Agent 按顺序依次执行多个子 Agent,前一个 Agent 的输出作为后一个的上下文
func NewSequentialAgent ¶
func NewSequentialAgent(name string, agents []Agent, opts ...Option) *SequentialAgent
NewSequentialAgent 创建顺序执行 Agent
子 Agent 将按添加顺序依次执行,每个 Agent 的输出 会作为下一个 Agent 输入的 Context 传递。
func (*SequentialAgent) Batch ¶
func (a *SequentialAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*SequentialAgent) BatchStream ¶
func (a *SequentialAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*SequentialAgent) Collect ¶
func (a *SequentialAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*SequentialAgent) Description ¶
func (a *SequentialAgent) Description() string
Description 返回描述
func (*SequentialAgent) InputSchema ¶
InputSchema 返回输入 Schema
func (*SequentialAgent) Invoke ¶
func (a *SequentialAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
Invoke 执行 Agent
func (*SequentialAgent) OutputSchema ¶
OutputSchema 返回输出 Schema
func (*SequentialAgent) Stream ¶
func (a *SequentialAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 Agent
func (*SequentialAgent) Tools ¶
func (a *SequentialAgent) Tools() []tool.Tool
Tools 返回工具列表(聚合所有子 Agent 的工具,按名称去重)
func (*SequentialAgent) Transform ¶
func (a *SequentialAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type SessionState ¶
type SessionState interface {
// ID 获取会话 ID
ID() string
// Get 获取值
Get(key string) (any, bool)
// Set 设置值
Set(key string, value any)
// Delete 删除值
Delete(key string)
// All 获取所有键值对
All() map[string]any
// CreatedAt 获取创建时间
CreatedAt() time.Time
// UpdatedAt 获取最后更新时间
UpdatedAt() time.Time
// TurnCount 获取轮次计数
TurnCount() int
// IncrementTurnCount 增加轮次计数
IncrementTurnCount()
}
SessionState 会话级状态 生命周期:一次会话/对话
type SharedMemory ¶
type SharedMemory struct {
// contains filtered or unexported fields
}
SharedMemory 团队级共享记忆容器
提供三层共享记忆:
- shortTerm: 短期记忆(BufferMemory),存储最近的任务结果,始终启用
- longTerm: 长期记忆(VectorMemory),支持语义检索,可选
- entity: 实体记忆(EntityMemory),构建实体知识库,可选
所有 Agent 通过 SharedMemoryProxy 读写共享记忆,实现跨 Agent 的记忆自动共享。
线程安全:所有方法都是并发安全的(底层 memory.Memory 实现自身保证线程安全)
func NewSharedMemory ¶
func NewSharedMemory(opts ...SharedMemoryOption) *SharedMemory
NewSharedMemory 创建共享记忆
默认只启用短期记忆(BufferMemory,容量 200)。 通过 Option 可启用长期记忆和实体记忆。
示例:
sm := NewSharedMemory() // 仅短期记忆 sm := NewSharedMemory(WithShortTermCapacity(500)) // 自定义容量 sm := NewSharedMemory(WithLongTermMemory(vecMem)) // 启用长期记忆
func (*SharedMemory) Clear ¶
func (sm *SharedMemory) Clear(ctx context.Context) error
Clear 清空所有共享记忆
func (*SharedMemory) Delete ¶
func (sm *SharedMemory) Delete(ctx context.Context, id string) error
Delete 删除条目(从 shortTerm 删除)
func (*SharedMemory) Save ¶
Save 保存条目到共享记忆
写入 shortTerm;如启用则同时写入 longTerm 和 entity。 longTerm/entity 写入失败仅输出 stderr 警告,不影响 shortTerm。
func (*SharedMemory) Search ¶
func (sm *SharedMemory) Search(ctx context.Context, query memory.SearchQuery) ([]memory.Entry, error)
Search 搜索共享记忆
合并 shortTerm 和 longTerm 的搜索结果(去重)。 longTerm 搜索失败仅输出 stderr 警告。
type SharedMemoryConfig ¶
type SharedMemoryConfig struct {
ShortTermCapacity int
}
SharedMemoryConfig 共享记忆配置
type SharedMemoryOption ¶
type SharedMemoryOption func(*SharedMemory)
SharedMemoryOption 共享记忆配置选项
func WithEntityMemory ¶
func WithEntityMemory(entity *memory.EntityMemory) SharedMemoryOption
WithEntityMemory 启用实体记忆(实体知识库)
传入已配置好 EntityExtractor 的 EntityMemory 实例
func WithLongTermMemory ¶
func WithLongTermMemory(longTerm memory.Memory) SharedMemoryOption
WithLongTermMemory 启用长期记忆(语义检索)
传入已配置好 Embedder 的 VectorMemory 实例
func WithShortTermCapacity ¶
func WithShortTermCapacity(capacity int) SharedMemoryOption
WithShortTermCapacity 设置短期记忆容量
type SharedMemoryProxy ¶
type SharedMemoryProxy struct {
// contains filtered or unexported fields
}
SharedMemoryProxy 共享记忆代理
每个 Agent 持有一个 Proxy 实例,拦截 Memory 的读写操作:
- Save: 写入本地记忆 + 同步到共享记忆(标记 _agent_id/_agent_name)
- Search: 合并本地记忆和共享记忆的搜索结果
- Get/Delete/Clear/Stats: 代理到本地记忆
对 Agent 完全透明,无需修改 Agent 代码。
线程安全:本身无状态,底层 local 和 shared 自身保证线程安全
func NewSharedMemoryProxy ¶
func NewSharedMemoryProxy(local memory.Memory, shared *SharedMemory, agentID, agentName string, opts ...ProxyOption) *SharedMemoryProxy
NewSharedMemoryProxy 创建共享记忆代理
参数:
- local: Agent 原始记忆
- shared: 团队共享记忆
- agentID: 所属 Agent 的 ID
- agentName: 所属 Agent 的名称
- opts: 代理配置选项
func (*SharedMemoryProxy) Clear ¶
func (p *SharedMemoryProxy) Clear(ctx context.Context) error
Clear 清空本地记忆(不清空共享记忆)
func (*SharedMemoryProxy) Delete ¶
func (p *SharedMemoryProxy) Delete(ctx context.Context, id string) error
Delete 删除条目(代理到本地记忆)
func (*SharedMemoryProxy) Local ¶
func (p *SharedMemoryProxy) Local() memory.Memory
Local 返回本地原始记忆(用于测试和调试)
func (*SharedMemoryProxy) Save ¶
Save 保存条目
标记 _agent_id 和 _agent_name 元数据,写入本地记忆, 如果启用 WriteToShared 则同步写入共享记忆(失败仅输出 stderr 警告)。
func (*SharedMemoryProxy) Search ¶
func (p *SharedMemoryProxy) Search(ctx context.Context, query memory.SearchQuery) ([]memory.Entry, error)
Search 搜索记忆
合并本地记忆和共享记忆的搜索结果,按 ID 去重,本地结果优先。
func (*SharedMemoryProxy) Stats ¶
func (p *SharedMemoryProxy) Stats() memory.MemoryStats
Stats 返回本地记忆统计信息
type StateManager ¶
type StateManager interface {
// Turn 获取单轮对话状态(生命周期:单次 Run 调用)
Turn() TurnState
// Session 获取会话级状态(生命周期:一次会话/对话)
Session() SessionState
// Agent 获取 Agent 持久状态(生命周期:Agent 实例)
Agent() AgentState
// Global 获取全局共享状态(生命周期:应用程序)
Global() GlobalState
// NewTurn 创建新的轮次,重置 TurnState
NewTurn() TurnState
// Snapshot 创建当前状态的快照
Snapshot() StateSnapshot
// Restore 从快照恢复状态
Restore(snapshot StateSnapshot) error
}
StateManager 分层状态管理器接口 提供四层状态管理: Turn -> Session -> Agent -> Global
func StateManagerFromContext ¶
func StateManagerFromContext(ctx context.Context) StateManager
StateManagerFromContext 从 context 中获取 StateManager
type StateSnapshot ¶
type StateSnapshot struct {
Timestamp time.Time `json:"timestamp"`
SessionID string `json:"session_id"`
TurnData map[string]any `json:"turn_data"`
SessionData map[string]any `json:"session_data"`
AgentData map[string]any `json:"agent_data"`
TurnCount int `json:"turn_count"`
Iteration int `json:"iteration"`
Messages []Message `json:"messages"`
}
StateSnapshot 状态快照
type SubtaskInput ¶
type SubtaskInput struct {
// Task 子任务描述
Task string `json:"task" desc:"Description of the subtask to delegate" required:"true"`
}
SubtaskInput 子任务工具的输入参数
type Summarizer ¶
type Summarizer interface {
// Summarize 汇总执行结果
Summarize(ctx context.Context, goal string, steps []*planner.Step) (string, error)
}
Summarizer 结果汇总器接口
type SupervisorAgent ¶
type SupervisorAgent struct {
// contains filtered or unexported fields
}
SupervisorAgent 监督者 Agent
由一个 Manager Agent 动态决定将任务分派给哪个 Worker Agent。 Manager 的工具列表中包含所有 Worker 的 AgentAsTool 包装, 通过 LLM 的 tool call 机制自动选择合适的 worker。
线程安全:SupervisorAgent 是不可变的,创建后可安全并发使用。
func NewSupervisor ¶
func NewSupervisor(name string, manager Agent, workers []Agent, opts ...SupervisorOption) *SupervisorAgent
NewSupervisor 创建监督者 Agent
参数:
- name: Agent 名称
- manager: 管理者 Agent(需要配置 LLM,用于决策分派)
- workers: 工人 Agent 列表
- opts: 可选配置
func (*SupervisorAgent) Batch ¶
func (s *SupervisorAgent) Batch(ctx context.Context, inputs []Input, opts ...core.Option) ([]Output, error)
Batch 批量执行 Agent
func (*SupervisorAgent) BatchStream ¶
func (s *SupervisorAgent) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*SupervisorAgent) Collect ¶
func (s *SupervisorAgent) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*SupervisorAgent) Description ¶
func (s *SupervisorAgent) Description() string
Description 返回描述
func (*SupervisorAgent) InputSchema ¶
InputSchema 返回输入 Schema
func (*SupervisorAgent) Invoke ¶
func (s *SupervisorAgent) Invoke(ctx context.Context, input Input, opts ...core.Option) (Output, error)
Invoke 执行 Agent
func (*SupervisorAgent) LLM ¶
func (s *SupervisorAgent) LLM() llm.Provider
LLM 返回 LLM Provider(使用 manager 的 LLM)
func (*SupervisorAgent) OutputSchema ¶
OutputSchema 返回输出 Schema
func (*SupervisorAgent) Run ¶
Run 执行监督者流程
流程:
- 构建包含所有 worker 工具的消息列表
- 让 manager LLM 选择要调用的 worker
- 执行选中的 worker,将结果添加到消息历史
- 继续让 manager 决定下一步,直到 manager 不再调用工具
func (*SupervisorAgent) Stream ¶
func (s *SupervisorAgent) Stream(ctx context.Context, input Input, opts ...core.Option) (*stream.StreamReader[Output], error)
Stream 流式执行 Agent
func (*SupervisorAgent) Tools ¶
func (s *SupervisorAgent) Tools() []tool.Tool
Tools 返回工具列表(聚合 manager + 所有 worker 工具)
func (*SupervisorAgent) Transform ¶
func (s *SupervisorAgent) Transform(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (*stream.StreamReader[Output], error)
Transform 转换流
type SupervisorOption ¶
type SupervisorOption func(*SupervisorAgent)
SupervisorOption SupervisorAgent 专用选项
func WithSupervisorRounds ¶
func WithSupervisorRounds(n int) SupervisorOption
WithSupervisorRounds 设置 Supervisor 最大轮次
每轮 manager 选择一个 worker 执行。超过最大轮次时 返回 manager 当前的最终回答。 默认值: 10
type SwarmRunner ¶
type SwarmRunner struct {
// InitialAgent 初始 Agent
InitialAgent Agent
// MaxHandoffs 最大交接次数
MaxHandoffs int
// GlobalState 全局状态
GlobalState GlobalState
// Verbose 详细输出
Verbose bool
}
SwarmRunner 多 Agent 交接(handoff)运行器 自动处理 Agent 之间的交接
func NewSwarmRunner ¶
func NewSwarmRunner(initialAgent Agent) *SwarmRunner
NewSwarmRunner 创建 Swarm 运行器
type Team ¶
type Team struct {
// contains filtered or unexported fields
}
Team 团队 多个 Agent 组成的协作团队
线程安全:所有方法都是并发安全的
func (*Team) AddAgent ¶
AddAgent 添加 Agent 到团队
如果团队启用了共享记忆,新添加的 Agent 会自动包装 SharedMemoryProxy。
线程安全:此方法是并发安全的
func (*Team) BatchStream ¶
func (t *Team) BatchStream(ctx context.Context, inputs []Input, opts ...core.Option) (*stream.StreamReader[Output], error)
BatchStream 批量流式执行
func (*Team) Collect ¶
func (t *Team) Collect(ctx context.Context, input *stream.StreamReader[Input], opts ...core.Option) (Output, error)
Collect 收集流式输入并执行
func (*Team) SharedMemory ¶
func (t *Team) SharedMemory() *SharedMemory
SharedMemory 返回团队共享记忆(如未设置则返回 nil)
type TeamMode ¶
type TeamMode int
TeamMode 团队工作模式
const ( // TeamModeSequential 顺序执行模式 // Agent 按顺序依次执行,前一个的输出作为后一个的输入 TeamModeSequential TeamMode = iota // TeamModeHierarchical 层级模式 // 由 Manager Agent 协调和分配任务给其他 Agent TeamModeHierarchical // TeamModeCollaborative 协作模式 // 所有 Agent 并行工作,通过消息传递协作 TeamModeCollaborative // TeamModeRoundRobin 轮询模式 // Agent 轮流执行,直到达到目标 TeamModeRoundRobin )
type TeamOption ¶
type TeamOption func(*Team)
TeamOption 团队配置选项
func WithSharedMemory ¶
func WithSharedMemory(sm *SharedMemory) TeamOption
WithSharedMemory 设置团队共享记忆
启用后,所有 Agent 的 Memory 会被自动包装为 SharedMemoryProxy, Agent 的写入会同步到共享记忆,搜索会合并共享记忆的结果。 后续通过 AddAgent 添加的 Agent 也会自动包装。
type ToolCall ¶
type ToolCall struct {
ID string `json:"id"`
Name string `json:"name"`
Arguments map[string]any `json:"arguments"`
Result string `json:"result,omitempty"`
}
ToolCall 工具调用记录
type ToolCallRecord ¶
type ToolCallRecord struct {
// Name 工具名称
Name string `json:"name"`
// Arguments 工具参数
Arguments map[string]any `json:"arguments"`
// Result 工具结果
Result tool.Result `json:"result"`
}
ToolCallRecord 记录工具调用
type TransferToInput ¶
type TransferToInput struct {
// Message 传递给目标 Agent 的消息
Message string `json:"message" desc:"Message to pass to the target agent" required:"true"`
// Reason 转交原因
Reason string `json:"reason" desc:"Reason for transferring to this agent"`
// Context 额外上下文
Context map[string]any `json:"context" desc:"Additional context to pass"`
}
TransferToInput 转交工具的输入
type TurnState ¶
type TurnState interface {
// Get 获取值
Get(key string) (any, bool)
// Set 设置值
Set(key string, value any)
// Delete 删除值
Delete(key string)
// Clear 清空所有值
Clear()
// All 获取所有键值对
All() map[string]any
// Iteration 获取当前迭代次数(ReAct 循环)
Iteration() int
// SetIteration 设置迭代次数
SetIteration(n int)
// Messages 获取本轮消息
Messages() []Message
// AddMessage 添加消息
AddMessage(msg Message)
}
TurnState 单轮对话状态 生命周期:单次 Run 调用
type Vote ¶
type Vote struct {
// AgentID 投票 Agent ID
AgentID string `json:"agent_id"`
// AgentName 投票 Agent 名称
AgentName string `json:"agent_name"`
// Value 投票值
Value any `json:"value"`
// Weight 权重(用于加权投票)
Weight float64 `json:"weight"`
// Score 评分(用于最佳选择)
Score float64 `json:"score"`
// Reason 投票理由
Reason string `json:"reason"`
// Ranking 排序(用于 Borda 计数)
Ranking []any `json:"ranking,omitempty"`
// Timestamp 投票时间
Timestamp time.Time `json:"timestamp"`
// Metadata 元数据
Metadata map[string]any `json:"metadata,omitempty"`
}
Vote 投票
type WatchEvent ¶
type WatchEvent struct {
// Type 事件类型
Type WatchEventType `json:"type"`
// Agent Agent 信息
Agent *AgentInfo `json:"agent"`
// Timestamp 事件时间
Timestamp time.Time `json:"timestamp"`
}
WatchEvent 监听事件
type WatchEventType ¶
type WatchEventType string
WatchEventType 事件类型
const ( // EventRegistered Agent 注册 EventRegistered WatchEventType = "registered" // EventDeregistered Agent 注销 EventDeregistered WatchEventType = "deregistered" // EventHealthChanged 健康状态变化 EventHealthChanged WatchEventType = "health_changed" // EventUpdated Agent 更新 EventUpdated WatchEventType = "updated" )
type WeightedBalancer ¶
type WeightedBalancer struct {
// contains filtered or unexported fields
}
WeightedBalancer 加权负载均衡器
func (*WeightedBalancer) Select ¶
func (b *WeightedBalancer) Select(agents []*AgentInfo) *AgentInfo
Select 加权选择
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package a2a 实现 Google A2A (Agent-to-Agent) 协议
|
Package a2a 实现 Google A2A (Agent-to-Agent) 协议 |
|
Package artifact 提供 Agent 生成文件的版本化管理
|
Package artifact 提供 Agent 生成文件的版本化管理 |
|
Package semantic 提供语义函数功能
|
Package semantic 提供语义函数功能 |
|
Package skill 提供 Skill(技能)注册、发现和执行系统
|
Package skill 提供 Skill(技能)注册、发现和执行系统 |