多任务和流式处理¶

结构化提取的一个常见用例是定义一个单一的 schema 类，然后创建另一个 schema 来生成一个列表以进行多次提取。

from typing import List
from pydantic import BaseModel


class User(BaseModel):
    name: str
    age: int


class Users(BaseModel):
    users: List[User]


print(Users.model_json_schema())
"""
{
    '$defs': {
        'User': {
            'properties': {
                'name': {'title': 'Name', 'type': 'string'},
                'age': {'title': 'Age', 'type': 'integer'},
            },
            'required': ['name', 'age'],
            'title': 'User',
            'type': 'object',
        }
    },
    'properties': {
        'users': {'items': {'$ref': '#/$defs/User'}, 'title': 'Users', 'type': 'array'}
    },
    'required': ['users'],
    'title': 'Users',
    'type': 'object',
}
"""

定义一个任务并创建类的列表是一个足够常见的模式，我们通过利用 Iterable[T] 使其变得方便。这使得我们可以动态地创建一个新的类，该类

具有基于任务的动态 docstrings 和类名
通过收集令牌直到任务被返回来支持流式处理。

使用 Iterable 提取任务¶

通过使用 Iterable，您可以获得一个非常方便的类，其提示和名称会自动定义。

import instructor
from openai import OpenAI
from typing import Iterable
from pydantic import BaseModel

client = instructor.from_openai(OpenAI(), mode=instructor.function_calls.Mode.JSON)


class User(BaseModel):
    name: str
    age: int


users = client.chat.completions.create(
    model="gpt-3.5-turbo-1106",
    temperature=0.1,
    response_model=Iterable[User],
    stream=False,
    messages=[
        {
            "role": "user",
            "content": "Consider this data: Jason is 10 and John is 30.\
                         Correctly segment it into entitites\
                        Make sure the JSON is correct",
        },
    ],
)
for user in users:
    print(user)
    #> name='Jason' age=10
    #> name='John' age=30

流式任务¶

我们还可以通过定义 Iterable[T] 类型来在令牌流进来时生成任务。

让我们看一个使用相同类的实际示例。

import instructor
import openai
from typing import Iterable
from pydantic import BaseModel

client = instructor.from_openai(openai.OpenAI(), mode=instructor.Mode.TOOLS)


class User(BaseModel):
    name: str
    age: int


users = client.chat.completions.create(
    model="gpt-4",
    temperature=0.1,
    stream=True,
    response_model=Iterable[User],
    messages=[
        {
            "role": "system",
            "content": "You are a perfect entity extraction system",
        },
        {
            "role": "user",
            "content": (f"Extract `Jason is 10 and John is 10`"),
        },
    ],
    max_tokens=1000,
)

for user in users:
    print(user)
    #> name='Jason' age=10
    #> name='John' age=10

异步流式处理¶

我还想在这个例子中指出，instructor 也支持异步流式处理。当您想要流式传输响应模型并在结果到达时处理它们时，这非常有用，但您需要使用 async for 语法来迭代结果。

import instructor
import openai
from typing import Iterable
from pydantic import BaseModel

client = instructor.from_openai(openai.AsyncOpenAI(), mode=instructor.Mode.TOOLS)


class UserExtract(BaseModel):
    name: str
    age: int


async def print_iterable_results():
    model = await client.chat.completions.create(
        model="gpt-4",
        response_model=Iterable[UserExtract],
        max_retries=2,
        stream=True,
        messages=[
            {"role": "user", "content": "Make two up people"},
        ],
    )
    async for m in model:
        print(m)
        #> name='John Doe' age=35
        #> name='Jane Smith' age=28


import asyncio

asyncio.run(print_iterable_results())