langfuse
This commit is contained in:
Haytham Amin
parent
8a08c1f497
commit
feb8aff923
11 changed files with 449 additions and 32 deletions
23
.env.example
23
.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
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
1
go.mod
1
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
|
||||
|
|
|
|||
2
go.sum
2
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=
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -326,6 +326,10 @@ func DefaultConfig() *Config {
|
|||
OTLPEndpoint: "localhost:4317",
|
||||
Insecure: true,
|
||||
SampleRatio: 0.1,
|
||||
Langfuse: LangfuseConfig{
|
||||
Enabled: false,
|
||||
Host: "http://localhost:3001",
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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{})
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue