DOCSEARCHbyAlgolia
Submit Docs

iFlytek Spark AI Python SDK

repository·main·Indexed 43 days ago

https://github.com/iflytek/spark-ai-python

Official Python SDK (v0.4.5) for iFlytek's Spark (Xinghuo) Large Language Models. It provides an OpenAI-compatible interface for chat completions, streaming responses, and image understanding. The SDK includes integrations for LlamaIndex and AutoGen, support for function calling, and embedding functions compatible with vector databases like ChromaDB.

Tokens
6.2K
Snippets
16
Records
18
Agent score
55%

What's inside Spark AI

  1. Handle streaming responses via Callbacks

    main

    To process tokens as they are generated, set streaming=True and provide a callback handler to the generate method. You can implement custom logic by inheriting from BaseCallbackHandler and overriding on_llm_new_token.

    on_llm_new_token receives the token and a kwargs dictionary. The kwargs dictionary contains llm_output (including token usage) and a final flag indicating if it is the last chunk.

    from sparkai.core.callbacks import BaseCallbackHandler
from typing import Any, Optional

class ChunkPrintHandler(BaseCallbackHandler):
    """Callback Handler that prints to std out."""
    def __init__(self, color: Optional[str] = None) -> None:
        self.color = color

    def on_llm_new_token(self, token: str, *, chunk: None = None, **kwargs: Any):
        print(token)
        # kwargs contains llm_output with usage info and 'final' status

  2. Run Spark API as a local OpenAI-compatible proxy

    main

    For debugging or integrating with third-party frameworks that expect an OpenAI-style API, you can run a local proxy. This converts Spark API calls into the OpenAI format.

    1. Run the proxy module:
      python -m sparkai.spark_proxy.main
    2. The server will start (defaulting to http://0.0.0.0:8008).
    3. Configure your OpenAI client with:
      • openai_base_url: http://<your-local-ip>:8008
      • open_api_key: A string in the format key&secret&appid.
    python -m sparkai.spark_proxy.main

  3. Integrate Spark AI with AutoGen

    main

    Spark AI can be used as a native model client within Microsoft's AutoGen framework. To do this, you must register the SparkAI class as a model client and provide a configuration list in JSON format.

    Configuration (sparkai_autogen.json):

    • api_key: A combined string in the format <spark_api_key>&<spark_api_secret>&<spark_app_id>.
    • base_url: The WebSocket endpoint (e.g., wss://spark-api.xf-yun.com/v3.5/chat).
    • model_client_cls: Must be set to "SparkAI".
    • model: The specific Spark model domain (e.g., "generalv3.5").

    Usage:

    1. Load the config using autogen.config_list_from_json with a filter for SparkAI.
    2. Register the client using assistant.register_model_client(model_client_cls=SparkAI).
    from sparkai.frameworks.autogen import SparkAI
import autogen
from autogen.agentchat.contrib.retrieve_assistant_agent import RetrieveAssistantAgent

spark_config = autogen.config_list_from_json(
    "sparkai_autogen.json",
    filter_dict={"model_client_cls": ["SparkAI"]},
)

llm_config = {
    "timeout": 600,
    "cache_seed": None,
    "config_list": spark_config,
    "temperature": 0,
}

assistant = RetrieveAssistantAgent(
    name="assistant",
    system_message="You are a helpful assistant.",
    llm_config=llm_config
)
assistant.register_model_client(model_client_cls=SparkAI)

  4. Install spark_ai_python via pip

    main

    Install the spark_ai_python SDK using pip. The project requires Python 3.8+.

    To install the latest version from PyPI:

    pip install --upgrade spark_ai_python

    If you are in China and need to use a mirror (e.g., Tsinghua):

    pip install -i https://pypi.tuna.tsinghua.edu.cn/simple spark_ai_python

    To upgrade to the latest version if the standard mirror is unavailable, use the iFlytek repository:

    pip install -i https://repo.model.xfyun.cn/api/packages/administrator/pypi/simple spark_ai_python --upgrade

    To install from source (for development):

    pip install -e .
    pip install --upgrade spark_ai_python

  5. Integrate Spark AI with LlamaIndex

    main

    You can use Spark AI as both an embedding model and an LLM within the LlamaIndex framework. This allows you to build RAG (Retrieval-Augmented Generation) applications using Spark AI's capabilities.

    Requirements:

    • Install the Chroma vector store extension: pip install llama-index-vector-stores-chroma.
    • Configure both SparkAiEmbeddingModel and SparkAI (for LLM) using your Spark AI credentials.
    • Set SPARKAI_Embedding_DOMAIN to either query or para as required by the API.

    Key Classes:

    • SparkAiEmbeddingModel: Used for generating embeddings.
    • SparkAI: Used as the LLM provider in LlamaIndex.
    from sparkai.embedding.sparkai_base import SparkAiEmbeddingModel
from sparkai.frameworks.llama_index import SparkAI
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
import os

# 1. Setup Embedding Model
embed_model = SparkAiEmbeddingModel(
    spark_embedding_app_id=os.environ['SPARK_Embedding_APP_ID'],
    spark_embedding_api_key=os.environ['SPARK_Embedding_API_KEY'],
    spark_embedding_api_secret=os.environ['SPARK_Embedding_API_SECRET'],
    spark_embedding_domain=os.environ['SPARKAI_Embedding_DOMAIN'],
    qps=2
)

# 2. Setup LLM
sparkai = SparkAI(
    spark_api_url=os.environ["SPARKAI_URL"],
    spark_app_id=os.environ["SPARKAI_APP_ID"],
    spark_api_key=os.environ["SPARKAI_API_KEY"],
    spark_api_secret=os.environ["SPARKAI_API_SECRET"],
    spark_llm_domain=os.environ["SPARKAI_DOMAIN"],
    streaming=False,
)

# 3. Use in LlamaIndex
chroma_client = chromadb.Client()
chroma_collection = chroma_client.get_or_create_collection(name="spark")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# index creation and querying
index = VectorStoreIndex.from_documents(documents, storage_context=storage_context, embed_model=embed_model)
query_engine = index.as_query_engine(llm=sparkai, similarity_top_k=2)
response = query_engine.query("What did the author do growing up?")

  6. Prerequisites for using Spark AI

    main

    Before using the spark-ai-python library, you must obtain the following credentials from xfyun.cn:

    • app_id
    • api_key
    • api_secret

    Additionally, ensure you have configured your environment variables. It is recommended to copy .env.example to .env and populate it with your credentials.

  7. Access Spark AI using OpenAI-style HTTP interface

    main

    The Spark AI interface supports an OpenAI-compatible ChatCompletion style for streaming responses. This allows you to use a familiar client pattern to interact with Spark models.

    Key features of this interface include:

    • Model Selection: Specify the Spark domain via the model parameter.
    • Streaming: Set stream=True to receive incremental responses.
    • Advanced Parameters: Use extra_body to pass Spark-specific advanced parameters (e.g., patch_id).
    • Control Parameters: Supports standard parameters like top_p and seed.
    def test_spark_openai_style_stream():
    # 使用客户端创建一个 completion
    completion = client.chat.completions.create(
        model="星火的domain",
        messages=[
            {"role": "system", "content": "帮我根据用户意图生成如下Json格式响应: {\"intent\": \"用户意图\"} \n可选的意图有: 天气查询，航班查询，知识库查询、订餐查询、火车票查询",},
            {"role": "user", "content": "今天天气如何?"},
        ],
        user="",
        stream=True,
        seed=2, 
        top_p=0.1,
        #tools=[ {"type":"function", "function": function} for function in tools],
       # tool_choice={"type":"function", "function": tools[1]},
        extra_body={
            # 星火支持的高级参数传入
            "patch_id": ["0"]

        },
        extra_headers={})
    print(completion)
    for response in completion:
       print(response)

  8. Configure Advanced Parameters and Patch IDs

    main

    You can pass advanced configuration to the ChatSparkLLM client using the model_kwargs dictionary. This is useful for enabling/disabling specific features or using specialized models.

    Common use cases:

    • General Parameters: Pass keys like search_disable to control model behavior.
    • Patch IDs: For LoRA or small-package type inference, pass a patch_id inside model_kwargs.

    Example:

    spark = ChatSparkLLM(
    spark_app_id=os.environ["SPARKAI_APP_ID"],
    spark_api_key=os.environ["SPARKAI_API_KEY"],
    spark_api_secret=os.environ["SPARKAI_API_SECRET"],
    spark_llm_domain="xspark13b6k",
    model_kwargs={"patch_id": "210267877777408"},
)

  10. Generate non-streaming completions with ChatSparkLLM

    main

    To receive a single, complete response from the model, initialize ChatSparkLLM with streaming=False. Note that when streaming is set to False, callbacks will not be triggered.

    Use spark.generate([messages]) to obtain the result. The response object contains generations (the text content) and llm_output (metadata like token usage).

    import os
from sparkai.llm.llm import ChatSparkLLM
from sparkai.core.messages import ChatMessage
from dotenv import load_dotenv

load_dotenv()

if __name__ == '__main__':
    spark = ChatSparkLLM(
        spark_api_url=os.environ["SPARKAI_URL"],
        spark_app_id=os.environ["SPARKAI_APP_ID"],
        spark_api_key=os.environ["SPARKAI_API_KEY"],
        spark_api_secret=os.environ["SPARKAI_API_SECRET"],
        spark_llm_domain=os.environ["SPARKAI_DOMAIN"],
        streaming=False,
    )
    messages = [ChatMessage(role="user", content='你好呀')]
    a = spark.generate([messages])
    print(a.generations[0][0].text)
    print(a.llm_output)

  11. Use Image Understanding with ChatSparkLLM

    main

    Spark AI supports image understanding. To use this feature, initialize ChatSparkLLM with spark_llm_domain="image" and provide messages containing both text and base64 encoded image content.

    Key Requirements:

    • spark_llm_domain must be set to "image".
    • Use ImageChatMessage to wrap image data with content_type="image" and text data with content_type="text".
    import base64
import os
from sparkai.models import ChatSparkLLM, ImageChatMessage

image_content = base64.b64encode(open("spark_llama_index.png",'rb').read())

spark = ChatSparkLLM(
    spark_app_id=os.environ["SPARKAI_APP_ID"],
    spark_api_key=os.environ["SPARKAI_API_KEY"],
    spark_api_secret=os.environ["SPARKAI_API_SECRET"],
    spark_llm_domain="image",
    streaming=False,
    user_agent="test"
)

messages = [
    ImageChatMessage(role="user", content=image_content, content_type="image"),
    ImageChatMessage(role="user", content="这是什么图", content_type="text")
]

a = spark.generate([messages], callbacks=[])

  12. Use stream and astream interfaces for streaming

    main

    Alternatively, you can use the generator-based interfaces for streaming:

    1. Synchronous stream: Returns a generator that you can iterate over.
    2. Asynchronous astream: Returns an async generator for use in asyncio environments. You can pass config={"callbacks": [handler]} to astream to use callbacks in an async context.
    # Synchronous stream
a = spark.stream(messages)
for message in a:
    print(message)

# Asynchronous astream
async def test_astream():
    handler = AsyncChunkPrintHandler()
    a = spark.astream(messages, config={"callbacks": [handler]})
    async for message in a:
        print(message)