本文章没有找到对应的语言版本
AIEINOAGENTMULTI-AGENT

Eino 学习第四阶段：生产级打磨与人机协作实战训练手册

📚 手册简介

欢迎来到第四阶段的学习！在前三个阶段，你已经从零开始，逐步构建了一个完整的多智能体专利代写系统，包含了主管Agent、检索分析员、撰写流水线和质量审核员。现在，我们将把这个"有趣的Demo"打磨成一个真正可用的生产级系统。

根据我们之前规划的路线图，第四阶段的目标是：

  • ✅ 可观测性增强：添加日志、监控、追踪
  • ✅ 人机协作机制：实现审批、审核编辑、追问模式
  • ✅ 状态持久化：支持中断恢复，保存长期运行的专利案卷
  • ✅ 性能优化：流式输出优化、并发控制、错误处理
  • ✅ 部署与服务化：将系统封装为HTTP服务

按照本手册，你将在4-5小时内完成全部学习任务，获得一个具备生产级别能力的专利代写智能系统。

第一部分：可观测性增强——让系统透明可控（60分钟）

在生产环境中，我们需要知道系统内部发生了什么：每个Agent花了多长时间、调用了什么工具、遇到了什么错误。Eino提供了完善的Callback机制来实现可观测性。

1.1 Callback机制概述

Callback可以理解为"钩子函数"，在Agent执行的关键节点自动触发：

回调点 触发时机 可用于
OnStart Agent开始执行 记录开始时间、输入内容
OnEnd Agent执行成功结束 记录耗时、输出内容
OnError Agent执行出错 记录错误信息、告警
OnSuspend Agent暂停（等待人工） 记录暂停原因、触发通知

1.2 实现自定义Callback Handler

创建pkg/observability/callback.go：

package observability

import (
    "context"
    "fmt"
    "log"
    "time"
    
    "github.com/cloudwego/eino/callbacks"
    "github.com/cloudwego/eino/schema"
)

// PatentCallbackHandler 专利系统的自定义Callback处理器
type PatentCallbackHandler struct {
    startTime time.Time
    agentName string
}

// NewPatentCallbackHandler 创建新的Callback处理器
func NewPatentCallbackHandler() *PatentCallbackHandler {
    return &PatentCallbackHandler{}
}

// OnStart 在Agent开始执行时调用
func (h *PatentCallbackHandler) OnStart(ctx context.Context, info *callbacks.RunInfo, input *callbacks.Input) context.Context {
    h.startTime = time.Now()
    h.agentName = info.Name
    
    // 记录开始信息
    log.Printf("🟢 [%s] 开始执行 at %s", info.Name, h.startTime.Format("15:04:05.000"))
    
    // 如果有输入消息，记录摘要
    if input != nil && len(input.Messages) > 0 {
        for i, msg := range input.Messages {
            if msg.Content != "" {
                contentPreview := msg.Content
                if len(contentPreview) > 100 {
                    contentPreview = contentPreview[:100] + "..."
                }
                log.Printf("   📥 输入[%d]: [%s] %s", i, msg.Role, contentPreview)
            }
        }
    }
    
    return ctx
}

// OnEnd 在Agent成功结束时调用
func (h *PatentCallbackHandler) OnEnd(ctx context.Context, info *callbacks.RunInfo, output *callbacks.Output) {
    elapsed := time.Since(h.startTime)
    
    log.Printf("🟢 [%s] 执行完成，耗时: %v", info.Name, elapsed)
    
    // 如果有输出，记录摘要
    if output != nil && len(output.Messages) > 0 {
        for i, msg := range output.Messages {
            if msg.Content != "" {
                contentPreview := msg.Content
                if len(contentPreview) > 100 {
                    contentPreview = contentPreview[:100] + "..."
                }
                log.Printf("   📤 输出[%d]: %s", i, contentPreview)
            }
        }
    }
}

// OnError 在Agent执行出错时调用
func (h *PatentCallbackHandler) OnError(ctx context.Context, info *callbacks.RunInfo, err error) {
    elapsed := time.Since(h.startTime)
    
    log.Printf("🔴 [%s] 执行出错 at %v，耗时: %v", info.Name, time.Now().Format("15:04:05.000"), elapsed)
    log.Printf("   ❌ 错误: %v", err)
}

// OnSuspend 在Agent暂停（等待人工）时调用
func (h *PatentCallbackHandler) OnSuspend(ctx context.Context, info *callbacks.RunInfo, suspend *callbacks.Suspend) {
    log.Printf("🟡 [%s] 暂停执行 at %s", info.Name, time.Now().Format("15:04:05.000"))
    
    if suspend != nil {
        log.Printf("   ⏸️ 暂停原因: %s", suspend.Reason)
        if suspend.Data != nil {
            log.Printf("   📋 待处理数据: %v", suspend.Data)
        }
    }
}

// 实现Handler接口
func (h *PatentCallbackHandler) IsCallbacksEnabled() bool { return true }
func (h *PatentCallbackHandler) GetRunInfo() *callbacks.RunInfo { 
    return &callbacks.RunInfo{Name: h.agentName} 
}

1.3 创建可观测的Agent运行器

创建pkg/observability/runner.go：

package observability

import (
    "context"
    
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/callbacks"
)

// NewObservableRunner 创建带有Callback监控的Runner
func NewObservableRunner(ctx context.Context, agent adk.Agent) *adk.Runner {
    // 创建Callback处理器
    handler := NewPatentCallbackHandler()
    
    // 创建Runner并注入Callback
    runner := adk.NewRunner(ctx, adk.RunnerConfig{
        Agent: agent,
        Callbacks: []callbacks.Handler{handler},
    })
    
    return runner
}

// WithTrace 添加追踪ID到上下文（用于分布式追踪）
func WithTrace(ctx context.Context, traceID string) context.Context {
    ctx = context.WithValue(ctx, "trace_id", traceID)
    return ctx
}

// GetTraceID 从上下文获取追踪ID
func GetTraceID(ctx context.Context) string {
    if traceID, ok := ctx.Value("trace_id").(string); ok {
        return traceID
    }
    return "unknown"
}

1.4 使用Callback的示例

修改cmd/patent_supervisor/main.go，集成Callback：

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"
    
    "github.com/joho/godotenv"
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/schema"
    
    "patent-agent-learning/pkg/agents"
    "patent-agent-learning/pkg/observability"
)

func main() {
    // 设置日志格式（包含时间戳）
    log.SetFlags(log.Ldate | log.Ltime | log.Lmicroseconds)
    
    // 加载环境变量
    if err := godotenv.Load(); err != nil {
        log.Println("警告: 未找到.env文件，使用系统环境变量")
    }
    
    ctx := context.Background()
    
    // 添加追踪ID
    traceID := fmt.Sprintf("patent_%d", time.Now().UnixNano())
    ctx = observability.WithTrace(ctx, traceID)
    
    log.Printf("🚀 启动专利撰写系统，追踪ID: %s", traceID)
    
    // 创建专利主管系统
    supervisor, err := agents.NewPatentSupervisor(ctx)
    if err != nil {
        log.Fatalf("创建专利主管失败: %v", err)
    }
    
    // 创建可观测的Runner
    runner := observability.NewObservableRunner(ctx, supervisor)
    
    // 准备用户输入
    techDisclosure := `本发明涉及一种基于区块链的供应链金融风控方法。
    
【技术背景】
传统供应链金融依赖中心化信用评估，存在数据造假、信息不对称等问题。

【发明内容】
本发明提出一种基于区块链的供应链金融风控方法，包括：
1. 将供应链上的交易数据（订单、物流、支付）上链存储
2. 通过智能合约自动验证交易的真实性
3. 基于验证结果动态调整信用额度`
    
    userMessage := schema.UserMessage(fmt.Sprintf(`
请帮我撰写一份专利文档。

技术交底书：
%s
    `, techDisclosure))
    
    log.Println("📥 收到用户请求")
    
    // 运行系统
    input := &adk.AgentInput{
        Messages:         []*schema.Message{userMessage},
        EnableStreaming:  true,
    }
    
    iter := runner.Run(ctx, input)
    
    // 处理输出...
    for {
        event, ok := iter.Next()
        if !ok {
            break
        }
        // ... 事件处理逻辑保持不变
    }
    
    log.Println("✅ 系统执行完成")
}

第二部分：人机协作（Human-in-the-Loop）机制（75分钟）

在专利撰写这样的专业领域，完全自动化的风险太高。我们需要让人参与到关键决策环节。Eino ADK提供了完善的人机协作（HITL）机制。

2.1 三种人机协作模式

模式 说明 在专利场景中的应用
审批模式 敏感操作前需人工确认 最终文档导出前需专利工程师确认
审核编辑模式 人工可修改Agent生成的参数 修改权利要求、补充实施例
追问模式 Agent发现信息缺失时主动提问 技术交底书不明确时向用户追问

2.2 审批模式实现

创建pkg/hitl/approval.go：

package hitl

import (
    "context"
    "encoding/json"
    "fmt"
    
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/schema"
)

// ApprovalRequest 审批请求结构
type ApprovalRequest struct {
    TaskID      string      `json:"task_id"`      // 任务ID
    AgentName   string      `json:"agent_name"`   // 请求审批的Agent
    Action      string      `json:"action"`       // 请求的操作
    Content     string      `json:"content"`      // 待审批的内容
    Reason      string      `json:"reason"`       // 请求审批的原因
    ExpireAt    int64       `json:"expire_at"`    // 过期时间戳
    Metadata    map[string]interface{} `json:"metadata"` // 附加元数据
}

// ApprovalResponse 审批响应结构
type ApprovalResponse struct {
    Approved    bool        `json:"approved"`     // 是否批准
    Comment     string      `json:"comment"`       // 审批意见
    ModifiedContent string  `json:"modified_content,omitempty"` // 修改后的内容
    Reviewer    string      `json:"reviewer"`      // 审批人
}

// ApprovalManager 审批管理器
type ApprovalManager struct {
    // 实际应用中，这里应该连接数据库或消息队列
    pendingApprovals map[string]*ApprovalRequest
}

func NewApprovalManager() *ApprovalManager {
    return &ApprovalManager{
        pendingApprovals: make(map[string]*ApprovalRequest),
    }
}

// RequestApproval 发起审批请求
func (m *ApprovalManager) RequestApproval(ctx context.Context, agent adk.Agent, req *ApprovalRequest) (*adk.AgentAction, error) {
    // 存储审批请求
    m.pendingApprovals[req.TaskID] = req
    
    // 创建Suspend Action，等待人工审批
    suspendData, _ := json.Marshal(req)
    
    return &adk.AgentAction{
        Suspend: &adk.AgentSuspend{
            Reason: fmt.Sprintf("需要人工审批: %s", req.Reason),
            Data:   suspendData,
        },
    }, nil
}

// HandleApproval 处理审批结果
func (m *ApprovalManager) HandleApproval(ctx context.Context, taskID string, resp *ApprovalResponse) error {
    req, exists := m.pendingApprovals[taskID]
    if !exists {
        return fmt.Errorf("审批任务不存在: %s", taskID)
    }
    
    if !resp.Approved {
        // 审批拒绝，需要通知Agent重新处理
        return fmt.Errorf("审批被拒绝: %s", resp.Comment)
    }
    
    // 审批通过，将结果存储供Agent恢复时使用
    // 实际应用中，这里应该通过某种方式通知等待的Agent
    delete(m.pendingApprovals, taskID)
    
    return nil
}

2.3 追问模式实现

创建pkg/hitl/clarify.go：

package hitl

import (
    "context"
    "fmt"
    
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/schema"
)

// ClarificationQuestion 澄清问题
type ClarificationQuestion struct {
    QuestionID  string   `json:"question_id"`  // 问题ID
    Question    string   `json:"question"`      // 问题内容
    Options     []string `json:"options,omitempty"` // 可选答案（如果是选择题）
    Context     string   `json:"context"`       // 上下文信息
    AgentName   string   `json:"agent_name"`    // 提问的Agent
}

// ClarificationAnswer 用户回答
type ClarificationAnswer struct {
    QuestionID string `json:"question_id"`
    Answer     string `json:"answer"`
}

// AskForClarification 发起追问
func AskForClarification(ctx context.Context, agent adk.Agent, question *ClarificationQuestion) (*adk.AgentAction, error) {
    // 创建Suspend Action，等待用户回答
    return &adk.AgentAction{
        Suspend: &adk.AgentSuspend{
            Reason: fmt.Sprintf("需要澄清: %s", question.Question),
            Data:   question,
        },
    }, nil
}

// CreateClarificationAgent 创建带有追问能力的Agent
func CreateClarificationAgent(baseAgent adk.Agent, mgr *ApprovalManager) adk.Agent {
    // 这是一个包装器，拦截Agent的Suspend动作并处理
    // 实际实现需要更复杂的逻辑，这里简化示意
    return &ClarificationAgent{
        baseAgent: baseAgent,
        mgr:       mgr,
    }
}

type ClarificationAgent struct {
    baseAgent adk.Agent
    mgr       *ApprovalManager
}

func (a *ClarificationAgent) Name() string {
    return a.baseAgent.Name()
}

func (a *ClarificationAgent) Description() string {
    return a.baseAgent.Description()
}

func (a *ClarificationAgent) Run(ctx context.Context, input *adk.AgentInput) adk.AgentIterator {
    // 在运行前注入自定义逻辑
    // 实际实现中，需要处理Suspend的恢复等
    return a.baseAgent.Run(ctx, input)
}

2.4 在专利系统中集成人机协作

修改检索分析员Agent，添加追问能力：

// 在检索分析员Agent的Instruction中添加
Instruction: `...（原有内容）
当你发现技术交底书信息不明确时，可以主动向用户提问澄清。
例如：
- 如果发明点描述过于模糊，可以询问具体的技术实现细节
- 如果术语不明确，可以询问术语的定义
- 如果缺少必要信息，可以询问补充

提问时，请使用AskForClarification工具。`,

第三部分：状态持久化与中断恢复（45分钟）

在真实的专利撰写中，一个案卷可能需要几天才能完成。系统需要支持中断恢复，让用户随时可以保存进度，稍后继续。

3.1 Session的序列化与存储

创建pkg/persistence/session_store.go：

package persistence

import (
    "context"
    "encoding/json"
    "fmt"
    "os"
    "path/filepath"
    "time"
    
    "github.com/cloudwego/eino/adk"
)

// SessionStore Session存储接口
type SessionStore interface {
    Save(ctx context.Context, sessionID string, session *adk.Session) error
    Load(ctx context.Context, sessionID string) (*adk.Session, error)
    List(ctx context.Context) ([]string, error)
    Delete(ctx context.Context, sessionID string) error
}

// FileSessionStore 文件系统实现的Session存储
type FileSessionStore struct {
    baseDir string
}

// NewFileSessionStore 创建文件系统Session存储
func NewFileSessionStore(baseDir string) (*FileSessionStore, error) {
    // 确保目录存在
    if err := os.MkdirAll(baseDir, 0755); err != nil {
        return nil, fmt.Errorf("创建存储目录失败: %v", err)
    }
    
    return &FileSessionStore{
        baseDir: baseDir,
    }, nil
}

// Save 保存Session到文件
func (s *FileSessionStore) Save(ctx context.Context, sessionID string, session *adk.Session) error {
    // 序列化Session
    data, err := json.MarshalIndent(session, "", "  ")
    if err != nil {
        return fmt.Errorf("序列化Session失败: %v", err)
    }
    
    // 写入文件
    filePath := filepath.Join(s.baseDir, fmt.Sprintf("%s.json", sessionID))
    if err := os.WriteFile(filePath, data, 0644); err != nil {
        return fmt.Errorf("写入Session文件失败: %v", err)
    }
    
    return nil
}

// Load 从文件加载Session
func (s *FileSessionStore) Load(ctx context.Context, sessionID string) (*adk.Session, error) {
    filePath := filepath.Join(s.baseDir, fmt.Sprintf("%s.json", sessionID))
    
    data, err := os.ReadFile(filePath)
    if err != nil {
        if os.IsNotExist(err) {
            return nil, fmt.Errorf("Session不存在: %s", sessionID)
        }
        return nil, fmt.Errorf("读取Session文件失败: %v", err)
    }
    
    var session adk.Session
    if err := json.Unmarshal(data, &session); err != nil {
        return nil, fmt.Errorf("反序列化Session失败: %v", err)
    }
    
    return &session, nil
}

// List 列出所有Session
func (s *FileSessionStore) List(ctx context.Context) ([]string, error) {
    files, err := os.ReadDir(s.baseDir)
    if err != nil {
        return nil, fmt.Errorf("读取存储目录失败: %v", err)
    }
    
    var sessions []string
    for _, file := range files {
        if !file.IsDir() && filepath.Ext(file.Name()) == ".json" {
            sessions = append(sessions, file.Name()[:len(file.Name())-5])
        }
    }
    
    return sessions, nil
}

// Delete 删除Session
func (s *FileSessionStore) Delete(ctx context.Context, sessionID string) error {
    filePath := filepath.Join(s.baseDir, fmt.Sprintf("%s.json", sessionID))
    if err := os.Remove(filePath); err != nil {
        if os.IsNotExist(err) {
            return nil // 文件不存在，视为已删除
        }
        return fmt.Errorf("删除Session文件失败: %v", err)
    }
    return nil
}

3.2 持久化Runner

创建pkg/persistence/persistent_runner.go：

package persistence

import (
    "context"
    "fmt"
    "time"
    
    "github.com/cloudwego/eino/adk"
)

// PersistentRunner 支持持久化的Runner
type PersistentRunner struct {
    runner      *adk.Runner
    sessionStore SessionStore
    sessionID   string
}

// NewPersistentRunner 创建持久化Runner
func NewPersistentRunner(runner *adk.Runner, store SessionStore, sessionID string) *PersistentRunner {
    return &PersistentRunner{
        runner:      runner,
        sessionStore: store,
        sessionID:   sessionID,
    }
}

// RunWithPersistence 运行并自动保存状态
func (p *PersistentRunner) RunWithPersistence(ctx context.Context, input *adk.AgentInput) (adk.AgentIterator, error) {
    // 尝试加载已有Session
    session, err := p.sessionStore.Load(ctx, p.sessionID)
    if err == nil {
        // 找到已保存的Session，恢复
        ctx = adk.WithSession(ctx, session)
        fmt.Printf("🔄 恢复Session: %s\n", p.sessionID)
    } else {
        // 创建新Session
        session = adk.NewSession()
        ctx = adk.WithSession(ctx, session)
        fmt.Printf("🆕 创建新Session: %s\n", p.sessionID)
    }
    
    // 运行Agent
    iter := p.runner.Run(ctx, input)
    
    // 在实际应用中，应该在每次Suspend或结束时保存Session
    // 这里简单示例：在迭代过程中定期保存
    return &persistentIterator{
        iter:        iter,
        runner:      p,
        ctx:         ctx,
        lastSave:    time.Now(),
    }, nil
}

// SaveCurrent 保存当前Session
func (p *PersistentRunner) SaveCurrent(ctx context.Context) error {
    session := adk.GetSession(ctx)
    if session == nil {
        return fmt.Errorf("上下文中没有Session")
    }
    
    return p.sessionStore.Save(ctx, p.sessionID, session)
}

// persistentIterator 包装的迭代器，自动保存状态
type persistentIterator struct {
    iter     adk.AgentIterator
    runner   *PersistentRunner
    ctx      context.Context
    lastSave time.Time
}

func (p *persistentIterator) Next() (*adk.AgentEvent, bool) {
    event, ok := p.iter.Next()
    
    // 每隔一段时间或遇到特定事件时保存
    if ok && time.Since(p.lastSave) > 30*time.Second {
        p.runner.SaveCurrent(p.ctx)
        p.lastSave = time.Now()
    }
    
    // 如果是Suspend事件，立即保存
    if ok && event != nil && event.Action != nil && event.Action.Suspend != nil {
        p.runner.SaveCurrent(p.ctx)
        fmt.Println("⏸️ Agent暂停，状态已保存")
    }
    
    return event, ok
}

第四部分：性能优化与错误处理（45分钟）

4.1 流式输出优化

创建pkg/optimization/streaming.go：

package optimization

import (
    "context"
    "fmt"
    "strings"
    "time"
    
    "github.com/cloudwego/eino/adk"
    "github.com/cloudwego/eino/schema"
)

// StreamingBuffer 流式输出缓冲优化
type StreamingBuffer struct {
    buffer     strings.Builder
    flushSize  int           // 触发刷新的字符数
    flushInterval time.Duration // 最大等待时间
    lastFlush  time.Time
    onFlush    func(string)   // 刷新回调
}

func NewStreamingBuffer(flushSize int, flushInterval time.Duration, onFlush func(string)) *StreamingBuffer {
    return &StreamingBuffer{
        flushSize:     flushSize,
        flushInterval: flushInterval,
        lastFlush:     time.Now(),
        onFlush:       onFlush,
    }
}

func (b *StreamingBuffer) Write(chunk string) {
    b.buffer.WriteString(chunk)
    
    // 检查是否需要刷新
    if b.buffer.Len() >= b.flushSize || time.Since(b.lastFlush) > b.flushInterval {
        b.Flush()
    }
}

func (b *StreamingBuffer) Flush() {
    if b.buffer.Len() > 0 {
        b.onFlush(b.buffer.String())
        b.buffer.Reset()
        b.lastFlush = time.Now()
    }
}

// OptimizedStreamingHandler 优化的流式处理Handler
type OptimizedStreamingHandler struct {
    buffer *StreamingBuffer
}

func NewOptimizedStreamingHandler(flushSize int, flushInterval time.Duration) *OptimizedStreamingHandler {
    return &OptimizedStreamingHandler{
        buffer: NewStreamingBuffer(flushSize, flushInterval, func(s string) {
            fmt.Print(s) // 实际应用中可能是WebSocket发送
        }),
    }
}

func (h *OptimizedStreamingHandler) HandleEvent(event *adk.AgentEvent) {
    if event == nil || event.Output == nil || event.Output.MessageOutput == nil {
        return
    }
    
    msg, err := event.Output.MessageOutput.GetMessage()
    if err != nil {
        return
    }
    
    if msg.Content != "" && event.Output.MessageOutput.IsStreaming {
        h.buffer.Write(msg.Content)
    }
}

4.2 错误处理与重试机制

创建pkg/optimization/retry.go：

package optimization

import (
    "context"
    "fmt"
    "time"
    
    "github.com/cloudwego/eino/adk"
)

// RetryConfig 重试配置
type RetryConfig struct {
    MaxAttempts     int           // 最大重试次数
    InitialInterval time.Duration // 初始重试间隔
    MaxInterval     time.Duration // 最大重试间隔
    Multiplier      float64       // 间隔倍数
}

// DefaultRetryConfig 默认重试配置
var DefaultRetryConfig = RetryConfig{
    MaxAttempts:     3,
    InitialInterval: 1 * time.Second,
    MaxInterval:     10 * time.Second,
    Multiplier:      2.0,
}

// WithRetry 添加重试能力的Agent包装器
func WithRetry(agent adk.Agent, config RetryConfig) adk.Agent {
    return &retryAgent{
        baseAgent: agent,
        config:    config,
    }
}

type retryAgent struct {
    baseAgent adk.Agent
    config    RetryConfig
}

func (r *retryAgent) Name() string {
    return r.baseAgent.Name()
}

func (r *retryAgent) Description() string {
    return r.baseAgent.Description()
}

func (r *retryAgent) Run(ctx context.Context, input *adk.AgentInput) adk.AgentIterator {
    return &retryIterator{
        baseIter:    r.baseAgent.Run(ctx, input),
        agent:       r.baseAgent,
        ctx:         ctx,
        input:       input,
        attempt:     1,
        config:      r.config,
        lastError:   nil,
    }
}

type retryIterator struct {
    baseIter    adk.AgentIterator
    agent       adk.Agent
    ctx         context.Context
    input       *adk.AgentInput
    attempt     int
    config      RetryConfig
    lastError   error
}

func (r *retryIterator) Next() (*adk.AgentEvent, bool) {
    for {
        event, ok := r.baseIter.Next()
        
        // 如果有错误，检查是否需要重试
        if ok && event != nil && event.Err != nil {
            r.lastError = event.Err
            
            if r.attempt < r.config.MaxAttempts {
                // 计算等待时间
                waitTime := r.config.InitialInterval
                for i := 1; i < r.attempt; i++ {
                    waitTime = time.Duration(float64(waitTime) * r.config.Multiplier)
                    if waitTime > r.config.MaxInterval {
                        waitTime = r.config.MaxInterval
                        break
                    }
                }
                
                fmt.Printf("🔄 Agent执行失败，%d秒后重试 (第%d次/%d次): %v\n", 
                    waitTime/time.Second, r.attempt, r.config.MaxAttempts, r.lastError)
                
                time.Sleep(waitTime)
                
                // 重试
                r.attempt++
                r.baseIter = r.agent.Run(r.ctx, r.input)
                continue
            }
            
            // 超过重试次数，返回错误
            return event, ok
        }
        
        // 正常事件或结束
        return event, ok
    }
}

4.3 并发控制

创建pkg/optimization/concurrency.go：

package optimization

import (
    "context"
    "sync"
    
    "github.com/cloudwego/eino/adk"
)

// RateLimiter 速率限制器
type RateLimiter struct {
    semaphore chan struct{}
}

func NewRateLimiter(maxConcurrent int) *RateLimiter {
    return &RateLimiter{
        semaphore: make(chan struct{}, maxConcurrent),
    }
}

func (r *RateLimiter) Acquire() {
    r.semaphore <- struct{}{}
}

func (r *RateLimiter) Release() {
    <-r.semaphore
}

// WithRateLimit 添加速率限制的Agent包装器
func WithRateLimit(agent adk.Agent, limiter *RateLimiter) adk.Agent {
    return &rateLimitAgent{
        baseAgent: agent,
        limiter:   limiter,
    }
}

type rateLimitAgent struct {
    baseAgent adk.Agent
    limiter   *RateLimiter
    mu        sync.Mutex
}

func (r *rateLimitAgent) Name() string {
    return r.baseAgent.Name()
}

func (r *rateLimitAgent) Description() string {
    return r.baseAgent.Description()
}

func (r *rateLimitAgent) Run(ctx context.Context, input *adk.AgentInput) adk.AgentIterator {
    // 获取令牌
    r.limiter.Acquire()
    
    // 创建包装的迭代器，在迭代结束时释放令牌
    return &rateLimitIterator{
        baseIter: r.baseAgent.Run(ctx, input),
        limiter:  r.limiter,
        released: false,
    }
}

type rateLimitIterator struct {
    baseIter adk.AgentIterator
    limiter  *RateLimiter
    released bool
    mu       sync.Mutex
}

func (r *rateLimitIterator) Next() (*adk.AgentEvent, bool) {
    event, ok := r.baseIter.Next()
    
    // 如果迭代结束，释放令牌
    if !ok {
        r.mu.Lock()
        if !r.released {
            r.limiter.Release()
            r.released = true
        }
        r.mu.Unlock()
    }
    
    return event, ok
}

第五部分：部署与服务化（60分钟）

最后，我们将系统封装为HTTP服务，提供RESTful API供前端或其他服务调用。

5.1 HTTP服务实现

创建cmd/server/main.go：

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
    
    "github.com/gorilla/mux"
    "github.com/joho/godotenv"
    "github.com/rs/cors"
    
    "patent-agent-learning/pkg/agents"
    "patent-agent-learning/pkg/observability"
    "patent-agent-learning/pkg/optimization"
    "patent-agent-learning/pkg/persistence"
)

// ServerConfig 服务器配置
type ServerConfig struct {
    Port            string
    SessionStoreDir string
    MaxConcurrent   int
    EnableTracing   bool
}

// PatentServer 专利撰写服务器
type PatentServer struct {
    config         ServerConfig
    supervisor     adk.Agent
    sessionStore   persistence.SessionStore
    rateLimiter    *optimization.RateLimiter
    activeSessions sync.Map
}

// NewPatentServer 创建服务器
func NewPatentServer(config ServerConfig) (*PatentServer, error) {
    ctx := context.Background()
    
    // 创建主管Agent
    supervisor, err := agents.NewPatentSupervisor(ctx)
    if err != nil {
        return nil, fmt.Errorf("创建主管Agent失败: %v", err)
    }
    
    // 创建Session存储
    sessionStore, err := persistence.NewFileSessionStore(config.SessionStoreDir)
    if err != nil {
        return nil, fmt.Errorf("创建Session存储失败: %v", err)
    }
    
    // 创建速率限制器
    rateLimiter := optimization.NewRateLimiter(config.MaxConcurrent)
    
    return &PatentServer{
        config:       config,
        supervisor:   supervisor,
        sessionStore: sessionStore,
        rateLimiter:  rateLimiter,
    }, nil
}

// Start 启动服务器
func (s *PatentServer) Start() error {
    router := mux.NewRouter()
    
    // API路由
    api := router.PathPrefix("/api/v1").Subrouter()
    api.HandleFunc("/health", s.handleHealth).Methods("GET")
    api.HandleFunc("/patent/write", s.handlePatentWrite).Methods("POST")
    api.HandleFunc("/patent/session/{sessionId}", s.handleGetSession).Methods("GET")
    api.HandleFunc("/patent/session/{sessionId}/resume", s.handleResumeSession).Methods("POST")
    api.HandleFunc("/patent/sessions", s.handleListSessions).Methods("GET")
    
    // CORS配置
    c := cors.New(cors.Options{
        AllowedOrigins:   []string{"*"},
        AllowedMethods:   []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
        AllowedHeaders:   []string{"*"},
        AllowCredentials: true,
    })
    
    handler := c.Handler(router)
    
    // 创建HTTP服务器
    server := &http.Server{
        Addr:         ":" + s.config.Port,
        Handler:      handler,
        ReadTimeout:  60 * time.Second,
        WriteTimeout: 300 * time.Second,
        IdleTimeout:  120 * time.Second,
    }
    
    // 优雅关闭
    done := make(chan os.Signal, 1)
    signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
    
    go func() {
        <-done
        log.Println("🛑 正在关闭服务器...")
        
        ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
        defer cancel()
        
        if err := server.Shutdown(ctx); err != nil {
            log.Fatalf("服务器关闭失败: %v", err)
        }
    }()
    
    log.Printf("🚀 服务器启动在端口 %s", s.config.Port)
    return server.ListenAndServe()
}

// 请求/响应结构
type PatentWriteRequest struct {
    SessionID       string `json:"sessionId,omitempty"` // 可选，续写已有Session
    TechDisclosure  string `json:"techDisclosure"`      // 技术交底书
    EnableStreaming bool   `json:"enableStreaming"`     // 是否启用流式输出
}

type PatentWriteResponse struct {
    SessionID   string `json:"sessionId"`
    Status      string `json:"status"`
    Message     string `json:"message,omitempty"`
    Result      string `json:"result,omitempty"`
    SuspendInfo *struct {
        Reason string `json:"reason"`
        Data   any    `json:"data"`
    } `json:"suspendInfo,omitempty"`
}

type SessionInfo struct {
    SessionID   string    `json:"sessionId"`
    CreatedAt   time.Time `json:"createdAt"`
    UpdatedAt   time.Time `json:"updatedAt"`
    Status      string    `json:"status"`
    HasResult   bool      `json:"hasResult"`
}

// 处理专利撰写请求
func (s *PatentServer) handlePatentWrite(w http.ResponseWriter, r *http.Request) {
    var req PatentWriteRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "无效的请求格式", http.StatusBadRequest)
        return
    }
    
    ctx := r.Context()
    
    // 获取或创建SessionID
    sessionID := req.SessionID
    if sessionID == "" {
        sessionID = fmt.Sprintf("patent_%d", time.Now().UnixNano())
    }
    
    // 创建持久化Runner
    runner := observability.NewObservableRunner(ctx, s.supervisor)
    persistentRunner := persistence.NewPersistentRunner(runner, s.sessionStore, sessionID)
    
    // 准备输入
    input := &adk.AgentInput{
        Messages: []*schema.Message{
            schema.UserMessage(req.TechDisclosure),
        },
        EnableStreaming: req.EnableStreaming,
    }
    
    // 运行
    iter, err := persistentRunner.RunWithPersistence(ctx, input)
    if err != nil {
        http.Error(w, fmt.Sprintf("执行失败: %v", err), http.StatusInternalServerError)
        return
    }
    
    // 处理输出
    var finalResponse string
    var suspendInfo *struct {
        Reason string `json:"reason"`
        Data   any    `json:"data"`
    }
    
    for {
        event, ok := iter.Next()
        if !ok {
            break
        }
        
        if event.Err != nil {
            log.Printf("执行出错: %v", event.Err)
            continue
        }
        
        if event.Action != nil && event.Action.Suspend != nil {
            suspendInfo = &struct {
                Reason string `json:"reason"`
                Data   any    `json:"data"`
            }{
                Reason: event.Action.Suspend.Reason,
                Data:   event.Action.Suspend.Data,
            }
            
            // 保存Session状态
            persistentRunner.SaveCurrent(ctx)
            break
        }
        
        if event.Output != nil && event.Output.MessageOutput != nil {
            msg, _ := event.Output.MessageOutput.GetMessage()
            if msg != nil && msg.Content != "" {
                finalResponse += msg.Content
            }
        }
    }
    
    // 返回响应
    resp := PatentWriteResponse{
        SessionID: sessionID,
        Status:    "completed",
        Result:    finalResponse,
    }
    
    if suspendInfo != nil {
        resp.Status = "suspended"
        resp.SuspendInfo = suspendInfo
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(resp)
}

// 健康检查
func (s *PatentServer) handleHealth(w http.ResponseWriter, r *http.Request) {
    json.NewEncoder(w).Encode(map[string]string{
        "status": "ok",
        "time":   time.Now().Format(time.RFC3339),
    })
}

// 获取Session信息
func (s *PatentServer) handleGetSession(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    sessionID := vars["sessionId"]
    
    session, err := s.sessionStore.Load(r.Context(), sessionID)
    if err != nil {
        http.Error(w, fmt.Sprintf("Session不存在: %v", err), http.StatusNotFound)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(session)
}

// 续写Session
func (s *PatentServer) handleResumeSession(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    sessionID := vars["sessionId"]
    
    // 续写逻辑（需要恢复Session并继续执行）
    // 这里简化处理
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]string{
        "sessionId": sessionID,
        "status":    "resumed",
    })
}

// 列出所有Sessions
func (s *PatentServer) handleListSessions(w http.ResponseWriter, r *http.Request) {
    sessions, err := s.sessionStore.List(r.Context())
    if err != nil {
        http.Error(w, fmt.Sprintf("获取Session列表失败: %v", err), http.StatusInternalServerError)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string][]string{
        "sessions": sessions,
    })
}

func main() {
    // 加载环境变量
    if err := godotenv.Load(); err != nil {
        log.Println("警告: 未找到.env文件，使用系统环境变量")
    }
    
    // 服务器配置
    config := ServerConfig{
        Port:            getEnv("PORT", "8080"),
        SessionStoreDir: getEnv("SESSION_STORE_DIR", "./sessions"),
        MaxConcurrent:   getEnvAsInt("MAX_CONCURRENT", 5),
        EnableTracing:   getEnvAsBool("ENABLE_TRACING", true),
    }
    
    // 创建服务器
    server, err := NewPatentServer(config)
    if err != nil {
        log.Fatalf("创建服务器失败: %v", err)
    }
    
    // 启动
    if err := server.Start(); err != nil && err != http.ErrServerClosed {
        log.Fatalf("服务器启动失败: %v", err)
    }
}

// 辅助函数
func getEnv(key, defaultValue string) string {
    if value := os.Getenv(key); value != "" {
        return value
    }
    return defaultValue
}

func getEnvAsInt(key string, defaultValue int) int {
    if value := os.Getenv(key); value != "" {
        var intVal int
        fmt.Sscanf(value, "%d", &intVal)
        return intVal
    }
    return defaultValue
}

func getEnvAsBool(key string, defaultValue bool) bool {
    if value := os.Getenv(key); value != "" {
        return value == "true" || value == "1" || value == "yes"
    }
    return defaultValue
}

5.2 创建Dockerfile

创建Dockerfile：

# 构建阶段
FROM golang:1.21-alpine AS builder

WORKDIR /app

# 复制go mod文件
COPY go.mod go.sum ./
RUN go mod download

# 复制源代码
COPY . .

# 构建
RUN CGO_ENABLED=0 GOOS=linux go build -o patent-server ./cmd/server

# 运行阶段
FROM alpine:latest

RUN apk --no-cache add ca-certificates

WORKDIR /root/

# 从构建阶段复制二进制文件
COPY --from=builder /app/patent-server .

# 创建sessions目录
RUN mkdir -p /root/sessions

# 暴露端口
EXPOSE 8080

# 运行
CMD ["./patent-server"]

5.3 创建docker-compose.yml

创建docker-compose.yml：

version: '3.8'

services:
  patent-agent:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - OPENAI_MODEL=${OPENAI_MODEL:-gpt-4}
      - SESSION_STORE_DIR=/app/sessions
      - MAX_CONCURRENT=5
      - ENABLE_TRACING=true
    volumes:
      - ./sessions:/app/sessions
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:8080/api/v1/health"]
      interval: 30s
      timeout: 10s
      retries: 3

5.4 创建Makefile

创建Makefile：

.PHONY: build run test docker-build docker-run clean

# 构建
build:
	go build -o bin/patent-server ./cmd/server

# 运行
run:
	go run ./cmd/server

# 测试
test:
	go test ./...

# Docker构建
docker-build:
	docker build -t patent-agent:latest .

# Docker运行
docker-run:
	docker-compose up -d

# Docker停止
docker-stop:
	docker-compose down

# 清理
clean:
	rm -rf bin/
	rm -rf sessions/
	docker-compose down -v

# 查看日志
logs:
	docker-compose logs -f

# 进入容器
shell:
	docker-compose exec patent-agent sh

第六部分：生产环境部署清单与监控（30分钟）

6.1 生产环境部署清单

类别 检查项 完成情况 备注
环境配置 使用环境变量配置所有参数 □ 避免硬编码
安全性 API密钥安全存储（如K8s Secrets） □
可观测性 集成日志聚合（如ELK/Loki） □
可观测性 添加指标监控（如Prometheus） □
可观测性 分布式追踪（如Jaeger） □
高可用 多副本部署 □
高可用 负载均衡 □
数据持久化 Session存储备份策略 □
错误处理 全局panic恢复机制 □
限流熔断 配置合理的限流策略 □
健康检查 配置健康检查和就绪检查 □

6.2 性能指标监控示例

创建pkg/monitoring/metrics.go：

package monitoring

import (
    "sync"
    "time"
    
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

// Metrics 监控指标
type Metrics struct {
    // Agent执行时间
    AgentExecutionTime *prometheus.HistogramVec
    // Agent调用次数
    AgentCallsTotal *prometheus.CounterVec
    // Agent错误次数
    AgentErrorsTotal *prometheus.CounterVec
    // 工具调用次数
    ToolCallsTotal *prometheus.CounterVec
    // Session数量
    ActiveSessions prometheus.Gauge
    // Token消耗（如果从模型获取）
    TokenUsage *prometheus.CounterVec
    
    once sync.Once
}

var metrics *Metrics

// GetMetrics 获取单例指标
func GetMetrics() *Metrics {
    if metrics == nil {
        metrics = &Metrics{
            AgentExecutionTime: promauto.NewHistogramVec(
                prometheus.HistogramOpts{
                    Name:    "agent_execution_seconds",
                    Help:    "Agent执行时间（秒）",
                    Buckets: prometheus.DefBuckets,
                },
                []string{"agent_name"},
            ),
            AgentCallsTotal: promauto.NewCounterVec(
                prometheus.CounterOpts{
                    Name: "agent_calls_total",
                    Help: "Agent调用总次数",
                },
                []string{"agent_name"},
            ),
            AgentErrorsTotal: promauto.NewCounterVec(
                prometheus.CounterOpts{
                    Name: "agent_errors_total",
                    Help: "Agent错误总次数",
                },
                []string{"agent_name", "error_type"},
            ),
            ToolCallsTotal: promauto.NewCounterVec(
                prometheus.CounterOpts{
                    Name: "tool_calls_total",
                    Help: "工具调用总次数",
                },
                []string{"tool_name"},
            ),
            ActiveSessions: promauto.NewGauge(
                prometheus.GaugeOpts{
                    Name: "active_sessions",
                    Help: "当前活跃Session数",
                },
            ),
            TokenUsage: promauto.NewCounterVec(
                prometheus.CounterOpts{
                    Name: "token_usage_total",
                    Help: "Token消耗总数",
                },
                []string{"model"},
            ),
        }
    }
    return metrics
}

// RecordAgentExecution 记录Agent执行
func (m *Metrics) RecordAgentExecution(agentName string, duration time.Duration) {
    m.AgentExecutionTime.WithLabelValues(agentName).Observe(duration.Seconds())
    m.AgentCallsTotal.WithLabelValues(agentName).Inc()
}

// RecordAgentError 记录Agent错误
func (m *Metrics) RecordAgentError(agentName, errorType string) {
    m.AgentErrorsTotal.WithLabelValues(agentName, errorType).Inc()
}

// RecordToolCall 记录工具调用
func (m *Metrics) RecordToolCall(toolName string) {
    m.ToolCallsTotal.WithLabelValues(toolName).Inc()
}

// SetActiveSessions 设置活跃Session数
func (m *Metrics) SetActiveSessions(count int) {
    m.ActiveSessions.Set(float64(count))
}

// RecordTokenUsage 记录Token消耗
func (m *Metrics) RecordTokenUsage(model string, tokens int) {
    m.TokenUsage.WithLabelValues(model).Add(float64(tokens))
}

第七阶段：阶段成果验收清单

恭喜你完成全部四个阶段的学习！现在，请对照最终验收清单：

任务 完成情况 备注
✅ Callback监控：实现了自定义CallbackHandler □
✅ 人机协作：实现了审批/追问机制 □
✅ 状态持久化：支持Session保存与恢复 □
✅ 性能优化：流式缓冲、重试机制 □
✅ 并发控制：速率限制 □
✅ HTTP服务：提供RESTful API □
✅ 容器化：Docker镜像和docker-compose □
✅ 监控指标：集成了Prometheus指标 □
✅ 错误处理：全局恢复机制 □
✅ 生产部署：完整的部署配置 □

最终成果：你拥有一个完整的、生产级别的多智能体专利代写系统，具备：

  • 多智能体协作的核心功能
  • 完整的可观测性（日志、监控、追踪）
  • 人机协作机制（审批、追问）
  • 状态持久化和中断恢复
  • 容器化部署能力
  • RESTful API服务

总结与展望

通过四个阶段的学习，你从一个Eino新手成长为能够构建生产级多智能体系统的开发者。在这个过程中，你不仅掌握了Eino框架的核心概念和高级特性，更重要的是，你学会了如何将一个复杂的业务场景（专利代写）拆解、设计并实现为一个可工作的智能系统。

学习的四个阶段回顾：

  1. 第一阶段：基础组件和Chain编排，实现检索关键词生成器
  2. 第二阶段：ReAct Agent和Tool开发，实现检索分析员
  3. 第三阶段：多智能体协作模式，实现完整的专利团队
  4. 第四阶段：生产级打磨，让系统真正可用

未来可能的扩展方向：

  • 更多专业Agent：添加附图生成员、翻译员等
  • 多模态支持：处理专利附图、化学结构式
  • RAG增强：集成向量数据库，提供更准确的检索
  • 联邦学习：保护企业数据隐私的前提下持续优化
  • 更复杂的审批流程：多级审批、并行审批

感谢你完成了整个学习旅程！希望这个手册对你构建实际的智能体系统有所帮助。如果你在实际开发中遇到问题，欢迎查阅Eino官方文档或参与社区讨论。

最后资源汇总：

祝你开发顺利！🚀