使用在线的DeepSeek模型

安装LiteLLM依赖库

pip install llama-index-llms-litellm -i https://mirrors.aliyun.com/pypi/simple/

完整代码

from llama_index.llms.litellm import LiteLLM
from llama_index.core.llms import ChatMessage

llm = LiteLLM(
    model="deepseek/deepseek-chat",
    api_base="https://api.deepseek.com",
    api_key="sk-xxxx",
)

message = ChatMessage(role="user", content="你是谁？")

chat_response = llm.chat([message])

print(chat_response)

运行结果：

assistant: 我是DeepSeek Chat，由深度求索公司（DeepSeek）创造的智能AI助手！🤖✨ 我的使命是帮助你解答各种问题，无论是学习、工作，还是日常生活中的小困惑，我都会尽力提供准确、有用的信息。  

使用在线的Embeddings模型

阿里云百炼平台文本向量模型：百炼控制台

安装 DashScope SDK

pip install dashscope -i https://mirrors.aliyun.com/pypi/simple/

• DashScope 文档: https://help.aliyun.com/zh/dashscope/developer-reference/quick-start

创建自定义 Embedding 类

在 LlamaIndex 中，你可以通过继承 TextNodeEmbeddingEngine 或使用 BaseEmbedding 创建自定义的 Embedding 类。

完整代码：

from typing import List, Optional

import dashscope
from llama_index.core.embeddings import BaseEmbedding
from pydantic import Field
from dashscope import TextEmbedding

class QwenAliyunEmbedding(BaseEmbedding):

    # model_name: Optional[str] = None  # 可选模型名
    # api_key: Optional[str] = None  # 支持传入 api_key
    # dimension: Optional[int] = None  # 默认 v1 是 768，v2 是 1536

    def __init__(self, model_name: Optional[str] = None, api_key: Optional[str] = None, dimension: Optional[int] = None, **kwargs):
        super().__init__(model_name=model_name, api_key=api_key,dimension=dimension, **kwargs)
        if api_key:
            dashscope.api_key = api_key

            # import os
            # os.environ["DASHSCOPE_API_KEY"] = api_key

    def _get_query_embedding(self, query: str) -> List[float]:
        response = TextEmbedding.call(
            model=self.model_name,
            input=query
        )
        return list(response.output["embeddings"][0]["embedding"])

    def _get_text_embedding(self, text: str) -> List[float]:
        response = TextEmbedding.call(
            model=self.model_name,
            input=text,
        )
        return list(response.output["embeddings"][0]["embedding"])

    def _get_text_embeddings(self, texts: List[str]) -> List[List[float]]:
        return [self._get_text_embedding(text) for text in texts]

    async def _aget_query_embedding(self, query: str) -> List[float]:
        return self._get_query_embedding(query)

    async def _aget_text_embedding(self, text: str) -> List[float]:
        return self._get_text_embedding(text)

    async def _aget_text_embeddings(self, texts: List[str]) -> List[List[float]]:
        return self._get_text_embeddings(texts)

测试：

from Qwen.QwenAliyunEmbedding import QwenAliyunEmbedding

embed_model = QwenAliyunEmbedding(model_name="text-embedding-v3", api_key="sk-xxxx", dimension=64)

embeddings = embed_model.get_text_embedding("测试。。")

print(len(embeddings))