本文探讨了如何利用OpenLLMetry技术，对大语言模型（LLM）的应用进行全面监控。文章首先介绍了LLM在内容创作、教育、医疗、客服和金融等领域的广泛应用，并强调了监控LLM的关键信号，如追踪请求元数据、响应元数据以及各项指标。随后，文章详细阐述了如何通过观测云和DataKit部署OpenLLMetry，实现LLM应用的全链路追踪和性能分析，从而帮助用户优化LLM的性能和资源利用率。

💡LLM 应用监控的核心在于追踪和指标的结合。追踪关注请求和响应的细节，包括温度、top_p、模型名称、提示词、Token数量和成本等，从而深入了解LLM的运行状态。

⚙️ OpenLLMetry 基于 OpenTelemetry，为 LLM 应用提供了专门的监控和调试工具。通过扩展 OpenTelemetry 的功能，它能够标准化输出关键性能指标和追踪信息。

☁️ 观测云与 DataKit 的结合，为 LLM 应用提供了完整的监控解决方案。DataKit 负责收集数据，而观测云则提供可视化界面，帮助用户进行全链路追踪、性能分析和问题排查。

📊 观测云的场景视图和链路追踪功能，能够实时展示 LLM 的调用链信息、性能指标和资源消耗。这有助于用户快速定位性能瓶颈，优化模型配置，提高 LLM 的运行效率。

LLM 应用

大语言模型（LLM）的应用广泛且极具创新性，它通过强大的自然语言处理能力，为多个领域带来了深刻的变革。在内容创作方面，LLM 能够生成高质量的文案、故事、诗歌等，为作家、广告商和创意工作者提供灵感和辅助创作工具。在教育领域，它可以根据学生的学习进度和需求，生成个性化的学习材料和辅导内容，提升学习效果。在医疗行业，LLM 可以辅助医生解读病历、分析症状，甚至生成初步的诊断建议，为医疗决策提供支持。此外，LLM 还广泛应用于智能客服，能够快速响应用户问题，提供精准解答，提升客户满意度；在金融领域，它能够分析市场数据，生成投资分析报告和风险预警信息。随着技术的不断进步，LLM 的应用场景还在不断拓展，其强大的语言生成和理解能力使其成为推动各行业智能化发展的重要力量。

LLM 关键信号

在应用程序中使用大规模语言模型（LLM）与传统机器学习（ML）模型有所不同。主要区别在于，LLM 通常通过外部 API 调用访问，而不是在本地或内部运行。因此，捕获事件序列（通过追踪）尤为重要，特别是在基于 RAG（检索增强生成）的应用程序中，LLM 使用前后可能会有多个事件。此外，分析聚合数据（通过指标）可以快速提供关于请求、令牌和成本的概览，这对于优化性能和管理成本非常重要。以下是需要监控的关键信号。

追踪（Tracing）

请求元数据：在 LLM 的上下文中，请求元数据非常重要，因为各种参数（如温度和 top_p）会显著影响响应质量和成本。具体需要监控的方面包括：

Temperature：表示希望从模型输出中获得的创造力或随机性水平。调整此参数会显著影响生成内容的性质。top_p：决定模型的选择性，即从最有可能的词汇中选择一定比例的词。较高的 top_p 值意味着模型考虑的词汇范围更广，使文本更加多样化。模型名称或版本：对于跟踪随时间的变化至关重要，因为 LLM 的更新可能会影响性能或响应特性。提示词：发送给 LLM 的确切输入。与内部 ML 模型不同，LLM 的输入可能变化很大，这会影响输出复杂性和成本。

响应元数据：由于 LLM 是通过 API 进行交互的，因此跟踪响应的具体细节对于成本管理和质量评估至关重要：

Token：直接影响成本，并且是响应长度和复杂性的度量。Cost：对于预算规划至关重要，因为基于 API 的成本会随着请求数量和每个请求的复杂性而增加。响应详情：类似于提示详情，但从响应的角度提供对模型输出特性的洞察，以及潜在的效率低下或意外成本领域。

指标（Metric）

请求量：发送到 LLM 服务的总请求数。这有助于了解需求模式，并识别使用中的任何异常情况，例如突然的激增或下降。请求时长：从发出请求到接收到 LLM 响应所需的时间。这包括网络延迟和 LLM 生成响应所需的时间，能够提供关于 LLM 服务性能和可靠性的洞察。成本和令牌计数：跟踪随时间累积的总成本和消耗的令牌数量对于预算规划和成本优化策略至关重要。监控这些指标可以提醒你是否存在意外增加的情况，这可能表明 LLM 的使用效率低下或需要进行优化。

OpenLLMetry

OpenLLMetry 是由 Traceloop 团队以 Apache 2.0 许可证开发和维护，通过扩展 OpenTelemetry 的功能，为 LLM 应用提供了专门的监控和调试工具。它利用 OpenTelemetry 的标准化遥测数据格式，将关键性能指标和追踪信息标准化输出。

观测云

观测云是一款专为 IT 工程师打造的全链路可观测产品，它集成了基础设施监控、应用程序性能监控和日志管理，为整个技术栈提供实时可观察性。这款产品能够帮助工程师全面了解端到端的用户体验追踪，了解应用内函数的每一次调用，以及全面监控云时代的基础设施。此外，观测云还具备快速发现系统安全风险的能力，为数字化时代提供安全保障。

部署 DataKit

DataKit 是一个开源的、跨平台的数据收集和监控工具，由观测云开发并维护。它旨在帮助用户收集、处理和分析各种数据源，如日志、指标和事件，以便进行有效的监控和故障排查。DataKit 支持多种数据输入和输出格式，可以轻松集成到现有的监控系统中。

登录观测云控制台，在「集成」 - 「DataKit」选择对应安装方式，当前采用 Linux 主机部署 DataKit。

采集器配置

DataKit 开启 OpenTelemetry 采集器

进入 DataKit 安装目录下的 conf.d/opentelemetry 目录，复制 opentelemetry.conf.sample 并命名为 opentelemetry.conf。如下：

[[inputs.opentelemetry]]  ## customer_tags will work as a whitelist to prevent tags send to data center.  ## All . will replace to _ ,like this :  ## "project.name" to send to GuanCe center is "project_name"    customer_tags = ["llm.request.type","traceloop.entity.path","llm.is_streaming","gen_ai.openai.api_base","gen_ai.prompt.1.content","gen_ai.response.model","gen_ai.completion.0.content","gen_ai.request.model","gen_ai.request.temperature","gen_ai.system","traceloop.workflow.name"]  ...

重启 DataKit

datakit service -R

LLM 应用开启 OpenLLMetry

OpenLLMetry 需要依赖 OpenTelemetry SDK。

安装 OpenTelemetry SDK

pip install opentelemetry-api opentelemetry-instrumentationpip install opentelemetry-instrumentation-flask

安装 OpenLLMetry SDK

pip install traceloop-sdk

在智能体中初始化 OpenLLMetry

from traceloop.sdk import TraceloopTraceloop.init(app_name="kimi_openllmetry_stream_flask")

OpenLLMetry 示例代码

import osimport httpxfrom flask import Flask, request, Response,jsonify,stream_with_contextfrom traceloop.sdk import Traceloopfrom traceloop.sdk.decorators import workflow,taskfrom openai import OpenAIfrom opentelemetry.instrumentation.flask import FlaskInstrumentorapp = Flask(__name__)# 使用FlaskInstrumentor自动插桩Flask应用FlaskInstrumentor().instrument_app(app)Traceloop.init(app_name="kimi_openllmetry_stream_flask")# 从环境变量中获取 API Keyapi_key = os.getenv("MOONSHOT_API_KEY")if not api_key:    raise ValueError("请设置 MOONSHOT_API_KEY 环境变量")client = OpenAI(    api_key=api_key,    base_url="https://api.moonshot.cn/v1",)def estimate_token_count(input_messages) -> int:    """    计算输入消息的 Tokens 数量。    """    try:        header = {            "Authorization": f"Bearer {api_key}",        }        data = {            "model": "moonshot-v1-128k",            "messages": input_messages,        }        with httpx.Client() as client:            print("请求接口")            r = client.post("https://api.moonshot.cn/v1/tokenizers/estimate-token-count", headers=header, json=data)            r.raise_for_status()            response_data = r.json()            print(response_data["data"]["total_tokens"])            return response_data["data"]["total_tokens"]    except httpx.RequestError as e:        print(f"请求失败: {e}")        raise    except (KeyError, ValueError) as e:        print(f"解析响应失败: {e}")        raisedef select_model(input_messages, max_tokens=1024) -> str:    """    根据输入的上下文消息和预期的 max_tokens 值选择合适的模型。    """    if not isinstance(max_tokens, int) or max_tokens <= 0:        raise ValueError("max_tokens 必须是正整数")    prompt_tokens = estimate_token_count(input_messages)    total_tokens = prompt_tokens + max_tokens    if total_tokens <= 8 * 1024:        return "moonshot-v1-8k"    elif total_tokens <= 32 * 1024:        return "moonshot-v1-32k"    elif total_tokens <= 128 * 1024:        return "moonshot-v1-128k"    else:        raise ValueError("tokens 数量超出限制 ?")@app.route('/ask', methods=['POST'])@workflow(name="ask_workflow")def ask():    data = request.json    messages = data.get('messages')    max_tokens = data.get('max_tokens', 2048)    if not messages:        return jsonify({"error": "messages 字段不能为空"}), 400    try:        model = select_model(messages, max_tokens)                completion = client.chat.completions.create(            model=model,            messages=messages,            max_tokens=max_tokens,            temperature=0.3,            stream=True  # 启用流式生成        )                def generate():            for chunk in completion:                # yield chunk.choices[0].delta.content or ''                delta = chunk.choices[0].delta                if delta.content:                    print(delta.content, end="")                    yield delta.content or ''        return Response(stream_with_context(generate()), content_type='text/event-stream')    except Exception as e:        return jsonify({"error": str(e)}), 500if __name__ == '__main__':    app.run(debug=True,port=5001)

配置 env，将数据通过 OpenTelemetry 上报到 DataKit。

export TRACELOOP_BASE_URL=http://localhost:9529/otel

关键指标

指标名称	描述	单位
gen_ai.client.generation.choices	客户端生成的选项数量	个
gen_ai.client.operation.duration_bucket	客户端操作的持续时间直方图桶	毫秒
gen_ai.client.operation.duration_count	客户端操作的总次数	次
gen_ai.client.operation.duration_max	客户端操作的最大持续时间	毫秒
gen_ai.client.operation.duration_min	客户端操作的最小持续时间	毫秒
gen_ai.client.operation.duration_sum	客户端操作的总持续时间	毫秒
llm.openai.chat_completions.streaming_time_to_first_token_bucket	OpenAI 聊天补全功能流式传输中首次生成 Token 的时间直方图桶	毫秒
llm.openai.chat_completions.streaming_time_to_first_token_count	OpenAI 聊天补全功能流式传输中首次生成 Token 的总次数	次
llm.openai.chat_completions.streaming_time_to_first_token_max	OpenAI 聊天补全功能流式传输中首次生成 Token 的最大时间	毫秒
llm.openai.chat_completions.streaming_time_to_first_token_min	OpenAI 聊天补全功能流式传输中首次生成 Token 的最小时间	毫秒
llm.openai.chat_completions.streaming_time_to_first_token_sum	OpenAI 聊天补全功能流式传输中首次生成 Token 的总时间	毫秒
llm.openai.chat_completions.streaming_time_to_generate_bucket	OpenAI 聊天补全功能流式传输中生成内容的总时间直方图桶	毫秒
llm.openai.chat_completions.streaming_time_to_generate_count	OpenAI 聊天补全功能流式传输中生成内容的总次数	次
llm.openai.chat_completions.streaming_time_to_generate_max	OpenAI 聊天补全功能流式传输中生成内容的最大时间	毫秒
llm.openai.chat_completions.streaming_time_to_generate_min	OpenAI 聊天补全功能流式传输中生成内容的最小时间	毫秒
llm.openai.chat_completions.streaming_time_to_generate_sum	OpenAI 聊天补全功能流式传输中生成内容的总时间	毫秒

场景视图

登录观测云控制台，点击「场景」 -「新建仪表板」，输入 “OpenLLMetry”， 选择 “OpenLLMetry”，点击 “确定” 即可添加视图。

链路追踪

在链路列表，可以查看模型调用链信息，链路展示了模型的 Input、Output 等关键信息。

总结

观测云通过集成 OpenLLMetry，能够实时采集和分析大语言模型（LLM）的链路信息，包括模型的调用链路、性能指标、资源消耗以及异常情况等。这种集成使得用户可以清晰地监控模型的运行状态，优化模型的性能和资源利用率，同时快速定位和解决潜在问题，从而提升大语言模型在生产环境中的稳定性和效率。