metalm-xclient 0.0.9b0

雪浪模型推理服务的客户端

LLMOps-Xclient

Xclient客户端是一个用户友好的工具，旨在与Triton推理服务器毫不费力地通信。它为您管理技术细节，让您专注于您的数据和您的目标实现的结果。以下是它的帮助:

• 获取模型配置:客户端从服务器检索关于模型的详细信息，例如输入和输出数据张量的形状和名称。这一步对于正确准备数据和解释响应是至关重要的。这个功能被封装在ModelClient类中。

• 发送请求:利用模型信息，客户端通过将传递给infer_sample或infer_batch方法的参数映射到模型输入来生成推理请求。它将您的数据发送到Triton服务器，请求模型执行推理。参数可以作为位置参数或关键字参数传递(不允许混合它们)，其余的由客户端处理。

• 返回响应:然后将模型的响应返回给您。它将输入解码为numpy数组，并将模型输出映射到从infer_sample或infer_batch方法返回给您的字典元素。如果批处理维度是由客户端添加的，它还会删除该维度。

由于获取模型配置的额外步骤，此过程可能会引入一些延迟。但是，您可以通过为多个请求重用Xclient客户端，或者使用预加载的模型配置(如果有的话)对其进行设置，从而将其最小化。

Xclient包括五个专门的基础客户端，以满足不同的需求:

• ModelClient:用于简单请求-响应操作的直接的同步客户端。
• FuturesModelClient:一个多线程客户端，可以并行处理多个请求，加快操作速度。
• DecoupledModelClient:为解耦模型设计的同步客户端，它允许与Triton服务器进行灵活的交互模式。
• AsyncioModelClient:一个异步客户端，可以很好地与Python的asyncio一起工作，以实现高效的并发操作。
• AsyncioDecoupledModelClient:一个与异步兼容的客户端，专门用于异步处理解耦模型。

三个专门为MetaLM 服务的高级客户端：

• ChatMetaLM：专门用于大模型对话，可以流式返回。
• MetaLMEmbeddings：专门用于嵌入的客户端，支持稠密嵌入和稀疏嵌入。
• MetaLMRerank：专门用于重排的客户端。

Xclient客户机使用来自Triton的tritonclient包。它是Triton Inference Server的Python客户端库。它提供了使用HTTP或gRPC协议与服务器通信的低级API。Xclient客户端构建在tritonclient之上，并提供用于与服务器通信的高级API。并不是所有tritonclient的特性都可以在Xclient客户端中使用。如果需要对与服务器的通信进行更多的控制，可以直接使用tritonclient。

三个高级客户端

ChatMetaLM

同步模式

from Xclient.llms import ChatMetaLM
llm = ChatMetaLM(server_url="10.88.36.58:8201", model_name="Qwen2-0.5B-Instruct")
result = llm.invoke("介绍一下你自己")
print(result)

流模式

from Xclient.llms import ChatMetaLM
llm = ChatMetaLM(server_url="10.88.36.58:8201", model_name="Qwen2-0.5B-Instruct")
for token in llm.stream("介绍一下你自己"):
    print(token)

提早终止

from Xclient.llms import ChatMetaLM
llm = ChatMetaLM(server_url="10.88.36.58:8201", model_name="Qwen2-0.5B-Instruct",stop=['。'])
for token in llm.stream("介绍一下你自己"):
    print(token)

MetaLMEmbeddings

from Xclient import MetaLMEmbeddings

xlembed = MetaLMEmbeddings(model="bge-m3-v",base_url="http://10.88.36.58:8200")

text = ['asdasda','asdwrfa']

res= xlembed.embed_query('asdasda')
print(res)

res= xlembed.embed_documents(text)
print(res)
#附带稀疏向量
res= xlembed.embed_documents_sparse(text)
print(res)

res= xlembed.embed_query_sparse('asdasda')
print(res)

MetaLMRerank

from Xclient import MetaLMRerank
import os
from typing import List


from langchain_core.documents import Document


class CharacterTextSplitter:
    def __init__(self, chunk_size: int):
        self.chunk_size = chunk_size

    def create_documents(self, text: str) -> List[Document]:
        words = text.split(',')
        chunks = []
        for i in range(0, len(words), self.chunk_size):
            chunk = " ".join(words[i : i + self.chunk_size])
            chunks.append(Document(page_content=chunk))
        return chunks
splitter = CharacterTextSplitter(1)
documents = splitter.create_documents("测试程序1,测试程序2,测试程序3,测试程序4,测试程序5,测试程序6,测试程序7",)
query = '测试程序5'

rerank = MetaLMRerank(model="bge-reranker-v2-m3",base_url="http://10.88.36.58:8200")
result_docs = rerank.compress_documents(documents=documents, query=query)
print(result_docs)

4种基础模式

import numpy as np

from Xclient.client import ModelClient,FuturesModelClient,DecoupledModelClient,AsyncioModelClient

sample = np.array(
    ['你从哪里来，要到哪里去'.encode("utf-8")], dtype=np.object_
)
sample2 = np.array([
    ['你从哪里来，要到哪里去'.encode("utf-8")],
    ['你从哪里来，要到哪里去.........'.encode("utf-8")]
], dtype=np.object_
)

同步模式

with ModelClient("grpc://10.88.36.58:8201", "bge_large_zh","2") as client:
    print(client.model_config)
    res = client.infer_sample(sample)
    print(res)
    res = client.infer_batch(sample2)
    print(res)

并发模式，不等待

with FuturesModelClient("grpc://10.88.36.58:8201", "bge_large_zh","2") as client:
    res = client.infer_sample(sample)
print(res.result())

异步模式

#async 暂时可以不用

解耦模式，流

import numpy as np
from Xclient import AsyncioDecoupledModelClient

async def main():
    client = AsyncioDecoupledModelClient("grpc://10.88.36.58:8201", "Qwen2-0.5B-Instruct")
    async for answer in client.infer_sample(np.array(["I'm Pickle Rick".encode('utf-8')])):
        print(answer)
    await client.close()

# Run the code as a coroutine using asyncio.run()
import asyncio
loop = asyncio.get_event_loop()
loop.run_until_complete(main())