LangChain/LangGraph框架

前提条件

• Python 环境版本 >= 3.9
• LangChain 版本 >= 0.1.0（如使用 LangChain）
• LangGraph 版本 >= 0.0.20（如使用 LangGraph）

探针初始化需要在应用启动时完成，建议在导入LangChain/LangGraph模块之前初始化。

• 在AgentArts观测界面查看数据不收费。
• 但数据上报会产生费用：
  • 调用链上报到APM（应用性能管理服务），费用明细请参考APM计费说明。
  • 指标上报到AOM（应用运维管理服务），费用明细请参考AOM计费说明。

• AOM和APM都是按需计费，并有一定量的免费额度，超出后按实际使用量收费。

框架介绍

LangChain是一个面向大语言模型应用开发的框架，提供了模型调用、Prompt组织、工具接入、检索增强生成（RAG）、Agent构建等能力，帮助开发者快速搭建复杂的LLM应用。

LangGraph是LangChain生态中的Agent编排框架，基于图结构构建多步推理和多Agent协作的LLM应用，支持循环控制流、状态管理和工具调用。

接入后，以下能力将被自动监控：

• LangChain Chain/Agent的执行链路
• LLM调用（模型名称、Token用量、输入/输出内容）
• 工具调用（Tool name、参数、返回结果）
• Retriever/RAG相关调用链路
• LangGraph图节点执行和状态流转

接入步骤

示例代码

以下示例展示了一个基于LangGraph的Agent应用，包含工具调用和多步推理：

示例中使用华为云MaaS服务提供的大模型接口，可登录华为云MaaS服务控制台，创建API Key。

图2 创建API Key

import logging
import os
from datetime import datetime

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
_logger = logging.getLogger(__name__)

SYSTEM_PROMPT = "你是一个有用的AI助手，可以使用工具来帮助用户完成任务。请用中文回答。"

# MaaS 模型配置
LLM_CONFIG = {
    "model": "deepseek-v4.1-flash",
    "base_url": "https://api.modelarts-maas.com/openai/v1",
    "api_key": os.environ.get("LLM_API_KEY", "<MODEL_API_KEY>"),  # 替换为在华为云 MaaS 控制台申请的 API Key
    "temperature": 0,
}


@tool
def calculator(expression: str) -> str:
    """计算数学表达式，例如 '2 + 3 * 4'"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"计算失败: {e}"


@tool
def read_file(file_path: str) -> str:
    """读取指定路径的文件内容"""
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"读取失败: {e}"


def create_llm() -> ChatOpenAI:
    """创建 LLM 实例"""
    return ChatOpenAI(
        model=LLM_CONFIG["model"],
        temperature=LLM_CONFIG["temperature"],
        api_key=LLM_CONFIG["api_key"],
        base_url=LLM_CONFIG["base_url"],
    )


def create_agent_executor():
    """创建 Agent 执行器"""
    llm = create_llm()
    tools = [calculator, read_file]
    return create_agent(
        model=llm,
        tools=tools,
        system_prompt=SYSTEM_PROMPT,
    )


def main():
    print("=" * 50)
    print("LangChain 简单 Agent")
    print("模型: DeepSeek-V4.1-Flash (MaaS)")
    print("工具: calculator, read_file")
    print("输入 'quit' 或 'exit' 退出")
    print("=" * 50)

    agent = create_agent_executor()
    messages = []

    thread_id = f"session-{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    config = {"configurable": {"thread_id": thread_id}}

    while True:
        try:
            user_input = input("\n你: ")
        except (EOFError, KeyboardInterrupt):
            print("\n再见!")
            break

        if user_input.lower() in ("quit", "exit", "q"):
            print("再见!")
            break
        if not user_input:
            continue

        messages.append({"role": "user", "content": user_input})

        if user_input == "/new":
            # 新建会话，基于当前时间生成新的 thread_id
            thread_id = f"session-{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            config = {"configurable": {"thread_id": thread_id}}
            print(f"[新会话: {thread_id}]")
            continue

        try:
            # 通过 OTel Baggage 传递会话信息，探针自动关联到 span
            from opentelemetry.baggage import set_baggage
            from opentelemetry import context as ctx_api
            context = set_baggage("session_id", thread_id)
            context = set_baggage("user_id", "langchain_user", context)
            token = ctx_api.attach(context)
            try:
                response = agent.invoke({"messages": messages}, config=config)
                ai_messages = response.get("messages", [])
                last_ai = ai_messages[-1] if ai_messages else None
                if last_ai:
                    assistant_reply = getattr(last_ai, "content", str(last_ai))
                    print(f"\n助手: {assistant_reply}")
                    messages.append({"role": "assistant", "content": assistant_reply})
                else:
                    print("\n助手: (无回复)")
            finally:
                ctx_api.detach(token)
        except Exception as e:
            _logger.error("agent 调用失败: %s", e)
            print(f"\n错误: {e}")


if __name__ == "__main__":
    main()

将上方示例代码保存为app.py，在配置好环境变量的终端中执行以下命令，通过探针启动应用：

opentelemetry-instrument python app.py 

启动后可在终端与智能体交互，输入问题进行对话。探针会自动采集所有LLM调用、工具调用等数据并上报到AgentArts平台。

如果使用LangChain v1及之后版本，请使用from langchain.agents import create_agent代替from langgraph.prebuilt import create_react_agent。

查看监控数据

数据上报成功后，可在 AgentArts 智能体平台的观测功能中查看监控数据。

1. 登录AgentArts 智能体平台。
2. 在左侧导航栏选择“观测与优化 > 观测”。
3. 在“智能体列表”页面，找到已接入的智能体，单击智能体名称进入详情页面。

查看业务指标

在“指标分析”页签中，可以查看关键业务指标：

• Tokens消耗：Input Tokens和Output Tokens的消耗总量
• 模型调用次数：应用调用大模型的累计次数
• 模型调用平均耗时：模型调用的平均响应时间
• 模型调用成功率：大模型调用成功的比例
• 会话数：应用产生的会话总数
• 响应成功率：服务响应的成功率

查看调用链信息

在“调用链分析”页签中，可以查看完整的Trace调用链路：

• 查看各Span类型的层级关系及各节点耗时
• 单击具体Span查看详细信息：
  • LLM节点：可查看模型名称、输入/输出消息（Input/Output Messages）、Token消耗（Total tokens）
  • Tool节点：可查看工具名称、输入参数、返回结果
  • LangGraph节点：可查看图节点执行顺序、状态转换

查看会话分析

在“会话分析”页签中，可以查看智能体与用户对话的历史数据：

• 追踪对话上下文
• 分析模型响应耗时
• 统计Token消耗
• 快速定位对话异常问题

更多参考

传递session_id与user_id

在调用agent.invoke时，可通过OTel Baggage传递会话信息，传入session_id和user_id，探针自动关联到span用于会话分析和用户分析：

from opentelemetry.baggage import set_baggage
from opentelemetry import context as ctx_api
context = set_baggage("session_id", "session-01")
context = set_baggage("user_id", "langgraph_user", context)
token = ctx_api.attach(context)
try:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "您好"}]}
    )
finally:
    ctx_api.detach(token)

• session_id：会话标识，用于关联同一会话内的多轮调用。
• user_id：用户标识，用于按用户维度统计和过滤。

控制内容采集

默认情况下，探针会采集LLM的输入/输出内容。如需关闭：

export OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=false

代码手动插桩

如果无法使用opentelemetry-instrument命令自动插桩，也可以通过创建sitecustomize.py手动初始化：

import os

os.environ["OTEL_PYTHON_LOG_CORRELATION"] = "True"
os.environ["OTEL_PYTHON_LOG_LEVEL"] = "debug"           # 设置日志级别，可根据情况调整
os.environ["OTEL_PYTHON_LOG_FORMAT"] = "%(asctime)s [%(levelname)s] [trace_id=%(otelTraceID)s span_id=%(otelSpanID)s] %(message)s"

import logging
from opentelemetry.instrumentation.logging import LoggingInstrumentor

root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(logging.DEBUG)                     # 设置日志级别，可根据情况调整

LoggingInstrumentor().instrument(
    set_logging_format=True,
    log_level=logging.DEBUG                             # 设置日志级别，可根据情况调整，确保 OTel 的 Handler 也接受对应日志级别
)

from opentelemetry.instrumentation.huawei_auto_instrumentation import initialize

initialize()

常见问题

无数据上报

1. 检查环境变量是否正确配置，特别是agentarts_agentops_apm_token、agentarts_agentops_apm_exporter_endpoint等关键参数是否完整且无多余空格。
2. 确认agentarts_agentops_otel_agent_apm_enable和agentarts_agentops_otel_agent_aom_enable已设置为true，默认值为false不会上报数据。
3. 检查网络连接，确保运行环境能访问APM/AOM端点地址。可通过curl <endpoint>命令验证连通性。
4. 检查凭证是否有效且未过期，可在AgentArts平台的接入指南页面重新获取最新凭证。
5. 查看应用启动日志，搜索opentelemetry或instrument关键字，排查探针初始化是否报错。

部分Span缺失

1. 确认是否通过opentelemetry-instrument python app.py命令启动应用。直接使用python app.py启动将无法触发自动插桩，需改用探针命令启动或参考代码手动插桩进行初始化。
2. 确认LangChain/LangGraph版本满足最低要求（LangChain >= 0.1.0，LangGraph >= 0.0.20），低版本框架的内部调用结构可能与探针不兼容。
3. 如使用手动插桩方式，确认探针初始化代码在import langchain之前执行。若LangChain模块先于探针加载，插桩将无法生效，导致相关Span丢失。

Token统计不准确

1. 确保使用最新版本的探针。
2. 流式场景下确认已开启stream_usage=True。
3. 模型在响应中不返回usage字段或返回的Token数为近似值，此为模型侧行为，非探针问题。如遇此情况，可在AgentArts调用链详情中确认模型节点的usage字段是否为空。