作为一名长期使用大模型 API 构建 Agent 系统的工程师，我在 2025 年经历了从官方 API 到多个中转服务的迁移折腾，最终在 HolySheep AI 找到了性能与成本的平衡点。今天我将分享如何为 LangGraph Agent 添加完整的执行轨迹监控，同时对比迁移到 HolySheep 的真实收益。

为什么需要监控 LangGraph Agent 执行轨迹

LangGraph 的核心优势在于其状态机架构，但当 Agent 链路变长时，节点间的数据流转、Token 消耗、延迟分布往往成为黑盒。我在生产环境中曾遇到一个 12 步的客服 Agent，每次用户请求的 Token 消耗波动从 2000 到 80000 不等，根本无法定位问题根源。

监控 Agent 轨迹的价值：

迁移决策：从官方 API 到 HolySheep 的 ROI 估算

我原本使用官方 API 调用 GPT-4 系列模型，但月度账单让我不得不考虑替代方案。以下是我做迁移决策时的核心数据对比：

维度官方 APIHolySheep AI节省比例
GPT-4.1 Output$8.00/MTok$8.00/MTok汇率节省 85%+
Claude Sonnet 4.5 Output$15.00/MTok$15.00/MTok汇率节省 85%+
Gemini 2.5 Flash Output$2.50/MTok$2.50/MTok汇率节省 85%+
DeepSeek V3.2 Output$0.42/MTok$0.42/MTok汇率节省 85%+
国内延迟200-400ms<50ms延迟降低 80%
充值方式国际信用卡微信/支付宝便捷度提升

以我目前的调用量（月均 5000 万 Token）计算：

环境准备与依赖安装

# 创建虚拟环境
python -m venv langgraph-monitor
source langgraph-monitor/bin/activate  # Windows: langgraph-monitor\Scripts\activate


安装核心依赖

pip install langgraph langchain-openai langchain-anthropic \
    langchain-google-genai psycopg2-binary redis openai \
    python-json-logger opentelemetry-api opentelemetry-sdk


安装可视化依赖

pip install streamlit plotly dash

核心实现：LangGraph 监控回调系统

我的监控方案基于 LangChain 的 Callback 机制，扩展了 Token 计数、延迟追踪和轨迹存储功能。以下是完整实现：

"""
LangGraph Agent 执行轨迹监控系统
基于 HolySheep AI API 实现
"""
import os
import json
import time
import uuid
from datetime import datetime
from typing import Any, Optional
from dataclasses import dataclass, field
from collections import defaultdict

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from langchain_core.outputs import LLMResult

import psycopg2
import redis
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource


@dataclass
class ExecutionNode:
    """单次 LLM 调用的完整轨迹节点"""
    node_id: str
    node_name: str
    input_tokens: int = 0
    output_tokens: int = 0
    latency_ms: float = 0.0
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None
    model_name: str = ""
    error: Optional[str] = None
    raw_response: Optional[dict] = None


@dataclass
class ExecutionTrace:
    """完整 Agent 执行轨迹"""
    trace_id: str
    user_id: str
    thread_id: str
    nodes: list = field(default_factory=list)
    total_tokens: int = 0
    total_latency_ms: float = 0.0
    created_at: datetime = field(default_factory=datetime.now)
    status: str = "running"  # running, completed, failed


class HolySheepCallbackHandler(BaseCallbackHandler):
    """
    HolySheep AI API 专用监控回调
    自动追踪 Token 消耗、延迟和执行轨迹
    """
    
    def __init__(
        self,
        trace_id: str,
        user_id: str,
        thread_id: str,
        db_config: dict,
        redis_url: str = "redis://localhost:6379/0"
    ):
        self.trace_id = trace_id
        self.user_id = user_id
        self.thread_id = thread_id
        self.current_node: Optional[ExecutionNode] = None
        self.trace = ExecutionTrace(
            trace_id=trace_id,
            user_id=user_id,
            thread_id=thread_id
        )
        
        # 初始化存储连接
        self.db_conn = psycopg2.connect(**db_config)
        self.redis_client = redis.from_url(redis_url)
        
        # 初始化 OpenTelemetry
        self._init_telemetry()
    
    def _init_telemetry(self):
        """初始化分布式追踪"""
        resource = Resource.create({
            "service.name": "langgraph-agent",
            "user.id": self.user_id,
            "trace.id": self.trace_id
        })
        provider = TracerProvider(resource=resource)
        trace.set_tracer_provider(provider)
        self.tracer = trace.get_tracer(__name__)
    
    def on_llm_start(
        self,
        serialized: dict,
        prompts: list,
        *,
        run_id: uuid.UUID,
        parent_run_id: Optional[uuid.UUID] = None,
        tags: Optional[list] = None,
        metadata: Optional[dict] = None,
        **kwargs
    ) -> None:
        """LLM 调用开始"""
        self.current_node = ExecutionNode(
            node_id=str(run_id),
            node_name=metadata.get("ls_provider", "unknown") if metadata else "unknown"
        )
        self.current_node.start_time = datetime.now()
        
        # 从 kwargs 提取模型名称（HolySheep 返回格式）
        if "model" in kwargs:
            self.current_node.model_name = kwargs["model"]
        elif metadata and "ls_model_name" in metadata:
            self.current_node.model_name = metadata["ls_model_name"]
    
    def on_llm_end(
        self,
        response: LLMResult,
        *,
        run_id: uuid.UUID,
        **kwargs
    ) -> None:
        """LLM 调用完成 - 核心统计点"""
        if self.current_node is None:
            return
        
        self.current_node.end_time = datetime.now()
        self.current_node.latency_ms = (
            self.current_node.end_time - self.current_node.start_time
        ).total_seconds() * 1000
        
        # 提取 Token 消耗（HolySheep API 返回格式）
        if response.generations:
            generation = response.generations[0][0]
            if hasattr(generation, 'generation_info') and generation.generation_info:
                info = generation.generation_info
                self.current_node.input_tokens = info.get('tokens_in', 0)
                self.current_node.output_tokens = info.get('tokens_out', 0)
                
                # 存储原始响应用于调试
                if 'usage' in info:
                    self.current_node.raw_response = {
                        'usage': info['usage'],
                        'model': info.get('model', 'unknown')
                    }
        
        # 更新轨迹统计
        self.trace.nodes.append(self.current_node)
        self.trace.total_tokens += (
            self.current_node.input_tokens + 
            self.current_node.output_tokens
        )
        self.trace.total_latency_ms += self.current_node.latency_ms
        
        # 异步持久化到 Redis（实时查询用）
        self._cache_to_redis()
        
        self.current_node = None
    
    def on_llm_error(
        self,
        error: BaseException,
        *,
        run_id: uuid.UUID,
        **kwargs
    ) -> None:
        """LLM 调用错误"""
        if self.current_node:
            self.current_node.error = str(error)
            self.current_node.end_time = datetime.now()
            self.trace.nodes.append(self.current_node)
    
    def _cache_to_redis(self):
        """缓存最新状态到 Redis，支持实时监控"""
        cache_key = f"trace:{self.trace_id}"
        cache_data = {
            "trace_id": self.trace_id,
            "total_tokens": self.trace.total_tokens,
            "total_latency_ms": self.trace.total_latency_ms,
            "node_count": len(self.trace.nodes),
            "last_update": datetime.now().isoformat()
        }
        self.redis_client.hset(cache_key, mapping=cache_data)
        self.redis_client.expire(cache_key, 86400)  # 24小时过期
    
    def persist_to_database(self):
        """持久化到 PostgreSQL（后续分析用）"""
        cursor = self.db_conn.cursor()
        
        # 插入轨迹记录
        cursor.execute("""
            INSERT INTO agent_traces 
            (trace_id, user_id, thread_id, total_tokens, total_latency_ms, status)
            VALUES (%s, %s, %s, %s, %s, %s)
            ON CONFLICT (trace_id) DO UPDATE SET
                total_tokens = EXCLUDED.total_tokens,
                total_latency_ms = EXCLUDED.total_latency_ms
        """, (
            self.trace.trace_id,
            self.trace.user_id,
            self.trace.thread_id,
            self.trace.total_tokens,
            self.trace.total_latency_ms,
            self.trace.status
        ))
        
        # 插入节点详情
        for node in self.trace.nodes:
            cursor.execute("""
                INSERT INTO trace_nodes
                (trace_id, node_id, node_name, model_name, 
                 input_tokens, output_tokens, latency_ms, error)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            """, (
                self.trace.trace_id,
                node.node_id,
                node.node_name,
                node.model_name,
                node.input_tokens,
                node.output_tokens,
                node.latency_ms,
                node.error
            ))
        
        self.db_conn.commit()
        cursor.close()
    
    def finalize(self, status: str = "completed"):
        """结束追踪并持久化"""
        self.trace.status = status
        try:
            self.persist_to_database()
        except Exception as e:
            print(f"持久化失败: {e}")
        finally:
            self.db_conn.close()
            self.redis_client.close()



数据库初始化 SQL

INIT_SQL = """
CREATE TABLE IF NOT EXISTS agent_traces (
    trace_id VARCHAR(64) PRIMARY KEY,
    user_id VARCHAR(64),
    thread_id VARCHAR(64),
    total_tokens BIGINT DEFAULT 0,
    total_latency_ms FLOAT DEFAULT 0.0,
    status VARCHAR(20) DEFAULT 'running',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS trace_nodes (
    id SERIAL PRIMARY KEY,
    trace_id VARCHAR(64),
    node_id VARCHAR(64),
    node_name VARCHAR(128),
    model_name VARCHAR(64),
    input_tokens INT DEFAULT 0,
    output_tokens INT DEFAULT 0,
    latency_ms FLOAT DEFAULT 0.0,
    error TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (trace_id) REFERENCES agent_traces(trace_id)
);

CREATE INDEX idx_traces_user ON agent_traces(user_id);
CREATE INDEX idx_traces_thread ON agent_traces(thread_id);
CREATE INDEX idx_nodes_trace ON trace_nodes(trace_id);
"""

集成 HolySheep API 的 LangGraph 应用

现在将监控回调与 HolySheep API 集成。以下是一个完整的客服 Agent 示例，支持多模型切换：

"""
LangGraph Agent with HolySheep API Integration
使用 HolySheep AI 实现低成本、高性能的 Agent 系统
"""
import os
import uuid
from typing import Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage, SystemMessage


HolySheep API 配置

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


监控回调（见上一节）

from monitoring import HolySheepCallbackHandler, INIT_SQL


数据库配置

DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "langgraph_monitor",
    "user": "postgres",
    "password": os.getenv("DB_PASSWORD", "password")
}


class AgentState(dict):
    """Agent 状态定义"""
    messages: list
    current_model: str
    trace_id: str
    user_id: str
    thread_id: str


def create_model(model_name: str, api_key: str = HOLYSHEEP_API_KEY):
    """创建 HolySheep API 模型实例"""
    base_config = {
        "api_key": api_key,
        "base_url": HOLYSHEEP_BASE_URL
    }
    
    if "gpt" in model_name.lower():
        return ChatOpenAI(
            model=model_name,
            temperature=0.7,
            max_tokens=2048,
            **base_config
        )
    elif "claude" in model_name.lower():
        return ChatAnthropic(
            model_name=model_name,
            max_tokens=2048,
            **base_config
        )
    elif "gemini" in model_name.lower():
        return ChatGoogleGenerativeAI(
            model=model_name,
            **base_config
        )
    else:
        # 默认使用 OpenAI 兼容接口
        return ChatOpenAI(
            model=model_name,
            **base_config
        )



系统提示词

CUSTOMER_SERVICE_PROMPT = """你是一个专业的客服助手。请根据用户的问题，提供准确、友好的回答。
回答要求：
1. 简洁明了，不超过 200 字
2. 如果需要更多信息，明确询问
3. 涉及退款、政策等问题，引导用户联系人工客服
"""


def intent_detection(state: AgentState) -> AgentState:
    """节点1：意图识别 - 决定使用哪个模型"""
    messages = state["messages"]
    trace_id = state["trace_id"]
    
    # 使用轻量模型进行意图识别
    model = create_model("gpt-4o-mini")
    
    handler = HolySheepCallbackHandler(
        trace_id=f"{trace_id}_intent",
        user_id=state["user_id"],
        thread_id=state["thread_id"],
        db_config=DB_CONFIG
    )
    
    response = model.invoke(
        [SystemMessage(content="判断用户意图：product_inquiry/product_complaint/refund_request/general")],
        config={"callbacks": [handler]}
    )
    
    intent = response.content.strip().lower()
    
    # 根据意图选择主模型
    if "refund" in intent or "complaint" in intent:
        selected_model = "claude-sonnet-4-20250514"  # Claude 更擅长处理复杂对话
    elif "product" in intent:
        selected_model = "gpt-4.1"  # GPT-4.1 产品知识更全面
    else:
        selected_model = "gemini-2.5-flash"  # 日常问题用 Flash 模型最经济
    
    state["current_model"] = selected_model
    handler.finalize("completed")
    
    return state


def response_generation(state: AgentState) -> AgentState:
    """节点2：生成回复 - 使用选定的模型"""
    messages = state["messages"]
    selected_model = state["current_model"]
    trace_id = state["trace_id"]
    
    model = create_model(selected_model)
    
    handler = HolySheepCallbackHandler(
        trace_id=f"{trace_id}_response",
        user_id=state["user_id"],
        thread_id=state["thread_id"],
        db_config=DB_CONFIG
    )
    
    response = model.invoke(
        [SystemMessage(content=CUSTOMER_SERVICE_PROMPT)] + messages,
        config={"callbacks": [handler]}
    )
    
    state["messages"].append(response)
    handler.finalize("completed")
    
    return state


def build_agent_graph():
    """构建 LangGraph 工作流"""
    workflow = StateGraph(AgentState)
    
    workflow.add_node("intent_detection", intent_detection)
    workflow.add_node("response_generation", response_generation)
    
    workflow.set_entry_point("intent_detection")
    workflow.add_edge("intent_detection", "response_generation")
    workflow.add_edge("response_generation", END)
    
    return workflow.compile()


def run_customer_service(user_message: str, user_id: str = "anonymous"):
    """运行客服 Agent"""
    trace_id = str(uuid.uuid4())
    thread_id = f"thread_{user_id}_{int(time.time())}"
    
    state = AgentState(
        messages=[HumanMessage(content=user_message)],
        current_model="",
        trace_id=trace_id,
        user_id=user_id,
        thread_id=thread_id
    )
    
    graph = build_agent_graph()
    result = graph.invoke(state)
    
    return {
        "trace_id": trace_id,
        "response": result["messages"][-1].content,
        "model_used": result["current_model"]
    }


if __name__ == "__main__":
    # 初始化数据库
    import psycopg2
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()
    cursor.execute(INIT_SQL)
    conn.commit()
    conn.close()
    
    # 测试调用
    result = run_customer_service(
        user_message="我想咨询一下产品的退款政策",
        user_id="user_123"
    )
    print(f"Trace ID: {result['trace_id']}")
    print(f"Response: {result['response']}")
    print(f"Model Used: {result['model_used']}")

执行轨迹可视化面板

监控数据需要可视化才能发挥价值。我使用 Streamlit 构建了一个实时监控面板：

"""
LangGraph Agent 轨迹可视化面板
实时监控 Token 消耗、延迟和执行状态
"""
import streamlit as st
import psycopg2
import redis
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
import pandas as pd


数据库连接

DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "langgraph_monitor",
    "user": "postgres",
    "password": "password"
}
REDIS_URL = "redis://localhost:6379/0"


def get_recent_traces(hours: int = 24) -> pd.DataFrame:
    """获取最近的执行轨迹"""
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT 
            trace_id,
            user_id,
            thread_id,
            total_tokens,
            total_latency_ms,
            status,
            created_at
        FROM agent_traces
        WHERE created_at > NOW() - INTERVAL '%s hours'
        ORDER BY created_at DESC
    """, (hours,))
    
    columns = [desc[0] for desc in cursor.description]
    data = cursor.fetchall()
    
    cursor.close()
    conn.close()
    
    return pd.DataFrame(data, columns=columns)


def get_trace_nodes(trace_id: str) -> pd.DataFrame:
    """获取指定轨迹的所有节点"""
    conn = psycopg2.connect(**DB_CONFIG)
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT 
            node_id,
            node_name,
            model_name,
            input_tokens,
            output_tokens,
            latency_ms,
            error,
            created_at
        FROM trace_nodes
        WHERE trace_id = %s
        ORDER BY created_at
    """, (trace_id,))
    
    columns = [desc[0] for desc in cursor.description]
    data = cursor.fetchall()
    
    cursor.close()
    conn.close()
    
    return pd.DataFrame(data, columns=columns)


def get_redis_stats() -> dict:
    """获取 Redis 缓存的实时状态"""
    r = redis.from_url(REDIS_URL)
    keys = r.keys("trace:*")
    
    stats = {
        "active_traces": len(keys),
        "total_active_tokens": 0,
        "total_active_latency": 0.0
    }
    
    for key

相关资源
相关文章