26可观测性、调试与部署

// Handler is the unified callback handler interface. Implement all five
// methods (OnStart, OnEnd, OnError, OnStartWithStreamInput,
// OnEndWithStreamOutput) or use [NewHandlerBuilder] to set only the timings
// you care about.
//
// Each method receives the context returned by the previous timing of the
// SAME handler, which lets a single handler pass state between its OnStart
// and OnEnd calls via context.WithValue. There is NO guaranteed execution
// order between DIFFERENT handlers, and the context chain does not flow
// from one handler to the next — do not rely on handler ordering.
//
// Implement [TimingChecker] (the Needed method) on your handler so the
// framework can skip timings you have not registered; this avoids unnecessary
// stream copies and goroutine allocations on every component invocation.
//
// Stream handlers (OnStartWithStreamInput, OnEndWithStreamOutput) receive a
// [*schema.StreamReader] that has already been copied; they MUST close their
// copy after reading. If any handler's copy is not closed, the original stream
// cannot be freed, causing a goroutine/memory leak for the entire pipeline.
//
// Important: do NOT mutate the Input or Output values. All downstream nodes
// and handlers share the same pointer (direct assignment, not a deep copy).
// Mutations cause data races in concurrent graph execution.
type Handler = callbacks.Handler

type Handler interface {
    OnStart(ctx, info *RunInfo, input CallbackInput) context.Context
    OnEnd(ctx, info *RunInfo, output CallbackOutput) context.Context
    OnError(ctx, info *RunInfo, err error) context.Context
    OnStartWithStreamInput(ctx, info *RunInfo, input *schema.StreamReader[CallbackInput]) context.Context
    OnEndWithStreamOutput(ctx, info *RunInfo, output *schema.StreamReader[CallbackOutput]) context.Context
}

type RunInfo struct {
  // Name is the graph node name for display purposes, not unique.
  // Passed from compose.WithNodeName().
  Name      string
  Type      string
  Component components.Component
}

// OnStart invokes the OnStart timing for all registered handlers in the
// context. This is called by component implementations that manage their own
// callbacks (i.e. implement [components.Checker] and return true from
// IsCallbacksEnabled). The returned context must be propagated to subsequent
// OnEnd/OnError calls so handlers can correlate start and end events.
//
// Handlers are invoked in reverse registration order (last registered = first
// called) to match the middleware wrapping convention.
//
// Example — typical usage inside a component's Generate method:
//
//  func (m *myChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.Message, error) {
//      ctx = callbacks.OnStart(ctx, &model.CallbackInput{Messages: input})
//      resp, err := m.doGenerate(ctx, input, opts...)
//      if err != nil {
//          callbacks.OnError(ctx, err)
//          return nil, err
//      }
//      callbacks.OnEnd(ctx, &model.CallbackOutput{Message: resp})
//      return resp, nil
//  }
func OnStart[T any](ctx context.Context, input T) context.Context {
  ctx, _ = callbacks.On(ctx, input, callbacks.OnStartHandle[T], TimingOnStart, true)

  return ctx
}

func NewLangfuseHandler(cfg *Config) (handler *CallbackHandler, flusher func()) {
  var langfuseOpts []langfuse.Option
  if cfg.Threads > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithThreads(cfg.Threads))
  }
  if cfg.Timeout > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithTimeout(cfg.Timeout))
  }
  if cfg.MaxTaskQueueSize > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithMaxTaskQueueSize(cfg.MaxTaskQueueSize))
  }
  if cfg.MaxEventSizeBytes > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithMaxEventSizeBytes(cfg.MaxEventSizeBytes))
  }
  if cfg.FlushAt > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithFlushAt(cfg.FlushAt))
  }
  if cfg.FlushInterval > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithFlushInterval(cfg.FlushInterval))
  }
  if cfg.SampleRate > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithSampleRate(cfg.SampleRate))
  }
  if len(cfg.LogMessage) > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithLogMessage(cfg.LogMessage))
  }
  if cfg.MaskFunc != nil {
    langfuseOpts = append(langfuseOpts, langfuse.WithMaskFunc(cfg.MaskFunc))
  }
  if cfg.MaxRetry > 0 {
    langfuseOpts = append(langfuseOpts, langfuse.WithMaxRetry(cfg.MaxRetry))
  }

  cli := langfuse.NewLangfuse(
    cfg.Host,
    cfg.PublicKey,
    cfg.SecretKey,
    langfuseOpts...,
  )

  return &CallbackHandler{
    cli: cli,

    name:      cfg.Name,
    userID:    cfg.UserID,
    sessionID: cfg.SessionID,
    release:   cfg.Release,
    tags:      cfg.Tags,
// … 省略 3 行；完整声明 L128–178，点击上方「浏览完整文件」

type CallbackHandler struct {
  cli langfuse.Langfuse

  name      string
  userID    string
  sessionID string
  release   string
  tags      []string
  public    bool
}

type langfuseStateKey struct{}

type GraphCompileCallback interface {
    OnFinish(ctx context.Context, info *GraphInfo)
}

// GraphInfo the info which end users pass in when they are compiling a graph.
// it is used in compile callback for user to get the node info and instance.
// you may need all details info of the graph for observation.
type GraphInfo struct {
  CompileOptions        []GraphCompileOption
  Nodes                 map[string]GraphNodeInfo // node key -> node info
  Edges                 map[string][]string      // edge start node key -> edge end node key, control edges
  DataEdges             map[string][]string
  Branches              map[string][]GraphBranch // branch start node key -> branch
  InputType, OutputType reflect.Type
  Name                  string

  NewGraphOptions []NewGraphOption
  GenStateFn      func(context.Context) any
}

// OnFinish is the compile callback entrypoint invoked by Eino after graph compilation.
// It reads the compile-time GraphInfo and writes a complete Mermaid diagram to the writer.
func (m *MermaidGenerator) OnFinish(_ context.Context, info *compose.GraphInfo) {
  m.generate(info)
}

// generate orchestrates diagram construction by delegating to renderGraph.
// The top-level direction is TD (top-down) for readability and consistency.
func (m *MermaidGenerator) generate(info *compose.GraphInfo) {
  isWorkflow := m.workflowStyle
  if !isWorkflow {
    if len(info.Edges) > len(info.DataEdges) {
      isWorkflow = true
    }

    if !isWorkflow {
      for from, edges := range info.Edges {
        dataEdges, ok := info.DataEdges[from]
        if !ok {
          isWorkflow = true
          break
        }

        if len(edges) != len(dataEdges) {
          isWorkflow = true
          break
        }

        for i := range edges {
          edge := edges[i]
          found := false
          for _, dEdge := range dataEdges {
            if dEdge == edge {
              found = true
              break
            }
          }
          if !found {
            isWorkflow = true
            break
          }
        }
      }
    }
  }

  sb := &strings.Builder{}
  sb.WriteString("graph TD\n")
  m.renderGraph(sb, info, "", 1, isWorkflow)
  if m.w != nil && !m.autoWrite {
    _, _ = fmt.Fprint(m.w, sb.String())
    return
  }

// … 省略 26 行；完整声明 L78–151，点击上方「浏览完整文件」

// 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，点击上方「浏览完整文件」

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)
  }
}