27综合大项目（生产级 Agent 应用）

func main() {
  ctx := context.Background()

  // setup cozeloop tracing (optional)
  // COZELOOP_WORKSPACE_ID=your workspace id
  // COZELOOP_API_TOKEN=your token
  cozeloopApiToken := os.Getenv("COZELOOP_API_TOKEN")
  cozeloopWorkspaceID := os.Getenv("COZELOOP_WORKSPACE_ID")
  if cozeloopApiToken != "" && cozeloopWorkspaceID != "" {
    client, err := cozeloop.NewClient(
      cozeloop.WithAPIToken(cozeloopApiToken),
      cozeloop.WithWorkspaceID(cozeloopWorkspaceID),
    )
    if err != nil {
      log.Fatalf("cozeloop.NewClient failed: %v", err)
    }
    defer func() {
      time.Sleep(5 * time.Second)
      client.Close(ctx)
    }()
    callbacks.AppendGlobalHandlers(clc.NewLoopHandler(client))
  }

  switch msgops.KindFromEnv() {
  case msgops.KindAgentic:
    runTyped[*schema.AgenticMessage](ctx)
  default:
    runTyped[*schema.Message](ctx)
  }
}

func runTyped[M adk.MessageType](ctx context.Context) {
  agent, err := buildAgentTyped[M](ctx)
  if err != nil {
    log.Fatalf("failed to build agent: %v", err)
  }

  checkpointStore := adkstore.NewInMemoryStore()

  sessionDir := msgops.DefaultSessionDir(msgops.KindOf[M]())
  log.Printf("message kind: %s", msgops.KindOf[M]())
  log.Printf("session dir: %s", sessionDir)

  workspaceDir := os.Getenv("WORKSPACE_DIR")
  if workspaceDir == "" {
    workspaceDir = "./data/workspace"
  }

  store, err := mem.NewStore[M](sessionDir)
  if err != nil {
    log.Fatalf("failed to create session store: %v", err)
  }

  port := os.Getenv("PORT")
  if port == "" {
    port = "8080"
  }

  projectRoot := os.Getenv("PROJECT_ROOT")
  if projectRoot == "" {
    // Default: the directory from which the binary is run.
    // Override with PROJECT_ROOT=/path/to/repo to give the agent full codebase access.
    if cwd, err := os.Getwd(); err == nil {
      projectRoot = cwd
    }
  }
  if abs, err := filepath.Abs(projectRoot); err == nil {
    projectRoot = abs
  }
  log.Printf("project root: %s", projectRoot)

  // EXAMPLES_DIR points to the root of the eino-examples repository.
  // Defaults to PROJECT_ROOT/examples if that directory exists, otherwise PROJECT_ROOT.
  examplesDir := os.Getenv("EXAMPLES_DIR")
  if examplesDir == "" {
    candidate := filepath.Join(projectRoot, "examples")
    if fi, err := os.Stat(candidate); err == nil && fi.IsDir() {
      examplesDir = candidate
    } else {
// … 省略 21 行；完整声明 L69–137，点击上方「浏览完整文件」

// Spin starts the HTTP server (blocking).
func (s *Server[M]) Spin() {
  h := hserver.Default(hserver.WithHostPorts(":" + s.cfg.Port))

  h.GET("/", func(ctx context.Context, c *app.RequestContext) {
    data, err := os.ReadFile("static/index.html")
    if err != nil {
      c.JSON(consts.StatusNotFound, map[string]string{"error": "index.html not found"})
      return
    }
    c.Data(consts.StatusOK, "text/html; charset=utf-8", data)
  })

  h.POST("/sessions", func(ctx context.Context, c *app.RequestContext) {
    id := uuid.New().String()
    if _, err := s.cfg.Store.GetOrCreate(id); err != nil {
      c.JSON(consts.StatusInternalServerError, map[string]string{"error": err.Error()})
      return
    }
    c.JSON(consts.StatusOK, map[string]string{"id": id})
  })

  h.GET("/sessions", func(ctx context.Context, c *app.RequestContext) {
    metas, err := s.cfg.Store.List()
    if err != nil {
      c.JSON(consts.StatusInternalServerError, map[string]string{"error": err.Error()})
      return
    }
    if metas == nil {
      metas = []mem.SessionMeta{}
    }
    c.JSON(consts.StatusOK, metas)
  })

  h.DELETE("/sessions/:id", func(ctx context.Context, c *app.RequestContext) {
    id := c.Param("id")
    // Stop any running loop for this session.
    ts := s.getTurnState(id)
    ts.mu.Lock()
    if ts.loop != nil {
      ts.loop.Stop(adk.WithImmediate())
      ts.loop = nil
    }
    ts.mu.Unlock()

    if err := s.cfg.Store.Delete(id); err != nil {
      c.JSON(consts.StatusInternalServerError, map[string]string{"error": err.Error()})
      return
// … 省略 27 行；完整声明 L181–255，点击上方「浏览完整文件」

var handlers []adk.TypedChatModelAgentMiddleware[M]
if skillsDir, ok := resolveSkillsDir(); ok {
    // 可选:文件系统技能中间件
    handlers = append(handlers, skillMiddleware)
}
handlers = append(handlers,
    newApprovalMiddleware[M](),          // 审批闸门(下面详解)
    helpers.NewSafeToolMiddleware[M](),  // 把工具报错转成字符串而非中断
)

cfg := &deep.TypedConfig[M]{
    Name:         "ChatWithEinoAgent",
    ChatModel:    cm,
    Backend:      backend,
    MaxIteration: 50,
    Handlers:     handlers,                       // 三层中间件
    ToolsConfig:  /* Tools: []tool.BaseTool{ragTool} */,
}
helpers.ApplyMessageModelRetry(cfg)  // 再叠一层:模型限流自动重试
return deep.NewTyped[M](ctx, cfg)

// ApplyMessageModelRetry enables model-call retries for transient rate-limit
// errors.
func ApplyMessageModelRetry[M adk.MessageType](cfg *deep.TypedConfig[M]) {
  cfg.ModelRetryConfig = &adk.TypedModelRetryConfig[M]{
    MaxRetries: 5,
    IsRetryAble: func(_ context.Context, err error) bool {
      return strings.Contains(err.Error(), "429") ||
        strings.Contains(err.Error(), "Too Many Requests") ||
        strings.Contains(err.Error(), "qpm limit")
    },
  }
}

// NewSafeToolMiddleware converts tool errors into error-message strings so that
// a non-zero exit code or mid-stream failure is returned to the model as a
// readable tool result instead of aborting the agent pipeline.
func NewSafeToolMiddleware[M adk.MessageType]() adk.TypedChatModelAgentMiddleware[M] {
  return &safeToolMiddleware[M]{
    TypedBaseChatModelAgentMiddleware: &adk.TypedBaseChatModelAgentMiddleware[M]{},
  }
}

func (m *safeToolMiddleware[M]) WrapInvokableToolCall(
  _ context.Context,
  endpoint adk.InvokableToolCallEndpoint,
  _ *adk.ToolContext,
) (adk.InvokableToolCallEndpoint, error) {
  return func(ctx context.Context, args string, opts ...tool.Option) (string, error) {
    result, err := endpoint(ctx, args, opts...)
    if err != nil {
      if _, ok := compose.IsInterruptRerunError(err); ok {
        return "", err
      }
      return fmt.Sprintf("[tool error] %v", err), nil
    }
    return result, nil
  }, nil
}

func (m *approvalMiddleware[M]) WrapInvokableToolCall(...) (adk.InvokableToolCallEndpoint, error) {
    if tCtx.Name != "answer_from_document" {
        return endpoint, nil        // 其他工具原样放行
    }
    return func(ctx context.Context, args string, opts ...tool.Option) (string, error) {
        wasInterrupted, _, storedArgs := tool.GetInterruptState[string](ctx)
        if !wasInterrupted {
            // 第一次进来:还没审批 → 中断,把工具参数存进 checkpoint
            return "", tool.StatefulInterrupt(ctx, &commontool.ApprovalInfo{
                ToolName: tCtx.Name, ArgumentsInJSON: args,
            }, args)
        }
        // 恢复回来:读出用户的审批结果
        isTarget, hasData, data := tool.GetResumeContext[*commontool.ApprovalResult](ctx)
        if isTarget && hasData {
            if data.Approved {
                return endpoint(ctx, storedArgs, opts...)   // 批准 → 真的执行工具
            }
            return fmt.Sprintf("tool '%s' disapproved", tCtx.Name), nil  // 拒绝 → 返回一句话
        }
        // ...
    }, nil
}

cfg := adk.TurnLoopConfig[*ChatItem, M]{
    GenInput:      s.makeGenInput(sess, sessionID),      // 从会话历史造 agent 输入
    PrepareAgent:  s.makePrepareAgent(),                 // 提供 agent
    OnAgentEvents: s.makeOnAgentEvents(sess, sessionID), // 把 agent 事件桥到 SSE
}
if s.cfg.CheckPointStore != nil {
    cfg.Store = s.cfg.CheckPointStore
    cfg.CheckpointID = sessionID
    cfg.GenResume = s.makeGenResume()                    // 恢复时把审批结果喂回去
}
return adk.NewTurnLoop(cfg)

// handleChat handles a new chat message. It creates or reuses a TurnLoop for the session.
// If a loop is already running (busy), it pushes with preempt to cancel the current turn.
func (s *Server[M]) handleChat(ctx context.Context, c *app.RequestContext) {
  id := c.Param("id")

  body, _ := c.Body()
  var req chatRequest
  if err := json.Unmarshal(body, &req); err != nil || req.Message == "" {
    c.JSON(consts.StatusBadRequest, map[string]string{"error": "message is required"})
    return
  }

  log.Printf("[chat] session=%s msg=%q", id, req.Message)

  sess, err := s.cfg.Store.GetOrCreate(id)
  if err != nil {
    c.JSON(consts.StatusInternalServerError, map[string]string{"error": err.Error()})
    return
  }

  item := &ChatItem{Query: req.Message}

  ts := s.getTurnState(id)

  // Each handler gets its own local iterReady channel reference and a
  // handlerDone channel. This avoids races when multiple preempts replace
  // the channels on ts concurrently.
  var localIterReady chan iterEnvelope[M]
  var localHandlerDone chan struct{}

  ts.mu.Lock()
  if ts.loop != nil {
    // Loop exists — try to push with preempt (AfterToolCalls).
    loop := ts.loop
    log.Printf("[chat] session=%s preempting current turn", id)
    // Signal any previous handler waiting on iterReady to bail.
    if ts.handlerDone != nil {
      close(ts.handlerDone)
    }
    ts.iterReady = make(chan iterEnvelope[M], 1)
    ts.iterDone = make(chan iterResult[M], 1)
    ts.handlerDone = make(chan struct{})
    localIterReady = ts.iterReady
    localHandlerDone = ts.handlerDone
    ts.mu.Unlock()
    ok, _ := loop.Push(item, adk.WithPreempt[*ChatItem, M](adk.AfterToolCalls))
    if !ok {
      // Loop already stopped (e.g. error on previous turn) — create new one.
// … 省略 95 行；完整声明 L281–423，点击上方「浏览完整文件」

// makeGenInput returns the GenInput callback. It builds agent messages from
// session history + workspace context.
func (s *Server[M]) makeGenInput(sess *mem.Session[M], sessionID string) func(ctx context.Context, loop *adk.TurnLoop[*ChatItem, M], items []*ChatItem) (*adk.GenInputResult[*ChatItem, M], error) {
  return func(ctx context.Context, loop *adk.TurnLoop[*ChatItem, M], items []*ChatItem) (*adk.GenInputResult[*ChatItem, M], error) {
    // Find the first item with a query.
    var consumed []*ChatItem
    var remaining []*ChatItem
    var queryItem *ChatItem
    for _, item := range items {
      if queryItem == nil && item.Query != "" {
        queryItem = item
        consumed = append(consumed, item)
      } else {
        remaining = append(remaining, item)
      }
    }
    if queryItem == nil {
      // No query items — stop the loop.
      loop.Stop(adk.WithStopCause("no query items"))
      return &adk.GenInputResult[*ChatItem, M]{
        Input:     &adk.TypedAgentInput[M]{Messages: []M{msgops.NewUser[M]("done")}},
        Remaining: items,
      }, nil
    }

    // Persist the user message NOW — GenInput fires only after any previous
    // turn's OnAgentEvents has finished persisting its intermediates, so the
    // session history order is guaranteed correct.
    userMsg := msgops.NewUser[M](queryItem.Query)
    if appendErr := sess.Append(userMsg); appendErr != nil {
      log.Printf("warn: failed to persist user message: %v", appendErr)
    }

    history := sess.GetMessages()
    runMessages := s.buildRunMessages(sessionID, history)

    log.Printf("[genInput] session=%s query=%q messages=%d", sessionID, queryItem.Query, len(runMessages))

    return &adk.GenInputResult[*ChatItem, M]{
      Input: &adk.TypedAgentInput[M]{
        Messages:        runMessages,
        EnableStreaming: true,
      },
      Consumed:  consumed,
      Remaining: remaining,
    }, nil
  }
}

// makeOnAgentEvents returns the OnAgentEvents callback — the bridge between
// the TurnLoop and the HTTP handler.
func (s *Server[M]) makeOnAgentEvents(sess *mem.Session[M], sessionID string) func(ctx context.Context, tc *adk.TurnContext[*ChatItem, M], events *adk.AsyncIterator[*adk.TypedAgentEvent[M]]) error {
  return func(ctx context.Context, tc *adk.TurnContext[*ChatItem, M], events *adk.AsyncIterator[*adk.TypedAgentEvent[M]]) error {
    ts := s.getTurnState(sessionID)

    history := sess.GetMessages()

    // Snapshot bridge channels under lock to avoid races with handleChat
    // which may recreate them for a preempt.
    ts.mu.Lock()
    ready := ts.iterReady
    done := ts.iterDone
    ts.mu.Unlock()

    // Send the iterator to the HTTP handler. Include the done channel
    // so the handler replies to THIS invocation, not a future one.
    select {
    case ready <- iterEnvelope[M]{events: events, history: history, done: done}:
    case <-ctx.Done():
      return ctx.Err()
    }

    // Wait for the HTTP handler to finish draining. Also select on ctx.Done
    // to avoid hanging when a preempt supersedes the handler — in that case
    // the old handler bails via handlerDone and nobody sends to our done channel.
    var result iterResult[M]
    select {
    case result = <-done:
    case <-ctx.Done():
      return ctx.Err()
    }

    // Persist all intermediate messages (assistant text+tool calls, tool results).
    // The intermediates already include the final assistant text message if any,
    // so we don't need to persist lastContent separately.
    for _, msg := range result.intermediates {
      if appendErr := sess.Append(msg); appendErr != nil {
        log.Printf("warn: failed to persist intermediate message: %v", appendErr)
      }
    }
    if result.interruptID != "" {
      sess.SetPendingInterruptID(result.interruptID)
      sess.SetMsgIdx(result.msgIdx)
      return errInterrupted
    }
    return result.err
  }
// … 省略 1 行；完整声明 L641–689，点击上方「浏览完整文件」

// handleApprove resumes an interrupted agent run with the user's approval decision.
// Creates a new TurnLoop with checkpoint/resume to continue from the interrupt.
func (s *Server[M]) handleApprove(ctx context.Context, c *app.RequestContext) {
  id := c.Param("id")

  sess, err := s.cfg.Store.GetOrCreate(id)
  if err != nil {
    c.JSON(consts.StatusInternalServerError, map[string]string{"error": err.Error()})
    return
  }

  interruptID := sess.GetPendingInterruptID()
  if interruptID == "" {
    c.JSON(consts.StatusBadRequest, map[string]string{"error": "no pending interrupt for this session"})
    return
  }

  body, _ := c.Body()
  var req approveRequest
  if err := json.Unmarshal(body, &req); err != nil {
    c.JSON(consts.StatusBadRequest, map[string]string{"error": "invalid request body"})
    return
  }

  var reason *string
  if req.Reason != "" {
    reason = &req.Reason
  }
  result := &commontool.ApprovalResult{Approved: req.Approved, DisapproveReason: reason}

  // Clear the pending interrupt so a double-approve returns 400.
  sess.SetPendingInterruptID("")

  log.Printf("[approve] session=%s interruptID=%s approved=%v", id, interruptID, req.Approved)

  // Create a new loop with checkpoint resume.
  ts := s.getTurnState(id)
  ts.mu.Lock()
  // Clear any old loop.
  if ts.loop != nil {
    ts.loop.Stop(adk.WithImmediate())
  }
  // Signal any previous handler to bail.
  if ts.handlerDone != nil {
    close(ts.handlerDone)
  }
  loop := s.newLoop(sess, id, true)
  ts.loop = loop
// … 省略 70 行；完整声明 L425–542，点击上方「浏览完整文件」

// makeGenResume returns the GenResume callback for interrupt/resume.
func (s *Server[M]) makeGenResume() func(ctx context.Context, loop *adk.TurnLoop[*ChatItem, M], canceledItems, unhandledItems, newItems []*ChatItem) (*adk.GenResumeResult[*ChatItem, M], error) {
  return func(ctx context.Context, loop *adk.TurnLoop[*ChatItem, M], canceledItems, unhandledItems, newItems []*ChatItem) (*adk.GenResumeResult[*ChatItem, M], error) {
    // Find the approval item in newItems.
    var approvalItem *ChatItem
    for _, item := range newItems {
      if item.ApprovalResult != nil {
        approvalItem = item
        break
      }
    }
    if approvalItem == nil {
      return nil, errors.New("no approval item found for resume")
    }

    return &adk.GenResumeResult[*ChatItem, M]{
      ResumeParams: &adk.ResumeParams{
        Targets: map[string]any{approvalItem.InterruptID: approvalItem.ApprovalResult},
      },
      Consumed:  canceledItems,
      Remaining: unhandledItems,
    }, nil
  }
}

// handleAbort immediately stops the current TurnLoop for a session.
func (s *Server[M]) handleAbort(_ context.Context, c *app.RequestContext) {
  id := c.Param("id")

  ts := s.getTurnState(id)
  ts.mu.Lock()
  loop := ts.loop
  ts.loop = nil
  ts.mu.Unlock()

  if loop == nil {
    c.JSON(consts.StatusOK, map[string]string{"status": "no active loop"})
    return
  }

  log.Printf("[abort] session=%s stopping loop immediately", id)
  loop.Stop(adk.WithImmediate())
  loop.Wait()
  log.Printf("[abort] session=%s loop stopped", id)

  c.JSON(consts.StatusOK, map[string]string{"status": "aborted"})
}

// Append adds a message to memory and persists it to disk.
func (s *Session[M]) Append(msg M) error {
  s.mu.Lock()
  defer s.mu.Unlock()

  msg = msgops.NormalizeForSession(msg)
  s.messages = append(s.messages, msg)

  data, err := json.Marshal(msg)
  if err != nil {
    return err
  }

  f, err := os.OpenFile(s.filePath, os.O_APPEND|os.O_WRONLY, 0o644)
  if err != nil {
    return err
  }
  defer f.Close()

  _, err = fmt.Fprintf(f, "%s\n", data)
  return err
}

// GetMessages returns a snapshot of all messages.
func (s *Session[M]) GetMessages() []M {
  s.mu.Lock()
  defer s.mu.Unlock()

  result := make([]M, len(s.messages))
  copy(result, s.messages)
  return result
}

func main() {
  var instruction string
  flag.StringVar(&instruction, "instruction", "You are a helpful assistant.", "")
  flag.Parse()

  query := strings.TrimSpace(strings.Join(flag.Args(), " "))
  if query == "" {
    _, _ = fmt.Fprintln(os.Stderr, "usage: go run ./cmd/ch01 -- \"your question\"")
    os.Exit(2)
  }

  ctx := context.Background()
  switch msgops.KindFromEnv() {
  case msgops.KindAgentic:
    runTyped[*schema.AgenticMessage](ctx, instruction, query)
  default:
    runTyped[*schema.Message](ctx, instruction, query)
  }
}