开源模型应用落地-LangChain实用小技巧-获取token消耗（五）

news2025/2/26 20:54:32

一、前言

在当今的自然语言处理领域，LangChain 框架因其强大的功能和灵活性而备受关注。掌握一些实用的小技巧，能够让您在使用 LangChain 框架时更加得心应手，从而更高效地开发出优质的自然语言处理应用。

计算 Token 消耗对有效管理和优化语言模型使用过程意义重大

帮助实现成本控制，因为语言模型服务多按 Token 消耗计费，通过计算可精准规划预算和控制开支，比如让小型企业明确每月模型使用投入并调整业务策略。
有助于性能优化，了解不同输入和操作的 Token 消耗能发现高消耗部分，进而针对性优化，像生成大量文本任务中可据此调整输入方式或生成逻辑。
用于资源分配，在有限资源下根据消耗合理分配计算资源，比如为不同应用场景按需分配。此外，还能作为评估不同模型或版本性能的指标，成本低且效果好的模型更具优势。
在面向用户的应用中，分析 Token 消耗模式能更好理解用户行为和需求，比如智能客服中能发现需完善解答的特定问题。

二、术语

2.1.LangChain

是一个全方位的、基于大语言模型这种预测能力的应用开发工具。LangChain的预构建链功能，就像乐高积木一样，无论你是新手还是经验丰富的开发者，都可以选择适合自己的部分快速构建项目。对于希望进行更深入工作的开发者，LangChain 提供的模块化组件则允许你根据自己的需求定制和创建应用中的功能链条。

LangChain本质上就是对各种大模型提供的API的套壳，是为了方便我们使用这些 API，搭建起来的一些框架、模块和接口。

LangChain的主要特性：
1.可以连接多种数据源，比如网页链接、本地PDF文件、向量数据库等
2.允许语言模型与其环境交互
3.封装了Model I/O（输入/输出）、Retrieval（检索器）、Memory（记忆）、Agents（决策和调度）等核心组件
4.可以使用链的方式组装这些组件，以便最好地完成特定用例。
5.围绕以上设计原则，LangChain解决了现在开发人工智能应用的一些切实痛点。

2.2.get_openai_callback

为开发者提供了一种直观且有效的方式来了解和监控与AI模型交互时的资源消耗情况，有助于优化应用程序的性能和成本管理。它能够帮助开发者更好地把握模型调用的资源利用情况，从而做出更合理的决策和优化措施。

2.3.LCEL

是一种用于构建复杂链式组件的语言。它提供了一些非常实用的功能,比如流式处理、并行化和日志记录。它的使用方式很简单:只需要使用 "|" 符号将不同的组件连接起来,就可以形成一个链式结构。这个 "|" 符号就像是Unix中的管道操作符,可以将一个组件的输出转换成下一个组件的输入,从而实现数据的传递和处理。

三、前提条件

3.1. 基础环境

操作系统：不限

3.2. 安装虚拟环境

conda create --name langchain python=3.10
conda activate langchain
pip install langchain langchain-openai langchain-community

四、技术实现

4.1.非流式输出示例

import os

from langchain.chains.conversation.base import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain_community.callbacks import get_openai_callback
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

if __name__ == '__main__':
    memory = ConversationBufferMemory()

    memory.save_context({"input": "hi，你好"}, {"output": "你好！有什么我可以帮助你的吗？"})
    memory.save_context({"input": "我家在广州，很好玩哦"}, {"output": "广州是一个美丽的城市，有很多有趣的地方可以去。"})

    llm = ChatOpenAI(model_name='gpt-3.5-turbo-1106', temperature=0.1, max_tokens=512)

    template = ''' 仅根据以下上下文回答问题:
    {history}

    问题: {input}
    '''
    prompt = ChatPromptTemplate.from_template(template)

    # 创建一个对话链
    conversation = ConversationChain(
        llm=llm,
        memory=memory,
        prompt=prompt,
        verbose=True
    )

    with get_openai_callback() as cb:
        print(conversation.predict(input="我家在哪里？"))
        print(cb)

调用结果：

4.2.流式输出示例

import os
import tiktoken
from langchain.memory import ConversationBufferMemory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

if __name__ == '__main__':
    memory = ConversationBufferMemory()

    memory.save_context({"input": "hi，你好"}, {"output": "你好！有什么我可以帮助你的吗？"})
    memory.save_context({"input": "我家在广州，很好玩哦"}, {"output": "广州是一个美丽的城市，有很多有趣的地方可以去。"})
    model = 'gpt-3.5-turbo-1106'
    llm = ChatOpenAI(model_name=model, temperature=0.1, max_tokens=512, streaming=True)

    template = ''' 仅根据以下上下文回答问题:
    {history}

    问题: {input}
    '''
    prompt = ChatPromptTemplate.from_template(template)
    enc = tiktoken.encoding_for_model(model)
    history = memory.load_memory_variables({})['history']
    input = "我家在哪里？"

    chain = ({
                "input": RunnablePassthrough(),
                "history":  RunnableLambda(memory.load_memory_variables)
            } | prompt | llm | StrOutputParser())


    result = []
    for chunk in chain.stream({input:input}):
        print(chunk, end='', flush=True)
        result.append(chunk)

    response = "".join(result)

    prompt_str = template.replace('{history}', history).replace('{input}', input)
    prompt_tokens = len(enc.encode(prompt_str))
    completion_tokens = len(enc.encode(response))

    total_tokens = f'''
Tokens Used: {prompt_tokens+completion_tokens}
  Prompt Tokens: {prompt_tokens}
  Completion Tokens: {completion_tokens}
    '''
    print(total_tokens)

调用结果：