diff --git a/.env.example b/.env.example index b9947b479..142e946db 100644 --- a/.env.example +++ b/.env.example @@ -25,3 +25,26 @@ TZ=Asia/Tokyo # PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT=otel-collector:4317 # PICOCLAW_OBSERVABILITY_INSECURE=true # PICOCLAW_OBSERVABILITY_SAMPLE_RATIO=0.1 + +# Langfuse via OTLP (optional) +# PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED=true +# PICOCLAW_OBSERVABILITY_LANGFUSE_HOST=http://langfuse-web:3000 +# PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY=pk-lf-xxxx +# PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY=sk-lf-xxxx + +# Optional host port overrides for local demo +# PROMETHEUS_PORT=9090 +# LANGFUSE_MINIO_API_PORT=9092 +# LANGFUSE_MINIO_CONSOLE_PORT=9093 + +# Langfuse self-hosted backing services (optional overrides) +# POSTGRES_USER=postgres +# POSTGRES_PASSWORD=postgres +# POSTGRES_DB=postgres +# REDIS_AUTH=myredissecret +# CLICKHOUSE_USER=clickhouse +# CLICKHOUSE_PASSWORD=clickhouse +# CLICKHOUSE_MIGRATION_URL=clickhouse://langfuse-clickhouse:9000 +# CLICKHOUSE_CLUSTER_ENABLED=false +# MINIO_ROOT_USER=minio +# MINIO_ROOT_PASSWORD=miniosecret diff --git a/cmd/picoclaw/cmd_status.go b/cmd/picoclaw/cmd_status.go index 2b9aead24..73828d6a3 100644 --- a/cmd/picoclaw/cmd_status.go +++ b/cmd/picoclaw/cmd_status.go @@ -28,16 +28,16 @@ func statusCmd() { fmt.Println() if _, err := os.Stat(configPath); err == nil { - fmt.Println("Config:", configPath, "✓") + fmt.Println("Config:", configPath, "ok") } else { - fmt.Println("Config:", configPath, "✗") + fmt.Println("Config:", configPath, "missing") } workspace := cfg.WorkspacePath() if _, err := os.Stat(workspace); err == nil { - fmt.Println("Workspace:", workspace, "✓") + fmt.Println("Workspace:", workspace, "ok") } else { - fmt.Println("Workspace:", workspace, "✗") + fmt.Println("Workspace:", workspace, "missing") } if _, err := os.Stat(configPath); err == nil { @@ -59,7 +59,7 @@ func statusCmd() { status := func(enabled bool) string { if enabled { - return "✓" + return "ok" } return "not set" } @@ -75,17 +75,22 @@ func statusCmd() { fmt.Println("VolcEngine API:", status(hasVolcEngine)) fmt.Println("Nvidia API:", status(hasNvidia)) if hasVLLM { - fmt.Printf("vLLM/Local: ✓ %s\n", cfg.Providers.VLLM.APIBase) + fmt.Printf("vLLM/Local: ok %s\n", cfg.Providers.VLLM.APIBase) } else { fmt.Println("vLLM/Local: not set") } if hasOllama { - fmt.Printf("Ollama: ✓ %s\n", cfg.Providers.Ollama.APIBase) + fmt.Printf("Ollama: ok %s\n", cfg.Providers.Ollama.APIBase) } else { fmt.Println("Ollama: not set") } if cfg.Observability.Enabled { - fmt.Printf("Observability: ✓ OTLP=%s sample_ratio=%.2f\n", cfg.Observability.OTLPEndpoint, cfg.Observability.SampleRatio) + fmt.Printf("Observability: ok OTLP=%s sample_ratio=%.2f\n", cfg.Observability.OTLPEndpoint, cfg.Observability.SampleRatio) + if cfg.Observability.Langfuse.IsConfigured() { + fmt.Printf("Langfuse: ok host=%s\n", cfg.Observability.Langfuse.Host) + } else { + fmt.Println("Langfuse: not set") + } } else { fmt.Println("Observability: not set") } diff --git a/docker-compose.yml b/docker-compose.yml index e0c42e52d..9a6ed245d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -27,6 +27,10 @@ services: - PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT=${PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT:-otel-collector:4317} - PICOCLAW_OBSERVABILITY_INSECURE=true - PICOCLAW_OBSERVABILITY_SAMPLE_RATIO=1.0 + - PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED=${PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED:-false} + - PICOCLAW_OBSERVABILITY_LANGFUSE_HOST=${PICOCLAW_OBSERVABILITY_LANGFUSE_HOST:-http://langfuse-web:3000} + - PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY=${PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY:-} + - PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY=${PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY:-} - PICOCLAW_AGENTS_DEFAULTS_PROVIDER=${PICOCLAW_AGENTS_DEFAULTS_PROVIDER:-} - PICOCLAW_AGENTS_DEFAULTS_MODEL=${PICOCLAW_AGENTS_DEFAULTS_MODEL:-} - PICOCLAW_PROVIDERS_OPENAI_API_KEY=${PICOCLAW_PROVIDERS_OPENAI_API_KEY:-} @@ -64,6 +68,10 @@ services: - PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT=${PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT:-otel-collector:4317} - PICOCLAW_OBSERVABILITY_INSECURE=true - PICOCLAW_OBSERVABILITY_SAMPLE_RATIO=1.0 + - PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED=${PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED:-false} + - PICOCLAW_OBSERVABILITY_LANGFUSE_HOST=${PICOCLAW_OBSERVABILITY_LANGFUSE_HOST:-http://langfuse-web:3000} + - PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY=${PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY:-} + - PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY=${PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY:-} - PICOCLAW_AGENTS_DEFAULTS_PROVIDER=${PICOCLAW_AGENTS_DEFAULTS_PROVIDER:-} - PICOCLAW_AGENTS_DEFAULTS_MODEL=${PICOCLAW_AGENTS_DEFAULTS_MODEL:-} - PICOCLAW_PROVIDERS_OPENAI_API_KEY=${PICOCLAW_PROVIDERS_OPENAI_API_KEY:-} @@ -101,7 +109,7 @@ services: - ./observability/prometheus.yml:/etc/prometheus/prometheus.yml:ro - prometheus-data:/prometheus ports: - - "9090:9090" + - "${PROMETHEUS_PORT:-9090}:9090" depends_on: - otel-collector @@ -143,8 +151,173 @@ services: - prometheus - loki + # Local Langfuse self-hosted demo stack + # Reference: https://langfuse.com/self-hosting/deployment/docker-compose + langfuse-worker: + image: docker.io/langfuse/langfuse-worker:3 + container_name: langfuse-worker + profiles: [observability] + restart: unless-stopped + depends_on: + langfuse-postgres: + condition: service_healthy + langfuse-redis: + condition: service_healthy + langfuse-clickhouse: + condition: service_healthy + langfuse-minio: + condition: service_healthy + environment: + NEXTAUTH_URL: ${LANGFUSE_NEXTAUTH_URL:-http://localhost:3001} + DATABASE_URL: ${LANGFUSE_DATABASE_URL:-postgresql://postgres:postgres@langfuse-postgres:5432/postgres} + SALT: ${LANGFUSE_SALT:-change-me-langfuse-salt} + ENCRYPTION_KEY: ${LANGFUSE_ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000} + TELEMETRY_ENABLED: ${LANGFUSE_TELEMETRY_ENABLED:-false} + CLICKHOUSE_MIGRATION_URL: ${CLICKHOUSE_MIGRATION_URL:-clickhouse://langfuse-clickhouse:9000} + CLICKHOUSE_URL: http://langfuse-clickhouse:8123 + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} + CLICKHOUSE_CLUSTER_ENABLED: ${CLICKHOUSE_CLUSTER_ENABLED:-false} + LANGFUSE_S3_EVENT_UPLOAD_BUCKET: ${LANGFUSE_S3_BUCKET:-langfuse} + LANGFUSE_S3_EVENT_UPLOAD_REGION: ${LANGFUSE_S3_REGION:-auto} + LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_ACCESS_KEY:-minio} + LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_SECRET_KEY:-miniosecret} + LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://langfuse-minio:9000 + LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true" + LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: ${LANGFUSE_S3_BUCKET:-langfuse} + LANGFUSE_S3_MEDIA_UPLOAD_REGION: ${LANGFUSE_S3_REGION:-auto} + LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_ACCESS_KEY:-minio} + LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_SECRET_KEY:-miniosecret} + LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://langfuse-minio:9000} + LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: "true" + REDIS_HOST: langfuse-redis + REDIS_PORT: 6379 + REDIS_AUTH: ${REDIS_AUTH:-myredissecret} + + langfuse-web: + image: docker.io/langfuse/langfuse:3 + container_name: langfuse-web + profiles: [observability] + restart: unless-stopped + depends_on: + langfuse-postgres: + condition: service_healthy + langfuse-redis: + condition: service_healthy + langfuse-clickhouse: + condition: service_healthy + langfuse-minio: + condition: service_healthy + ports: + - "3001:3000" + environment: + NEXTAUTH_URL: ${LANGFUSE_NEXTAUTH_URL:-http://localhost:3001} + NEXTAUTH_SECRET: ${LANGFUSE_NEXTAUTH_SECRET:-change-me-langfuse-nextauth-secret} + DATABASE_URL: ${LANGFUSE_DATABASE_URL:-postgresql://postgres:postgres@langfuse-postgres:5432/postgres} + SALT: ${LANGFUSE_SALT:-change-me-langfuse-salt} + ENCRYPTION_KEY: ${LANGFUSE_ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000} + TELEMETRY_ENABLED: ${LANGFUSE_TELEMETRY_ENABLED:-false} + CLICKHOUSE_MIGRATION_URL: ${CLICKHOUSE_MIGRATION_URL:-clickhouse://langfuse-clickhouse:9000} + CLICKHOUSE_URL: http://langfuse-clickhouse:8123 + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} + CLICKHOUSE_CLUSTER_ENABLED: ${CLICKHOUSE_CLUSTER_ENABLED:-false} + LANGFUSE_S3_EVENT_UPLOAD_BUCKET: ${LANGFUSE_S3_BUCKET:-langfuse} + LANGFUSE_S3_EVENT_UPLOAD_REGION: ${LANGFUSE_S3_REGION:-auto} + LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_ACCESS_KEY:-minio} + LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_SECRET_KEY:-miniosecret} + LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://langfuse-minio:9000 + LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true" + LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: ${LANGFUSE_S3_BUCKET:-langfuse} + LANGFUSE_S3_MEDIA_UPLOAD_REGION: ${LANGFUSE_S3_REGION:-auto} + LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_ACCESS_KEY:-minio} + LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: ${LANGFUSE_S3_SECRET_KEY:-miniosecret} + LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://langfuse-minio:9000} + LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: "true" + REDIS_HOST: langfuse-redis + REDIS_PORT: 6379 + REDIS_AUTH: ${REDIS_AUTH:-myredissecret} + + langfuse-postgres: + image: postgres:17 + container_name: langfuse-postgres + profiles: [observability] + restart: unless-stopped + environment: + POSTGRES_USER: ${POSTGRES_USER:-postgres} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} + POSTGRES_DB: ${POSTGRES_DB:-postgres} + volumes: + - langfuse-postgres-data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres"] + interval: 5s + timeout: 5s + retries: 20 + + langfuse-redis: + image: redis:7 + container_name: langfuse-redis + profiles: [observability] + restart: unless-stopped + command: >- + --requirepass ${REDIS_AUTH:-myredissecret} + --maxmemory-policy noeviction + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 5s + retries: 20 + + langfuse-clickhouse: + image: docker.io/clickhouse/clickhouse-server + container_name: langfuse-clickhouse + profiles: [observability] + restart: unless-stopped + user: "101:101" + environment: + CLICKHOUSE_DB: default + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} + volumes: + - langfuse-clickhouse-data:/var/lib/clickhouse + - langfuse-clickhouse-logs:/var/log/clickhouse-server + healthcheck: + test: ["CMD-SHELL", "wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1"] + interval: 5s + timeout: 5s + retries: 20 + + langfuse-minio: + image: cgr.dev/chainguard/minio + container_name: langfuse-minio + profiles: [observability] + restart: unless-stopped + entrypoint: sh + command: + - -c + - 'mkdir -p /data/langfuse && minio server --address ":9000" --console-address ":9001" /data' + ports: + - "${LANGFUSE_MINIO_API_PORT:-9092}:9000" + - "${LANGFUSE_MINIO_CONSOLE_PORT:-9093}:9001" + environment: + MINIO_ROOT_USER: ${MINIO_ROOT_USER:-minio} + MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:-miniosecret} + volumes: + - langfuse-minio-data:/data + healthcheck: + test: ["CMD", "mc", "ready", "local"] + interval: 1s + timeout: 5s + retries: 5 + start_period: 1s + volumes: picoclaw-workspace: prometheus-data: loki-data: grafana-data: + langfuse-postgres-data: + langfuse-clickhouse-data: + langfuse-clickhouse-logs: + langfuse-minio-data: diff --git a/go.mod b/go.mod index 55d0f2a77..84963e7ae 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/tencent-connect/botgo v0.2.1 go.opentelemetry.io/otel v1.40.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.40.0 go.opentelemetry.io/otel/sdk v1.40.0 go.opentelemetry.io/otel/trace v1.40.0 golang.org/x/oauth2 v0.35.0 diff --git a/go.sum b/go.sum index 126cf8a5b..3e98e8ed3 100644 --- a/go.sum +++ b/go.sum @@ -172,6 +172,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0 h1:QKdN8ly8zEMrByybbQg go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0/go.mod h1:bTdK1nhqF76qiPoCCdyFIV+N/sRHYXYCTQc+3VCi3MI= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 h1:DvJDOPmSWQHWywQS6lKL+pb8s3gBLOZUtw4N+mavW1I= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0/go.mod h1:EtekO9DEJb4/jRyN4v4Qjc2yA7AtfCBuz2FynRUWTXs= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.40.0 h1:wVZXIWjQSeSmMoxF74LzAnpVQOAFDo3pPji9Y4SOFKc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.40.0/go.mod h1:khvBS2IggMFNwZK/6lEeHg/W57h/IX6J4URh57fuI40= go.opentelemetry.io/otel/metric v1.40.0 h1:rcZe317KPftE2rstWIBitCdVp89A2HqjkxR3c11+p9g= go.opentelemetry.io/otel/metric v1.40.0/go.mod h1:ib/crwQH7N3r5kfiBZQbwrTge743UDc7DTFVZrrXnqc= go.opentelemetry.io/otel/sdk v1.40.0 h1:KHW/jUzgo6wsPh9At46+h4upjtccTmuZCFAc9OJ71f8= diff --git a/observability/README.md b/observability/README.md index 1fed7b918..7fe50daa7 100644 --- a/observability/README.md +++ b/observability/README.md @@ -4,7 +4,26 @@ ## Start observability + gateway # docker compose --profile observability --profile gateway up -d +## Local Langfuse (self-hosted) is included in the observability profile +# Langfuse UI: http://localhost:3001 + +## Enable PicoClaw -> local Langfuse OTLP export +# export PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED=true +# export PICOCLAW_OBSERVABILITY_LANGFUSE_HOST=http://langfuse-web:3000 +# export PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY=pk-lf-xxxx +# export PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY=sk-lf-xxxx +# docker compose --profile observability --profile gateway up -d --force-recreate picoclaw-gateway + +## Smoke test (inside Linux/devcontainer shell) +# docker compose --profile observability --profile gateway exec picoclaw-gateway \ +# picoclaw agent -m "list files in workspace" + +## Verify status +# docker compose --profile observability --profile gateway exec picoclaw-gateway \ +# picoclaw status + ## URLs # Grafana: http://localhost:3000 (admin/admin) # Prometheus: http://localhost:9090 # Loki API: http://localhost:3100 +# Langfuse: http://localhost:3001 diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 80950cdcc..4cfc748f0 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -461,6 +461,24 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt finalContent = opts.DefaultResponse } + // Emit a dedicated final-response span so evaluators can target completed user-facing output + finalCtx, finalSpan := observability.Tracer("picoclaw.agent").Start(ctx, "agent.final_response") + _ = finalCtx + finalSpan.SetAttributes( + attribute.String("agent.id", agent.ID), + attribute.String("channel", opts.Channel), + attribute.String("chat_id", opts.ChatID), + attribute.String("session_key", opts.SessionKey), + attribute.String("input.value", opts.UserMessage), + attribute.String("langfuse.observation.input", opts.UserMessage), + attribute.String("output.value", finalContent), + attribute.String("langfuse.observation.output", finalContent), + attribute.Bool("llm.has_output", strings.TrimSpace(finalContent) != ""), + attribute.String("langfuse.observation.type", "generation"), + ) + finalSpan.SetStatus(codes.Ok, "final response recorded") + finalSpan.End() + // 6. Save final assistant message to session agent.Sessions.AddMessage(opts.SessionKey, "assistant", finalContent) agent.Sessions.Save(opts.SessionKey) @@ -544,7 +562,7 @@ func (al *AgentLoop) runLLMIteration( if len(agent.Candidates) > 1 && al.fallback != nil { fbResult, fbErr := al.fallback.Execute(ctx, agent.Candidates, func(ctx context.Context, provider, model string) (*providers.LLMResponse, error) { - return agent.Provider.Chat(ctx, messages, providerToolDefs, model, map[string]any{ + return al.callProviderChat(ctx, agent, messages, providerToolDefs, model, map[string]any{ "max_tokens": agent.MaxTokens, "temperature": agent.Temperature, }) @@ -560,7 +578,7 @@ func (al *AgentLoop) runLLMIteration( } return fbResult.Response, nil } - return agent.Provider.Chat(ctx, messages, providerToolDefs, agent.Model, map[string]any{ + return al.callProviderChat(ctx, agent, messages, providerToolDefs, agent.Model, map[string]any{ "max_tokens": agent.MaxTokens, "temperature": agent.Temperature, }) @@ -773,6 +791,94 @@ func (al *AgentLoop) runLLMIteration( return finalContent, iteration, nil } +func (al *AgentLoop) callProviderChat( + ctx context.Context, + agent *AgentInstance, + messages []providers.Message, + tools []providers.ToolDefinition, + model string, + options map[string]any, +) (*providers.LLMResponse, error) { + ctx, span := observability.Tracer("picoclaw.agent").Start(ctx, "agent.llm_call") + startedAt := time.Now() + modelForCost := model + if idx := strings.LastIndex(model, "/"); idx >= 0 && idx < len(model)-1 { + modelForCost = model[idx+1:] + } + providerName := "unknown" + if idx := strings.Index(model, "/"); idx > 0 { + providerName = model[:idx] + } + lastUserInput := "" + for i := len(messages) - 1; i >= 0; i-- { + if messages[i].Role == "user" { + lastUserInput = messages[i].Content + break + } + } + span.SetAttributes( + attribute.String("llm.model", model), + attribute.String("llm.model.short", modelForCost), + attribute.Int("llm.messages_count", len(messages)), + attribute.Int("llm.tools_count", len(tools)), + // Langfuse/OpenTelemetry GenAI conventions for usage & cost dashboards + attribute.String("gen_ai.operation.name", "chat"), + attribute.String("gen_ai.provider.name", providerName), + attribute.String("gen_ai.request.model", modelForCost), + attribute.String("gen_ai.response.model", modelForCost), + attribute.String("langfuse.observation.type", "generation"), + attribute.String("langfuse.generation.model", modelForCost), + ) + if lastUserInput != "" { + span.SetAttributes( + attribute.String("input.value", lastUserInput), + attribute.String("gen_ai.prompt", lastUserInput), + attribute.String("langfuse.observation.input", lastUserInput), + ) + } + if agent != nil { + span.SetAttributes(attribute.String("agent.id", agent.ID)) + } + if v, ok := options["max_tokens"].(int); ok { + span.SetAttributes(attribute.Int("llm.max_tokens", v)) + } + if v, ok := options["temperature"].(*float64); ok && v != nil { + span.SetAttributes(attribute.Float64("llm.temperature", *v)) + } + defer span.End() + + resp, err := agent.Provider.Chat(ctx, messages, tools, model, options) + span.SetAttributes(attribute.Int64("llm.duration_ms", time.Since(startedAt).Milliseconds())) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return nil, err + } + + span.SetAttributes( + attribute.String("llm.finish_reason", resp.FinishReason), + attribute.Int("llm.content_chars", len(resp.Content)), + attribute.Int("llm.tool_calls_count", len(resp.ToolCalls)), + attribute.Bool("llm.has_output", strings.TrimSpace(resp.Content) != ""), + attribute.String("output.value", resp.Content), + attribute.String("gen_ai.completion", resp.Content), + attribute.String("langfuse.observation.output", resp.Content), + ) + if resp.Usage != nil { + span.SetAttributes( + attribute.Int("llm.tokens.prompt", resp.Usage.PromptTokens), + attribute.Int("llm.tokens.completion", resp.Usage.CompletionTokens), + attribute.Int("llm.tokens.total", resp.Usage.TotalTokens), + attribute.Int("gen_ai.usage.prompt_tokens", resp.Usage.PromptTokens), + attribute.Int("gen_ai.usage.completion_tokens", resp.Usage.CompletionTokens), + attribute.Int("gen_ai.usage.total_tokens", resp.Usage.TotalTokens), + attribute.Int("gen_ai.usage.input_tokens", resp.Usage.PromptTokens), + attribute.Int("gen_ai.usage.output_tokens", resp.Usage.CompletionTokens), + ) + } + return resp, nil +} + // updateToolContexts updates the context for tools that need channel/chatID info. func (al *AgentLoop) updateToolContexts(agent *AgentInstance, channel, chatID string) { // Use ContextualTool interface instead of type assertions @@ -1000,8 +1106,9 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) { s1, s2, ) - resp, err := agent.Provider.Chat( + resp, err := al.callProviderChat( ctx, + agent, []providers.Message{{Role: "user", Content: mergePrompt}}, nil, agent.Model, @@ -1050,8 +1157,9 @@ func (al *AgentLoop) summarizeBatch( } prompt := sb.String() - response, err := agent.Provider.Chat( + response, err := al.callProviderChat( ctx, + agent, []providers.Message{{Role: "user", Content: prompt}}, nil, agent.Model, diff --git a/pkg/config/config.go b/pkg/config/config.go index 6a537ae88..53d257698 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -47,16 +47,16 @@ func (f *FlexibleStringSlice) UnmarshalJSON(data []byte) error { } type Config struct { - Agents AgentsConfig `json:"agents"` - Bindings []AgentBinding `json:"bindings,omitempty"` - Session SessionConfig `json:"session,omitempty"` - Channels ChannelsConfig `json:"channels"` - Providers ProvidersConfig `json:"providers,omitempty"` - ModelList []ModelConfig `json:"model_list"` // New model-centric provider configuration - Gateway GatewayConfig `json:"gateway"` - Tools ToolsConfig `json:"tools"` - Heartbeat HeartbeatConfig `json:"heartbeat"` - Devices DevicesConfig `json:"devices"` + Agents AgentsConfig `json:"agents"` + Bindings []AgentBinding `json:"bindings,omitempty"` + Session SessionConfig `json:"session,omitempty"` + Channels ChannelsConfig `json:"channels"` + Providers ProvidersConfig `json:"providers,omitempty"` + ModelList []ModelConfig `json:"model_list"` // New model-centric provider configuration + Gateway GatewayConfig `json:"gateway"` + Tools ToolsConfig `json:"tools"` + Heartbeat HeartbeatConfig `json:"heartbeat"` + Devices DevicesConfig `json:"devices"` Observability ObservabilityConfig `json:"observability"` } @@ -318,11 +318,23 @@ type DevicesConfig struct { } type ObservabilityConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_OBSERVABILITY_ENABLED"` - ServiceName string `json:"service_name" env:"PICOCLAW_OBSERVABILITY_SERVICE_NAME"` - OTLPEndpoint string `json:"otlp_endpoint" env:"PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT"` - Insecure bool `json:"insecure" env:"PICOCLAW_OBSERVABILITY_INSECURE"` - SampleRatio float64 `json:"sample_ratio" env:"PICOCLAW_OBSERVABILITY_SAMPLE_RATIO"` + Enabled bool `json:"enabled" env:"PICOCLAW_OBSERVABILITY_ENABLED"` + ServiceName string `json:"service_name" env:"PICOCLAW_OBSERVABILITY_SERVICE_NAME"` + OTLPEndpoint string `json:"otlp_endpoint" env:"PICOCLAW_OBSERVABILITY_OTLP_ENDPOINT"` + Insecure bool `json:"insecure" env:"PICOCLAW_OBSERVABILITY_INSECURE"` + SampleRatio float64 `json:"sample_ratio" env:"PICOCLAW_OBSERVABILITY_SAMPLE_RATIO"` + Langfuse LangfuseConfig `json:"langfuse"` +} + +type LangfuseConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED"` + Host string `json:"host" env:"PICOCLAW_OBSERVABILITY_LANGFUSE_HOST"` + PublicKey string `json:"public_key" env:"PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY"` + SecretKey string `json:"secret_key" env:"PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY"` +} + +func (c LangfuseConfig) IsConfigured() bool { + return c.Enabled && c.Host != "" && c.PublicKey != "" && c.SecretKey != "" } type ProvidersConfig struct { @@ -516,6 +528,17 @@ func LoadConfig(path string) (*Config, error) { data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { + if err := env.Parse(cfg); err != nil { + return nil, err + } + applyObservabilityEnvOverrides(cfg) + applyProviderEnvOverrides(cfg) + if len(cfg.ModelList) == 0 && cfg.HasProvidersConfig() { + cfg.ModelList = ConvertProvidersToModelList(cfg) + } + if err := cfg.ValidateModelList(); err != nil { + return nil, err + } return cfg, nil } return nil, err @@ -543,6 +566,8 @@ func LoadConfig(path string) (*Config, error) { return nil, err } + applyObservabilityEnvOverrides(cfg) + // ProviderConfig uses a shared type and cannot reliably express provider-specific // env tags using template placeholders. Apply explicit overrides here. applyProviderEnvOverrides(cfg) @@ -580,6 +605,21 @@ func applyProviderEnvOverrides(cfg *Config) { setProviderFromEnv("PICOCLAW_PROVIDERS_QWEN", &cfg.Providers.Qwen) } +func applyObservabilityEnvOverrides(cfg *Config) { + if v, ok := os.LookupEnv("PICOCLAW_OBSERVABILITY_LANGFUSE_ENABLED"); ok { + cfg.Observability.Langfuse.Enabled = v == "1" || v == "true" || v == "TRUE" || v == "True" + } + if v, ok := os.LookupEnv("PICOCLAW_OBSERVABILITY_LANGFUSE_HOST"); ok { + cfg.Observability.Langfuse.Host = v + } + if v, ok := os.LookupEnv("PICOCLAW_OBSERVABILITY_LANGFUSE_PUBLIC_KEY"); ok { + cfg.Observability.Langfuse.PublicKey = v + } + if v, ok := os.LookupEnv("PICOCLAW_OBSERVABILITY_LANGFUSE_SECRET_KEY"); ok { + cfg.Observability.Langfuse.SecretKey = v + } +} + func setProviderFromEnv(prefix string, provider *ProviderConfig) { if v, ok := os.LookupEnv(prefix + "_API_KEY"); ok { provider.APIKey = v diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 99f861c79..fc404b24f 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -209,6 +209,12 @@ func TestDefaultConfig_Observability(t *testing.T) { if cfg.Observability.OTLPEndpoint == "" { t.Error("Observability OTLP endpoint should not be empty") } + if cfg.Observability.Langfuse.Enabled { + t.Error("Langfuse should be disabled by default") + } + if cfg.Observability.Langfuse.Host == "" { + t.Error("Langfuse host should not be empty") + } } // TestDefaultConfig_WorkspacePath verifies workspace path is correctly set diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go index 6509fea51..8ab8412da 100644 --- a/pkg/config/defaults.go +++ b/pkg/config/defaults.go @@ -326,6 +326,10 @@ func DefaultConfig() *Config { OTLPEndpoint: "localhost:4317", Insecure: true, SampleRatio: 0.1, + Langfuse: LangfuseConfig{ + Enabled: false, + Host: "http://localhost:3001", + }, }, } } diff --git a/pkg/observability/otel.go b/pkg/observability/otel.go index 2d3c4f8a5..11f436f65 100644 --- a/pkg/observability/otel.go +++ b/pkg/observability/otel.go @@ -2,13 +2,16 @@ package observability import ( "context" + "encoding/base64" "fmt" + "strings" "github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/logger" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" @@ -29,7 +32,7 @@ func Init(ctx context.Context, cfg config.ObservabilityConfig) (func(context.Con clientOpts = append(clientOpts, otlptracegrpc.WithInsecure()) } - exporter, err := otlptracegrpc.New(ctx, clientOpts...) + otlpExporter, err := otlptracegrpc.New(ctx, clientOpts...) if err != nil { return nil, fmt.Errorf("create OTLP trace exporter: %w", err) } @@ -50,11 +53,44 @@ func Init(ctx context.Context, cfg config.ObservabilityConfig) (func(context.Con return nil, fmt.Errorf("create otel resource: %w", err) } - tp := sdktrace.NewTracerProvider( + options := []sdktrace.TracerProviderOption{ sdktrace.WithSampler(sdktrace.TraceIDRatioBased(ratio)), - sdktrace.WithBatcher(exporter), sdktrace.WithResource(res), - ) + sdktrace.WithBatcher(otlpExporter), + } + + if cfg.Langfuse.IsConfigured() { + auth := base64.StdEncoding.EncodeToString([]byte(cfg.Langfuse.PublicKey + ":" + cfg.Langfuse.SecretKey)) + langfuseEndpoint := strings.TrimRight(cfg.Langfuse.Host, "/") + switch { + case strings.HasSuffix(langfuseEndpoint, "/api/public/otel/v1/traces"): + case strings.HasSuffix(langfuseEndpoint, "/api/public/otel"): + langfuseEndpoint += "/v1/traces" + default: + langfuseEndpoint += "/api/public/otel/v1/traces" + } + + langfuseExporter, lfErr := otlptracehttp.New(ctx, + otlptracehttp.WithEndpointURL(langfuseEndpoint), + otlptracehttp.WithHeaders(map[string]string{ + "Authorization": "Basic " + auth, + }), + ) + if lfErr != nil { + logger.WarnCF("otel", "Langfuse exporter disabled (init failed)", map[string]any{ + "error": lfErr.Error(), + "host": cfg.Langfuse.Host, + }) + } else { + options = append(options, sdktrace.WithBatcher(langfuseExporter)) + logger.InfoCF("otel", "Langfuse OTLP exporter enabled", map[string]any{ + "host": cfg.Langfuse.Host, + "endpoint": langfuseEndpoint, + }) + } + } + + tp := sdktrace.NewTracerProvider(options...) otel.SetTracerProvider(tp) otel.SetTextMapPropagator(propagation.TraceContext{})