1# Usage:
 2#! ./llama-server -m some-model.gguf &
 3#! pip install pydantic
 4#! python json_schema_pydantic_example.py
 5
 6from pydantic import BaseModel, Field, TypeAdapter
 7from annotated_types import MinLen
 8from typing import Annotated, List, Optional
 9import json, requests
10
11if True:
12
13    def create_completion(*, response_model=None, endpoint="http://localhost:8080/v1/chat/completions", messages, **kwargs):
14        '''
15        Creates a chat completion using an OpenAI-compatible endpoint w/ JSON schema support
16        (llama.cpp server, llama-cpp-python, Anyscale / Together...)
17
18        The response_model param takes a type (+ supports Pydantic) and behaves just as w/ Instructor (see below)
19        '''
20        response_format = None
21        type_adapter = None
22
23        if response_model:
24            type_adapter = TypeAdapter(response_model)
25            schema = type_adapter.json_schema()
26            messages = [{
27                "role": "system",
28                "content": f"You respond in JSON format with the following schema: {json.dumps(schema, indent=2)}"
29            }] + messages
30            response_format={"type": "json_object", "schema": schema}
31
32        data = requests.post(endpoint, headers={"Content-Type": "application/json"},
33                             json=dict(messages=messages, response_format=response_format, **kwargs)).json()
34        if 'error' in data:
35            raise Exception(data['error']['message'])
36
37        content = data["choices"][0]["message"]["content"]
38        return type_adapter.validate_json(content) if type_adapter else content
39
40else:
41
42    # This alternative branch uses Instructor + OpenAI client lib.
43    # Instructor support streamed iterable responses, retry & more.
44    # (see https://python.useinstructor.com/)
45    #! pip install instructor openai
46    import instructor, openai
47    client = instructor.patch(
48        openai.OpenAI(api_key="123", base_url="http://localhost:8080"),
49        mode=instructor.Mode.JSON_SCHEMA)
50    create_completion = client.chat.completions.create
51
52
53if __name__ == '__main__':
54
55    class QAPair(BaseModel):
56        class Config:
57            extra = 'forbid'  # triggers additionalProperties: false in the JSON schema
58        question: str
59        concise_answer: str
60        justification: str
61        stars: Annotated[int, Field(ge=1, le=5)]
62
63    class PyramidalSummary(BaseModel):
64        class Config:
65            extra = 'forbid'  # triggers additionalProperties: false in the JSON schema
66        title: str
67        summary: str
68        question_answers: Annotated[List[QAPair], MinLen(2)]
69        sub_sections: Optional[Annotated[List['PyramidalSummary'], MinLen(2)]]
70
71    print("# Summary\n", create_completion(
72        model="...",
73        response_model=PyramidalSummary,
74        messages=[{
75            "role": "user",
76            "content": f"""
77                You are a highly efficient corporate document summarizer.
78                Create a pyramidal summary of an imaginary internal document about our company processes
79                (starting high-level, going down to each sub sections).
80                Keep questions short, and answers even shorter (trivia / quizz style).
81            """
82        }]))