LLM 大模型框架 LangChain 可观测性最佳实践

    LLM(Large Language Model)大模型的可观测性是指对模型内部运行过程的理解和监控能力。由于LLM大模型通常具有庞大的参数量和复杂的网络结构,因此对其内部状态和运行过程的理解和监控是一个重要的问题。

    什么是 LangChain?

    LangChain 是一个开源的应用开发框架,旨在将大型语言模型(LLM)与开发者现有的知识和系统相结合,以提供更智能化的服务。具体来说,LangChain 可以帮助开发者f轻松地管理与 LLM 的交互,将多个组件链接在一起,并集成额外的资源,例如 API 和数据库。

    此外,LangChain 还提供了多种工具、组件和接口,以简化创建由 LLM 提供支持的应用程序的过程。例如,它提供了模型输入/输出管理工具,用于管理 LLM 及其输入和格式化输出;Prompt 模板支持自定义 Prompt 工程的快速实现以及和 LLMs 的对接;Utils 组件提供了大模型常见的植入能力的封装,如搜索引擎、Python 编译器、Bash 编译器、数据库等;Chains 组件则提供了大模型针对一系列任务的顺序执行逻辑链。

    因此,LangChain 的应用场景非常广泛,可以应用于自然语言处理、聊天机器人、智能客服、文本生成等多个领域。同时,由于其开源的特性,LangChain 也可以被广大开发者用于定制开发自己的 LLM 应用,从而推动人工智能技术的不断发展和创新。

    LangChain 可观测性体系

    LangChain 本身并没有提供可观测性能力,但可以通过集成其他工具来实现对 LangChain 应用的可观测性,以此来提高大模型的可解释性和可控性。

    • 日志:收集、存储和分析 LangChain 应用的日志数据。通过日志数据,可以监控应用程序的运行状态、错误日志、用户行为等信息。
    • 指标:使用监控工具收集 LangChain 应用的性能指标,如响应时间、吞吐量、错误率等。这些指标可以帮助你了解应用程序的性能瓶颈和潜在问题,并及时进行优化和调整。
    • 组件依赖:监控 LangChain 应用所依赖的服务或组件,如数据库、API接口等。确保这些依赖服务的可用性和性能稳定,以避免因依赖问题导致的应用故障。
    • 自定义监控:根据具体需求,可以编写自定义的监控脚本或插件,用于监控特定的业务逻辑或功能。例如,你可以编写脚本监控某个特定任务的执行情况、API 接口的调用频率等。
    • 告警与通知:设置告警规则,当监控数据达到预设的阈值时,触发告警并通知相关人员。这可以帮助你及时发现并处理潜在问题,确保应用程序的稳定运行。

    需要注意的是,可观测性方案的选择和实施应根据具体的应用场景和需求进行定制。

    LangChain 可观测性接入方案

    准备工作

    安装 DataKit

    token 在「观测云控制台」-「集成」-「Datakit」 上面获取。

    DK_DATAWAY="https://openway.guance.com?token=tkn_xxxxxx" bash -c "$(curl -L https://static.guance.com/datakit/install.sh)" 
    

    DataKit 采集器配置

    • 开启 DDTrace 采集器

    DDTrace 采集器用于采集链路信息,进入到 DataKit 安装目录下,执行 conf.d/ddtrace/,复制 ddtrace.conf.sample 并重命名为 ddtrace.conf

    • 开启 StatsD 采集器

    StatsD 采集器用于采集指标信息,默认端口为 8125

    • 重启 DataKit
    systemctl restart datakit
    

    准备 OpenAI API Key

    在 Langchain 应用中使用 OpenAI 的模型,需要一个 API Key,点击 https://platform.openai.com/account/api-keys 创建你的 API Key。

    模拟 LLM 应用并接入观测云

    主要实现了一个基于 OpenAI 的 GPT-3.5 模型的聊天对话系统,并设置了一个函数来跟踪 API 令牌的使用情况。

    • 安装 Python 依赖
    pip install langchain
    pip install openai
    pip install ddtrace
    
    • 创建 LLM

    使用 Langchain 创建 LLM,模型选择 gpt-3.5

    llm = ChatOpenAI(model='gpt-3.5-turbo',temperature=0)
    
    • 创建 ConversationChain 对象

    创建一个 ConversationChain 对象,传入之前创建的 llm 对象,设置 verbose 为 True(表示输出详细的调试信息),并使用 ConversationBufferMemory 作为记忆储存。

    conversation = ConversationChain(
        llm=llm, verbose=True, memory=ConversationBufferMemory()
    )
    
    • 创建对话函数

    定义一个函数 track_tokens_usage ,它接受一个对话链和一个查询作为参数。

    def track_tokens_usage(chain, query):
        with get_openai_callback() as cb:
            result = chain.run(query)
            print(f"prompt token {cb.prompt_tokens}, res token {cb.completion_tokens}, total token {cb.total_tokens}")
            print(f"successful requests {cb.successful_requests}, total cost {cb.total_cost}USD")
            print(result)
        return result
    

    通过回调函数 get_openai_callback 创建一个回调对象 cb ,并通过对象 cb 可以获取请求相关信息:请求数、消耗金额、令牌数量等。

    • 调用
    track_tokens_usage(conversation, "介绍一下观测云可观测性的产品及优势")
    

    由于执行完成后,进程会退出,添加以下代码阻止退出,方便相关数据及时上报。

    print("脚本执行完成,按 Enter 键退出...")
    input()
    
    • 脚本全文

    脚本名称:langchain_demo.py

    from langchain.chat_models import ChatOpenAI
    from langchain.chains import ConversationChain
    from langchain.memory import ConversationBufferMemory
    
    from langchain.callbacks import get_openai_callback
    
    
    import os
    #os.environ["http_proxy"] = "http://localhost:7890"
    #os.environ["https_proxy"] = "http://localhost:7890"
    
    os.environ["OPENAI_API_BASE"] = 'https://chatapi.a3e.top/v1'
    
    os.environ["OPENAI_API_KEY"] = '<key>'
    
    llm = ChatOpenAI(model='gpt-3.5-turbo',temperature=0)
    
    ## verbose=True会答应类似debug信息
    conversation = ConversationChain(
        llm=llm, verbose=True, memory=ConversationBufferMemory()
    )
    
    def track_tokens_usage(chain, query):
        with get_openai_callback() as cb:
            result = chain.run(query)
            print(f"prompt token {cb.prompt_tokens}, res token {cb.completion_tokens}, total token {cb.total_tokens}")
            print(f"successful requests {cb.successful_requests}, total cost {cb.total_cost}USD")
            print(result)
        return result
    
    track_tokens_usage(conversation, "介绍一下观测云可观测性的产品及优势")
    
    print("脚本执行完成,按Enter键退出...")
    input()
    
    • 执行脚本
    DD_SERVICE="my-langchain" DD_ENV="dev" DD_AGENT_HOST="localhost" DD_AGENT_PORT="9529" ddtrace-run python3 langchain_demo.py 
    

    LangChain 可观测性实践效果

    • 指标

    • 调用链

    常见问题

    Token 异常

    如果没有配置 Token,则会报错:

    pydantic.v1.error_wrappers.ValidationError: 1 validation error for ChatOpenAI
    __root__
      Did not find openai_api_key, please add an environment variable `OPENAI_API_KEY` which contains it, or pass `openai_api_key` as a named parameter. (type=value_error)
    

    关于 Token,可以在代码中定义:

    import os
    os.environ["OPENAI_API_KEY"] = 'sk-...DcE2'
    

    但不建议那么做,可以使用环境变量方式:

    OPENAI_API_KEY="sk-...DcE2"

    余额不足

    openai.RateLimitError: Error code: 429 - {'error': {'message': 'You exceeded your current quota, please check your plan and billing details. For more information on this error, read the docs: https://platform.openai.com/docs/guides/error-codes/api-errors.', 'type': 'insufficient_quota', 'param': None, 'code': 'insufficient_quota'}}
    

    代理

    如需配置代理,可以加入以下配置:

    import os
    os.environ["http_proxy"] = "http://localhost:7890"
    os.environ["https_proxy"] = "http://localhost:7890"
    

    其他第三方 API

    如果需要更换第三方 API 地址,可以使用以下方式:

    import os
    os.environ["OPENAI_API_BASE"] = 'https://chatapi.a3e.top/v1'
    

    参考资料

    LangChain 集成文档

    联系我们

    加入社区

    微信扫码
    加入官方交流群

    立即体验

    在线开通,按量计费,真正的云服务!

    免费开启

    支持私有云环境部署

    代码托管平台