人工编码和部署人工智能服务

定义generate()函数的准则

generate() 功能可用于处理您的授权令牌。 该函数处理对推理端点 /ml/v4/deployments/{id_or_name}/ai_service 的REST调用。

请按照以下指南在人工智能服务中定义 generate() 功能：

• 您必须使用 generate 定义函数。
• 您只能向 generate() 函数提供一个参数： context 。
• generate() 函数必须返回数据类型为 dict （字典）的值。
• 可选 ：您可以选择指定 body 或 header 键。

示例

def generate(context):
    user_token = context.get_token()
    headers = context.get_headers()
    json_body = context.get_json()

    return {
        "headers": {
             "Content-Type": "text/plain"
        },
        "body": "Hello WatsonX"
    }

定义generate_stream()函数的准则

您可以使用 generate_stream() 功能来生成需要流式传输的AI用例。 该功能处理对服务器发送事件（SSE）推断端点 POST /ml/v4/deployments/{id_or_name}/ai_service_stream 的REST调用。

请按照以下指南在人工智能服务中定义 generate_stream() 功能：

• 您必须使用 generate_stream 定义函数。
• 您只能向 generate_stream() 函数提供一个参数： context 。

示例

def generate_stream(context):
    user_token = context.get_token()
    headers = context.get_headers()
    json_body = context.get_json()

    for x in ["Hello", "WatsonX", "!"]:
        yield x

输出

id: 1
event: message
data: 你好

id: 2
event: message
data: WatsonX

id: 3
event: message
data: !

id: 4
event: eos

如果流式传输的块中包含换行符 (\n)，则换行符后面的字符在流式响应中可能无法正确显示。 出现这种情况是因为某些流式传输实现将换行符视为消息边界。

可能的解决方法：

1. 转义换行符：

DATA = {"a": ["1\\n2", "3"]}

2. 流式传输 JSON 对象（推荐）：

DATA = {
    "a": [
        {"id": 1, "value": "Hello\nWorld"},
        {"id": 2, "value": "Foo\nBar"}
    ]
}

for chunk in DATA["a"]:
    yield chunk

流式传输结构化 JSON 对象：

• 消除由换行符分隔符引起的歧义
• 常用于现代 AI 流式传输 API

定义generate_batch()函数的指南

generate_batch() 功能可用于需要批量推理的用例。 此功能处理对作业端点 /ml/v4/deployments_jobs 的REST API调用。

请按照以下指南在人工智能服务中定义 generate_batch() 功能：

• 您必须使用 generate_batch() 定义函数。

示例

def generate_batch(input_data_references: list[dict], output_data_reference: dict):
    # context from outer function is visible
    batch_token = context.generate_token()
    print(f"batch_token: {batch_token[-5:]}", flush=True)
    print(
        f"generate_batch:\n{input_data_references=}\n{output_data_reference=}",
        flush=True,
    )

主要功能

• generate_stream()generate()对所有三种 AI 服务方法（、和）均支持异步操作 generate_batch()
• 运行时对同步和异步执行的自动检测
• 支持流式处理和批处理工作流

示例：异步 AI 服务

async def qna_with_rag_ai_service(context):
    """
    Async AI Service example with all three methods
    """
    from typing import Any, Dict

    async def stream_chunks(text):
        for ch in text:
            yield ch

    async def get_value(param):
        return param

    # Async generate() - Standard inference
    async def generate(context: Any):
        result = await get_value("Generate Flow")
        return {"body": {"message": "This is an async AI service"}}

    # Async generate_stream() - Streaming inference
    async def generate_stream(context: Any):
        result = await get_value("Generate Stream Flow")
        llm_response = "This is a dummy AI generate stream"
        async for chunk in stream_chunks(llm_response):
            yield chunk

    # Async generate_batch() - Batch jobs
    async def generate_batch(
        input_data_references: dict, output_data_reference: dict
    ) -> None:
        batch_token = context.get_token()
        print(f"batch_token: {batch_token[-5:]}")
        print(
            f"generate_batch:\n"
            f"{input_data_references=}\n"
            f"{output_data_reference=}"
        )

    return generate, generate_stream, generate_batch

在此示例中：

• 这三种方法（generate, generate_stream, generate_batch）均被定义为异步函数
• 外部函数中的对象 context 可在所有方法中访问
• 支持LLM流式响应、聊天机器人和批处理

generate() — 同步推理

generate()：向 /ml/v4/deployments/{id_or_name}/ai_service 终端发送REST API调用。 它接收一个上下文对象，提取令牌、标头和JSON正文，并根据JSON正文中的模式键返回响应。 响应格式可以是JSON、字节或字符串，并带有可选的自定义标头。

请求 JSON 可以包含任何必需的值，例如：

{ "content-type": "<value>" }

模式决定了响应格式。 请参考此示例代码，其中提供了适用于不同内容类型的简化版本，并附有简短示例：

def deployable_ai_service_f1(context, params={"k1": "v1"}):
     """
    The outer function handles the REST call to the deployment endpoint
    POST /ml/v4/deployments

        context.generate_token() - generate a user token

    To use `generate` and `generate_stream`, the deployment has to be ONLINE
    To use `generate_batch`, the deployment has to be BATCH
    """
    gen_token = context.generate_token()
    print(f"outer function: {gen_token[-5:]}", flush=True)

     def generate(context) -> dict:
        //context is mandatory parameter for ai service
        """
        The `generate` function handles the REST call to the inference endpoint
        POST /ml/v4/deployments/{id_or_name}/ai_service

            context.get_token()     - get the Bearer token from the header of the request
            context.get_json()      - get the body of the request
            context.get_headers()   - get the headers of the request

        The generate function should return a dict
        The following optional keys are supported currently
        - body
        - headers

        This particular example accepts a json body of the format:
        { "content-type" : <value> }

        Depending on the <value> of the content type, it will return different response

        """
        user_token = context.get_token()
        headers = context.get_headers()
        json_body = context.get_json()
        print(f"my_generate: {user_token=}", flush=True)
        print(f"request headers: {headers=}", flush=True)
        print(f"json body: {json_body=}", flush=True)

         match json_body.get("content-type", "no-match"):
            case "json":
                # response Content-Type is "application/json"
                return {
                    "headers": {
                        "Content-Type": "application/json",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": {
                        "user_token": user_token[-5:],
                        "gen_token": gen_token[-5:],
                        "json_body": json_body,
                        "params": params
                    },
                }

其他适用案例：

Content-Type: json-no-header:

case "json-no-header":
    return {"body": {"json_body": json_body}}

Content-Type: json-custom-header:

case "json-custom-header":
    return {
        "headers": {"Content-Type": "text/plain; charset=utf-8"},
        "body": {"json_body": json_body},
    }

内容类型：字节：

case "bytes":
    return {
        "headers": {"Content-Type": "application/octet-stream"},
        "body": b"12345678910",
    }

Content-Type: str:

case "str":
    return {
        "headers": {"Content-Type": "text/plain"},
        "body": f"Hello WatsonX: {json_body}",
    }

Content-Type: negative-str-return:

case "negative-str-return":
    return "Should give 400 bad request"

默认情况：

case _:
    return {"body": "No match"}

generate_stream() — SSE 流处理

generate_stream()：向SSE（服务器发送事件）推断端点 /ml/v4/deployments/{id_or_name}/ai_service_stream 发出REST API调用。 它接收上下文对象，提取标记、标头和JSON正文，并返回由 eos （流结束）指示的SSE事件流。

generate_stream() 使用 yield 遍历“sse”列表中的每个元素：

def generate_stream(context):
        """
        The generate_stream function handles the REST call to the SSE inference endpoint
        POST /ml/v4/deployments/{id_or_name}/ai_service_stream

            context.get_token()     - get the Bearer token from the header of the request
            context.get_json()      - get the body of the request
            context.get_headers()   - get the headers of the request

        The generate_stream function be a python `generator` with yield
        The data in yield will the "data" for the SSE event

        Example: The following request json
            { "sse": ["Hello" , "", "WatsonX"," ", "!"]}
        will return the following stream of events
            --------------
            id: 1
            event: message
            data: Hello

            id: 2
            event: message
            data:

            id: 3
            event: message
            data: WatsonX

            id: 4
            event: message
            data:

            id: 5
            event: message
            data: !

            id: 6
            event: eos
            ---------------
        The end of the stream will be marked by the event "eos"

        """
        user_token = context.get_token()
        headers = context.get_headers()
        json_body = context.get_json()
        print(f"generate_stream: {user_token=}", flush=True)
        print(f"generate_stream: {headers=}", flush=True)
        print(f"generate_stream: {json_body=}", flush=True)

        import time
        for x in json_body.get("sse", ["default", "message"]):
            time.sleep(1)
            yield x

generate_batch() — 批量处理

generate_batch()：向作业端点 /ml/v4/deployments_jobs 发出REST API调用。 它从请求的JSON正文中获取 input_data_references 和 output_data_reference ，生成批处理令牌，并记录输入和输出数据引用。

generate_batch() 支持输入和输出引用，并对其进行记录：

    def generate_batch(input_data_references: list[dict], output_data_reference: dict) -> None:
        """
        The generate_batch function handles the REST jobs endpoint
        POST /ml/v4/deployments_jobs

            Arguments to the function are from the json body of the request to jobs
            - input_data_references : scoring.input_data_references
            - output_data_reference : scoring.output_data_reference

        context.generate_token() : can access context object
        from outer function scope if token is required
        """
        batch_token = context.generate_token()
        print(f"batch_token: {batch_token[-5:]}", flush=True)
        print(
            f"generate_batch:\n{input_data_references=}\n{output_data_reference=}",
            flush=True,
        )

    return generate, generate_stream, generate_batch

使用REST API创建AI服务资产

您可以使用 /ml/v4/ai_services REST API 端点在 Watson Machine Learning 资源库中创建 AI 服务资产。

创建在线部署

以下示例展示了如何使用 watsonx.ai Python 客户端库为您的AI服务创建在线部署：

deployment_details = client.deployments.create(
    artifact_id=ai_service_id,
    meta_props={
        client.deployments.ConfigurationMetaNames.NAME: "ai-service - online test",
        client.deployments.ConfigurationMetaNames.ONLINE: {},
        client.deployments.ConfigurationMetaNames.HARDWARE_SPEC: {
            "id": client.hardware_specifications.get_id_by_name("XS")
        },
    },
)
deployment_id = client.deployments.get_uid(deployment_details)
print("The deployment id:", deployment_id)

创建批量部署

以下示例展示了如何使用 watsonx.ai Python 客户端库为您的AI服务创建批处理部署：

deployment_details = client.deployments.create(
    artifact_id=ai_service_id,
    meta_props={
        client.deployments.ConfigurationMetaNames.NAME: f"ai-service - batch",
        client.deployments.ConfigurationMetaNames.BATCH: {},
        client.deployments.ConfigurationMetaNames.HARDWARE_SPEC: {
            "id": client.hardware_specifications.get_id_by_name("XS")
        },
    },
)
deployment_id = client.deployments.get_uid(deployment_details)
print("The batch deployment id:", deployment_id)

创建在线部署

以下示例展示了如何使用REST API为您的AI服务创建在线部署：

# POST /ml/v4/deployments
response = requests.post(
    f'{HOST}/ml/v4/deployments?version={VERSION}',
    headers=headers,
    verify=False,
    json={
        "space_id": space_id,
        "name": "genai flow online",
        "custom": {
            "key1": "value1",
            "key2": "value2",
            "model": "meta-llama/llama-3-8b-instruct"
        },
        "asset": {
            "id": asset_id
        },
        "online": {}
    }
)