MaxCompute:Função de IA do MaxFrame

Visão geral

A função de IA do MaxFrame é uma solução ponta a ponta para inferência offline de modelos de linguagem de grande porte (LLM) na plataforma Alibaba Cloud MaxCompute. Ela integra perfeitamente o processamento de dados às capacidades de IA, simplificando a adoção de modelos de linguagem corporativos.

• Filosofia de design: "Dados entram, resultados saem." Esse conceito permite usar o framework Python MaxFrame e APIs no estilo pandas para concluir todo o fluxo de trabalho — desde a preparação e o processamento de dados até a inferência do modelo e o armazenamento dos resultados — inteiramente dentro do ecossistema MaxCompute.

• Casos de uso: Esta função é ideal para processar grandes volumes de dados estruturados, como em análises de logs e registros de comportamento de usuários, além de dados não estruturados para tarefas como tradução de texto, resumo de documentos e vetorização. É possível processar dados na escala de petabytes em um único job, obtendo baixa latência e escalabilidade linear por meio da arquitetura de computação distribuída. Utilize-a para extrair informações estruturadas de textos, organizar e resumir conteúdos, gerar resumos, traduzir idiomas, avaliar a qualidade textual e classificar sentimentos. Isso simplifica significativamente o processamento de dados para modelos de linguagem e melhora a qualidade dos resultados.

• Arquitetura

  A função de IA do MaxFrame oferece uma interface generate flexível e de uso geral, além de interfaces task concisas e específicas para cenários como tradução de texto, extração de dados estruturados e vetorização. Selecione um modelo e forneça uma tabela do MaxCompute e prompts como entrada.

  Ao chamar uma interface, o MaxFrame executa primeiro o particionamento dos dados na tabela de entrada. Defina um nível adequado de concorrência com base no volume de dados e inicie um grupo de workers para executar o job de computação. Cada worker utiliza o modelo de prompt fornecido para renderizar e construir as entradas do modelo a partir das linhas de dados, executa o job de inferência e grava os resultados e o status de sucesso de volta no MaxCompute.

  A figura a seguir ilustra a arquitetura e o fluxo de trabalho.

  

• Principais vantagens:

  • Facilidade de uso: APIs Python familiares, biblioteca de modelos pronta para uso e custo zero de implantação.

  • Escalabilidade: Aproveita recursos de cota CU, cota GU e Inference Quota do MaxCompute para suportar processamento paralelo em larga escala e aumentar o throughput geral de tokens.

  • Integração de dados e IA: Conclua todo o fluxo de trabalho, desde a leitura e o processamento de dados até a inferência de IA e o armazenamento de resultados, em uma única plataforma. Essa abordagem reduz custos de migração de dados e aumenta a eficiência do desenvolvimento.

  • Ampla cobertura de cenários: Abrange mais de 10 casos de uso comuns, incluindo tradução, extração de dados estruturados e vetorização.

Requisitos

• Regiões suportadas:

  China (Hangzhou), China (Shanghai), China (Beijing), China (Ulanqab), China (Shenzhen), China (Chengdu), China (Hong Kong), China (Hangzhou) Finance Cloud e China (Shanghai) Finance Cloud.

• Versão do Python suportada: 3.11.

• Versão do SDK suportada: Certifique-se de que sua versão do MaxFrame SDK seja 2.7.1 ou posterior. Verifique a versão executando um dos seguintes comandos:

  // For Windows
  pip list | findstr maxframe
  
  // For Linux
  pip list | grep maxframe

  Caso sua versão esteja desatualizada, execute pip install --upgrade maxframe para atualizar para a versão mais recente.

• Instale o cliente MaxFrame mais recente.

Modelos suportados

O MaxFrame oferece suporte pronto para uso a uma série de modelos de linguagem de grande porte integrados, incluindo Qwen 3, DeepSeek-R1-Distill-Qwen e Qwen3-embedding. Também permite chamar modelos comerciais principais do Model Studio, como os modelos multimodais qwen3.7-max, qwen3.6-plus, qwen3.6-flash, deepseek-v4-pro, deepseek-v4-flash e qwen3-vl-embedding, além do modelo de texto text-embedding-v4. Todos os modelos são hospedados offline na plataforma MaxCompute. Isso elimina a necessidade de gerenciar downloads de modelos, distribuição ou limites de concorrência de API. Utilize os modelos por meio de chamadas de API simples, aproveitando os vastos recursos de computação do MaxCompute para concluir tarefas de inferência offline com alto throughput geral de tokens e concorrência.

# Use CU quota compute resources.
options.session.quota_name = "mf_cpu_quota"

# Use GU quota compute resources.
options.session.gu_quota_name = "mf_gpu_quota"

Referência da API

A função de IA do MaxFrame fornece duas interfaces para equilibrar flexibilidade e facilidade de uso: a interface generate de uso geral e as interfaces task específicas para tarefas.

Casos de uso

import os
import maxframe.dataframe as md
from maxframe import new_session
from maxframe.config import options
from maxframe.udf import with_running_options
from odps import ODPS

options.dag.settings = {
    "engine_order": ["DPE", "MCSQL"]
}

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

# Initialize the MaxFrame session.
session = new_session(o)

# Print the Logview URL for the job.
print(session.get_logview_address())

# 1. Use GU quota compute resources.
options.session.gu_quota_name = "mf_gu_quota"

# 2. Use the Qwen3-1.7B model.
from maxframe.learn.contrib.llm.models.managed import ManagedTextLLM

llm = ManagedTextLLM(name="Qwen3-1.7B")

# 3. Prepare the data. You can skip this step if you already have data.
# o.execute_sql("""
#   CREATE TABLE IF NOT EXISTS raw_contracts (
#       en STRING
#   );
#   """)
# 
# o.execute_sql("""
#   INSERT INTO raw_contracts VALUES
#   ('This agreement is governed by the laws of the State of California.'),
#   ('The tenant shall pay rent on the first day of each month.'),
#   ('Either party may terminate this contract with 30 days written notice.'),
#   ('All intellectual property rights shall remain with the original owner.'),
#   ('The warranty period for this product is twelve months from the date of purchase.');
#   """)

df = md.read_odps_table("raw_contracts")

# 4. Define the prompt template.
messages = [
    {
        "role": "system",
        "content": "You are a document translation expert who can fluently translate English text into Chinese.",
    },
    {
        "role": "user",
        "content": "Translate the following English text into Chinese. Output only the translated text, with no other content.\n\n Example:\nInput: Hi\nOutput: 你好。\n\n Text to translate:\n\n{en}",
    },
]

# 5. Call the `generate` interface, define the prompt, and reference the corresponding data column.
result_df = llm.generate(
    df,
    prompt_template=messages,
    params={
        "temperature": 0.7,
        "top_p": 0.8,
    },
).execute()

# 6. Write the results to a MaxCompute table.
result_df.to_odps_table("raw_contracts_result")

import maxframe.dataframe as md
from maxframe import new_session
from maxframe.config import options
from maxframe.udf import with_running_options
from odps import ODPS

options.dag.settings = {
    "engine_order": ["DPE", "MCSQL"]
}

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

# Initialize the MaxFrame session.
session = new_session(o)

# Print the Logview URL for the job.
print(session.get_logview_address())

# 1. Use GU quota compute resources.
options.session.gu_quota_name = "mf_gu_quota"

# 2. Use the Qwen3-4B model.
from maxframe.learn.contrib.llm.models.managed import ManagedTextLLM

llm = ManagedTextLLM(name="Qwen3-4B-Instruct-2507-FP8")

df = md.read_odps_table("traditional_chinese_medicine", index_col="index")

# Specify four concurrent partitions.
parallel_partitions = 4
df = df.mf.rebalance(num_partitions=parallel_partitions)

# 3. Use the preset `extract` task interface.
result_df = llm.extract(
    df["text"],
    description="Please extract structured data from the following medical record in order. Return the final output in strict JSON format according to the schema.",
    schema=MedicalRecord,
    examples=[(example_input, example_output)],
)
result_df.execute()

import os
import maxframe.dataframe as md
from maxframe import new_session
from maxframe.config import options
from maxframe.udf import with_running_options
from odps import ODPS

options.dag.settings = {
    "engine_order": ["DPE", "MCSQL"]
}

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

# Initialize a MaxFrame session.
session = new_session(o)

# Print the Logview URL of the job.
print(session.get_logview_address())

# 1. Create a text-embedding-v4 model instance (using Inference Quota).
# Load the text-embedding-v4 model from the public model set.
from maxframe.learn.utils import read_odps_model

llm = read_odps_model("text-embedding-v4", project="bigdata_public_modelset")

# 2. Read data.
df = md.read_odps_table("user_feedback_table")

# 3. Perform vectorization.
result_df = llm.embed(
    df["query"],
    running_options={"max_tokens": 1024, "verbose": True},
    simple=True,
).execute()

# 4. Write the results to a MaxCompute table.
result_df.to_odps_table("feedback_analysis_result")

import os

import pandas as pd
import maxframe.dataframe as md
from maxframe import new_session
from maxframe.config import options
from maxframe.learn.contrib.llm import ImageContentType
from maxframe.learn.utils import read_odps_model
from odps import ODPS

# Configure the execution engine: DPE first.
options.dag.settings = {
    "engine_order": ["DPE", "MCSQL"],
}

# 1. Create an ODPS connection.
o = ODPS(
    os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
    os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
    project="<your-project-name>",
    endpoint="<your-endpoint>",  # Example: https://service.cn-shanghai.maxcompute.aliyun.com/api
)

# 2. Configure the engine and create a session.
# Model Studio VL model inference runs on the DPE engine and is billed by token consumption.
# No CU/GU resource provisioning is required.

# Create a MaxFrame session.
session = new_session(o)
print(f"Session ID: {session.session_id}")
print(f"Logview: {session.get_logview_address()}")

## 3. Load the Model Studio VL model.
# Load the pre-registered multimodal understanding model from the MaxCompute public model set.
image_llm = read_odps_model(
    "qwen3.6-plus", project="bigdata_public_modelset", odps_entry=o
)

## 4. Prepare OSS image data.
# Use the `oss:////` format and pass OSS access credentials through `storage_options`.
# OSS configuration.
OSS_ENDPOINT = "oss-cn-shanghai-internal.aliyuncs.com"  # Replace with your OSS Endpoint.
OSS_BUCKET = "your-bucket"                              # Replace with your Bucket name.
OSS_PREFIX = "images"                                   # Replace with the image path prefix.

storage_options = {
    "access_key_id": os.getenv("OSS_ACCESS_KEY_ID"),
    "access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET"),
}

# Construct the list of image URLs.
image_urls = [
    f"oss://{OSS_ENDPOINT}/{OSS_BUCKET}/{OSS_PREFIX}/image{i}.jpg"
    for i in range(1, 9)
]

df = md.DataFrame({"image_url": image_urls})
df.execute()

## 5. Call the VL model for image understanding.
# Use the `generate()` interface to pass a text prompt and OSS images to the model.

cp = image_llm.content_part

result_df = image_llm.generate(
    df,
    messages=[
        {
            "role": "user",
            "content": [
                cp.text("Describe this image in one sentence."),
                cp.image(
                    data=df.image_url,
                    type=ImageContentType.IMAGE_URL,
                    storage_options=storage_options,
                ),
            ],
        }
    ],
    simple_output=True,
    params={"max_tokens": 128},
)

## 6. Run inference and view the results.
# MaxFrame uses lazy execution. Call `.execute()` to trigger the actual computation.
result = result_df.execute()
print(result)

## 7. Clean up resources.
session.destroy()
print("Session destroyed.")

Otimização de desempenho