解密prompt系列46. LLM结构化输出代码示例和原理分析

from enum import Enum
from typing import List
from pydantic import BaseModel, Field

class SWIndustry(Enum):
    AGRICULTURE = "农林牧渔"
    MINING = "采掘"
    CHEMICALS = "化工"
    STEEL = "钢铁"
    NONFERROUS_METALS = "有色金属"
    BUILDING_MATERIALS = "建筑材料"
    ELECTRICAL_EQUIPMENT = "电气设备"
    APPLIANCES = "家用电器"
    FOOD_BEVERAGE = "食品饮料"
    TEXTILE_APPAREL = "纺织服装"
    LIGHT_MANUFACTURING = "轻工制造"
    PHARMACEUTICALS = "医药生物"
    PUBLIC_UTILITIES = "公用事业"
    TRANSPORTATION = "交通运输"
    REAL_ESTATE = "房地产"
    COMMERCE_TRADE = "商业贸易"
    COMPUTER = "计算机"
    MEDIA = "传媒"
    COMMUNICATION = "通信"
    BANKING = "银行"
    NON_BANK_FINANCIAL = "非银金融"
    AUTOMOBILE = "汽车"
    MACHINERY = "机械设备"
    DEFENSE_MILITARY = "国防军工"
    BUILDING_CONSTRUCTION = "建筑装饰"
    ELECTRONICS = "电子"
    COMPREHENSIVE = "综合"
    LEISURE_SERVICES = "休闲服务"
    COMPUTER_APPLICATIONS = "计算机应用"
    CHEMICAL_FIBERS = "化纤"
    METAL_PRODUCTS = "金属制品"

class ViewAspect(Enum):
    POSITIVE = '正面'
    NETURAL = '中性'
    NEGATIVE = '负面'

    
class View(BaseModel):
    extract_view: str = Field(description="抽取文档中中对某个金融行业、或行业相关的主题或概念表达观点的句子")
    extract_view_entities: List[str] = Field(description="抽取观点金融主体,该主体必须出现在观点句子中，可以是金融行业，或行业相关概念或主题")
    related_industry: list[SWIndustry] = Field(description=f"观点金融主体最相关的1个或多个申万一级行业")
    view_aspect: ViewAspect = Field(description=f'对观点情绪进行精准分类，模糊情绪均为中性')

class ViewExtraction(BaseModel):
    views: list[View] = Field(
        ...,
        description="每个观点都应该是一个单独的对象，包含原文中表达观点的句子，观点主体，观点情绪分类和关联的申万一级行业",
    )

from openai import AzureOpenAI

client = AzureOpenAI(
    api_key = '...',
    api_version="2024-08-01-preview",
    azure_endpoint= "..."
)

completion = client.beta.chat.completions.parse(
    model='gpt-4o',
 messages=[
            {
                "role": "system",
                "content": "你是一个完美的金融观点解析系统，可以从文档中抽取观点，和观点对应主体并对观点进行分类。请从以下文档中抽取一系列观点",
            },
            {
                "role": "user",
                "content": content,
            },
        ],
    response_format= ViewExtraction
)

from typing import Literal,Union,Optional
class BingSearch(BaseModel):
    query: str = Field(description="网页搜索query")
        
class FundInfo(BaseModel):
    """
    可以通过基金代码或基金名称，查询基金基础信息
    """
    fund_code_or_name: Optional[str] = Field(description="提问提及的基金代码或名称，没有则为空")
    lookup_field: Literal["fund_manager", "unit_value","contract_date","manage_fee","net_value"]

class Task(BaseModel):
    name: str = Field(description="任务名称")
    tool: Union[BingSearch, FundInfo] = Field(description="完成任务所需调用的工具")
        
class TaskSequence(BaseModel):
    reason: str = Field(description="先逐步思考要解决用户的问题需要哪些步骤")
    task_actions: List[Task] = Field(description="任务列表，按执行顺序依次排列")
   
completion = client.beta.chat.completions.parse(
    model='gpt-4o',
 messages=[
            {
                "role": "system",
                "content": "你是一个金融工具助手，可以完美根据用户提问选择需要调用的工具列表",
            },
            {
                "role": "user",
                "content": '天弘中证500当前的管理费是多少，是否随着基金规模的增加而增加',
            },
        ],
    response_format= TaskSequence
)

from openai import AzureOpenAI
import instructor
client = AzureOpenAI(
    api_key = '...',
    api_version="2024-08-01-preview",
    azure_endpoint= "..."
)
client = instructor.from_openai(client)

resp = client.chat.completions.create(
    model='thgpt4o',
    response_model= ViewExtraction,
    messages=[
        {
            "role": "system",
            "content": "你是一个完美的金融观点解析系统，可以从文档中抽取观点，和观点对应主体并对观点进行分类。请从以下文档中抽取一系列观点",
        },
        {
            "role": "user",
            "content": content,
        },
    ],
) 

def sequence_generator(model, sampler, fsms, token_ids, sequence_weights, attention_masks, fsm_states, rng):
    while True:
        # 1. 获取模型输出的logits
        logits, kv_cache = model(token_ids, attention_masks, kv_cache)
        
        # 2. 获取FSM允许的下一个token
        allowed_tokens = get_allowed_tokens(fsms, fsm_states)
        
        # 3. 基于allowed_tokens对logits进行mask,不允许的token均为-inf
        biased_logits = bias_logits(logits, allowed_tokens)
        
        # 4. 采样下一个token
        next_token_ids, ancestors, sequence_weights = sampler(biased_logits, sequence_weights, rng)
        
        # 5. 更新FSM状态
        fsm_states = get_next_fsm_states(fsms, fsm_states, next_token_ids)
        
        # 6. 检查是否生成完成
        is_finished = is_generation_finished(fsms, fsm_states)

@overload
def from_openai(
    client: openai.OpenAI,
    mode: instructor.Mode = instructor.Mode.TOOLS,
    **kwargs: Any,
) -> Instructor:
    pass
    
@overload
def patch(
    client: OpenAI,
    mode: Mode = Mode.TOOLS,
) -> OpenAI: ...

def patch(  # type: ignore
    client: OpenAI | AsyncOpenAI | None = None,
    create: Callable[T_ParamSpec, T_Retval] | None = None,
    mode: Mode = Mode.TOOLS,
) -> OpenAI | AsyncOpenAI:
    """
    Patch the `client.chat.completions.create` method

    Enables the following features:

    - `response_model` parameter to parse the response from OpenAI's API
    - `max_retries` parameter to retry the function if the response is not valid
    - `validation_context` parameter to validate the response using the pydantic model
    - `strict` parameter to use strict json parsing
    - `hooks` parameter to hook into the completion process
    """

    logger.debug(f"Patching `client.chat.completions.create` with {mode=}")

    if create is not None:
        func = create
    elif client is not None:
        func = client.chat.completions.create
    else:
        raise ValueError("Either client or create must be provided")

    @wraps(func)  # type: ignore
    def new_create_sync(
        response_model: type[T_Model] | None = None,
        validation_context: dict[str, Any] | None = None,
        context: dict[str, Any] | None = None,
        max_retries: int | Retrying = 1,
        strict: bool = True,
        hooks: Hooks | None = None,
        *args: T_ParamSpec.args,
        **kwargs: T_ParamSpec.kwargs,
    ) -> T_Model:
        context = handle_context(context, validation_context)

        response_model, new_kwargs = handle_response_model(
            response_model=response_model, mode=mode, **kwargs
        )

        new_kwargs = handle_templating(new_kwargs, context)

        response = retry_sync(
            func=func,  # type: ignore
            response_model=response_model,
            context=context,
            max_retries=max_retries,
            args=args,
            hooks=hooks,
            strict=strict,
            kwargs=new_kwargs,
            mode=mode,
        )
        return response  # type: ignore
    new_create = new_create_async if func_is_async else new_create_sync
    if client is not None:
        client.chat.completions.create = new_create  # type: ignore
        return client
    else:
        return new_create  # type: ignore

def openai_schema(cls) -> dict[str, Any]:
    """
    Return the schema in the format of OpenAI's schema as jsonschema

    Note:
        Its important to add a docstring to describe how to best use this class, it will be included in the description attribute and be part of the prompt.

    Returns:
        model_json_schema (dict): A dictionary in the format of OpenAI's schema as jsonschema
    """
    schema = cls.model_json_schema()
    docstring = parse(cls.__doc__ or "")
    parameters = {
        k: v for k, v in schema.items() if k not in ("title", "description")
    }
    for param in docstring.params:
        if (name := param.arg_name) in parameters["properties"] and (
            description := param.description
        ):
            if "description" not in parameters["properties"][name]:
                parameters["properties"][name]["description"] = description

    parameters["required"] = sorted(
        k for k, v in parameters["properties"].items() if "default" not in v
    )

    if "description" not in schema:
        if docstring.short_description:
            schema["description"] = docstring.short_description
        else:
            schema["description"] = (
                f"Correctly extracted `{cls.__name__}` with all "
                f"the required parameters with correct types"
            )

    return {
        "name": schema["title"],
        "description": schema["description"],
        "parameters": parameters,
    }

{'name': 'TaskSequence',
 'description': 'Correctly extracted `TaskSequence` with all the required parameters with correct types',
 'parameters': {'$defs': {'BingSearch': {'properties': {'query': {'description': '网页搜索query',
      'title': 'Query',
      'type': 'string'}},
    'required': ['query'],
    'title': 'BingSearch',
    'type': 'object'},
   'FundInfo': {'description': '可以通过基金代码或基金名称，查询基金基础信息',
    'properties': {'fund_code_or_name': {'anyOf': [{'type': 'string'},
       {'type': 'null'}],
      'description': '提问提及的基金代码或名称，没有则为空',
      'title': 'Fund Code Or Name'},
     'lookup_field': {'enum': ['fund_manager',
       'unit_value',
       'contract_date',
       'manage_fee',
       'net_value'],
      'title': 'Lookup Field',
      'type': 'string'}},
    'required': ['fund_code_or_name', 'lookup_field'],
    'title': 'FundInfo',
    'type': 'object'},
   'Task': {'properties': {'name': {'description': '任务名称',
      'title': 'Name',
      'type': 'string'},
     'tool': {'anyOf': [{'$ref': '#/$defs/BingSearch'},
       {'$ref': '#/$defs/FundInfo'}],
      'description': '完成任务所需调用的工具',
      'title': 'Tool'}},
    'required': ['name', 'tool'],
    'title': 'Task',
    'type': 'object'}},
  'properties': {'reason': {'description': '先逐步思考要解决用户的问题需要哪些步骤',
    'title': 'Reason',
    'type': 'string'},
   'task_actions': {'description': '任务列表，按执行顺序依次排列',
    'items': {'$ref': '#/$defs/Task'},
    'title': 'Task Actions',
    'type': 'array'}},
  'required': ['reason', 'task_actions'],
  'type': 'object'}}