Pydantic과 OpenAI 함수 호출을 활용한 더 나은 데이터 추출

functions = [{
 "name": "ExtractUser",
 "description": "Extract User information",
 "parameters": {
  "type": "object",
  "properties": {
   "name": {
    "type": "string"
   },
   "age": {
    "type": "integer"
   }
  },
  "required": ["age","name"]
 }
}]

functions = [{
 "name": "MultiSearch",
 "description": "Correct segmentation of `Search` tasks", # prompting 
 "parameters": {
  "type": "object",
  "properties": {
   "tasks": {
    "type": "array",
    "items": {"$ref": "#/definitions/Search"}
   }
  },
  "definitions": {
   "Source": {
    "description": "An enumeration.",
    "enum": ["VIDEO","TRANSCRIPT","DOCUMENT"]
   },
   "Search": {
    "type": "object",
    "properties": {
     "query": {
      "description": "Detailed, comprehensive, and specific query to be used for semantic search", # prompting 
      "type": "string"
     },
     "source": { "$ref": "#/definitions/Source"}
    },
    "required": ["title","query","source"]
   }
  },
  "required": ["tasks"]
 }
}]

# Writing code like this gets crazier the more you need to trust your outputs
data = json.loads(completion...["function_call"]
assert hasattr(data, "tasks")
for task in data.tasks:
	assert isinstance(task, dict), "task is not a dict"
	assert list(task.keys) == ["title", "query", "source"], "task missing keys"
	assert tasks.source in {"VIDEO","TRANSCRIPT","DOCUMENT"}, "source not valid"

from openai_function_call import OpenAISchema
from pydantic import Field
import enum
import openai

class Source(enum.Enum):
    VIDEO = "VIDEO"
    TRANSCRIPT = "TRANSCRIPT"
    DOCUMENT = "DOCUMENT"

class Search(OpenAISchema):
    query: str = Field(
        ..., description="Detailed, comprehensive, and specific query to be used for semantic search",
    )
    source: Source

    def search(self):
	# any logic can go here since its just python
    	return f"Fake results: `{self.query}` from {self.source}"

class MultiSearch(OpenAISchema):
    "Correct segmentation of `Search` tasks"
    tasks: list[Search]

completion = openai.ChatCompletion.create(
    model="gpt-3.5-turbo-0613",
    functions=[MultiSearch.openai_schema],
    function_call={"name": MultiSearch.openai_schema["name"]},
    messages=[
        {
	"role": "user", 
	"content": "Can you show me the cat video you found last week and the onboarding documents as well?"
	},
    ],
)

response = MultiSearch.from_response(completion)
response 
>>> MultiSearch(tasks=[
	Search(title="cat videos", query="cat videos from last week", source=Source.VIDEO),
	Search(title="onboarding documents", query="onboarding documents", source=Source.DOCUMENT),
])

from openai_function_call import OpenAISchema
from pydantic import Field 

class ExtractUser(OpenAISchema):
    "Correctly extracted user information" #(1)
    name: str = Field(..., description="User's full name") #(2)
    age: int

>>> ExtractUser.openai_schema
{
"name": "ExtractUser",
"description": "Correctly extracted user information", #(1)
"parameters": {
    "type": "object",
    "properties": {
    "name": {
        "description": "User's full name", #(2)
        "type": "string"
    },
    "age": {
        "type": "integer"
    }
    },
    "required": ["age","name"]
}}

response = MultiSearch.from_response(completion)
assert isinstance(response, MultiSearch)
assert isinstance(response.tasks[0], Search)
assert isinstance(response.tasks[0].source, Source)

# adding methods to your schema is totally fine
class Search(OpenAISchema):
    ...
    def search(self) -> list:
        if self.source == Source.VIDEO:
            return video_index.search(self.query) 
        else if self.source == Source.DOCUMENTS:
            return doc_index.search(self.query)

ms = MultiSearch.from_response(completion)
>>> [s.search() for s in ms]

def search(query: Search) -> list:
    if query.source == Source.VIDEO:
            return video_index.search(self.query) 
        else if query.source == Source.DOCUMENTS:
            return doc_index.search(self.query)

ms = MultiSearch.from_response(completion)
>>> [search(s) for s in ms]

from pydantic import Field
from openai_function_call import OpenAISchema


class Fact(OpenAISchema):
    """
    Each fact has a body and a list of sources.
    If there are multiple facts, make sure to break them apart such that each one only uses a set of sources that are relevant to it.
    """

    fact: str = Field(..., description="Body of the sentence as part of a response")
    substring_quote: list[str] = Field(
        ...,
        description="Each source should be a direct quote from the context, as a substring of the original content",
    )

    def _get_span(self, quote, context, errs=100):
        import regex

        minor = quote
        major = context

        errs_ = 0
        s = regex.search(f"({minor}){{e<={errs_}}}", major)
        while s is None and errs_ <= errs:
            errs_ += 1
            s = regex.search(f"({minor}){{e<={errs_}}}", major)

        if s is not None:
            yield from s.spans()

    def get_spans(self, context):
        for quote in self.substring_quote:
            yield from self._get_span(quote, context)


class QuestionAnswer(OpenAISchema):
    """
    Class representing a question and its answer as a list of facts, where each fact should have a source.
    Each sentence contains a body and a list of sources.
    """

    question: str = Field(..., description="Question that was asked")
    answer: list[Fact] = Field(
        ...,
        description="Body of the answer, each fact should be its separate object with a body and a list of sources",
    )

def ask_ai(question: str, context: str) -> QuestionAnswer:
    # Making a request to the hypothetical 'openai' module
    completion = openai.ChatCompletion.create(
        model="gpt-3.5-turbo-0613",
        temperature=0.2,
        max_tokens=1000,
        functions=[QuestionAnswer.openai_schema],
        function_call={"name": QuestionAnswer.openai_schema["name"]},
        messages=[
            {
                "role": "system",
                "content": f"You are a world class algorithm to answer questions with correct and exact citations. ",
            },
            {"role": "user", "content": f"Answer question using the following context"},
            {"role": "user", "content": f"{context}"},
            {"role": "user", "content": f"Question: {question}"},
            {
                "role": "user",
                "content": f"Tips: Make sure to cite your sources, and use the exact words from the context.",
            },
        ],
    )

    # Creating an Answer object from the completion response
    return QuestionAnswer.from_response(completion)

def highlight(text, span):
    return ...

question = "What did the author do during college?"
context = """
My name is Jason Liu, and I grew up in Toronto Canada but I was born in China.
I went to an arts high school but in university I studied Computational Mathematics and physics.
As part of coop I worked at many companies including Stitchfix, Facebook.
I also started the Data Science club at the University of Waterloo and I was the president of the club for 2 years.
"""

answer = ask_ai(question, context)

print("Question:", question)
print()
for fact in answer.answer:
    print("Statement:", fact.fact)
    for span in fact.get_spans(context):
        print("Citation:", highlight(context, span))
    print()

Question: What did the author do during college?

Statement: The author studied Computational Mathematics and physics in university.
Citation: ...rts high school but <in university I studied Computational Mathematics and physics> .As part of coop I ...

Statement: The author started the Data Science club at the University of Waterloo.
Citation: ...titchfix, Facebook.<I also started the Data Science club at the University of Waterloo>  and I was the presi...

Statement: The author was the president of the Data Science club for 2 years.
Citation: ...ity of Waterloo and <I was the president of the club for 2 years> ....

class QueryType(str, enum.Enum):
    """Enumeration representing the types of queries that can be asked to a question answer system."""

    SINGLE_QUESTION = "SINGLE"
    MERGE_MULTIPLE_RESPONSES = "MERGE_MULTIPLE_RESPONSES"


class Query(OpenAISchema):
    """Class representing a single question in a query plan."""

    id: int = Field(..., description="Unique id of the query")
    question: str = Field(
        ...,
        description="Question asked using a question answering system",
    )
    dependancies: List[int] = Field(
        default_factory=list,
        description="List of sub questions that need to be answered before asking this question",
    )
    node_type: QueryType = Field(
        default=QueryType.SINGLE_QUESTION,
        description="Type of question, either a single question or a multi-question merge",
    )


class QueryPlan(OpenAISchema):
    """Container class representing a tree of questions to ask a question answering system."""

    query_graph: List[Query] = Field(
        ..., description="The query graph representing the plan"
    )

    def _dependencies(self, ids: List[int]) -> List[Query]:
        """Returns the dependencies of a query given their ids."""
        return [q for q in self.query_graph if q.id in ids]

def query_planner(question: str) -> QueryPlan:
    PLANNING_MODEL = "gpt-4-0613"

    messages = [
        {
            "role": "system",
            "content": "You are a world class query planning algorithm capable ofbreaking apart questions into its dependency queries such that the answers can be used to inform the parent question. Do not answer the questions, simply provide a correct compute graph with good specific questions to ask and relevant dependencies. Before you call the function, think step-by-step to get a better understanding of the problem.",
        },
        {
            "role": "user",
            "content": f"Consider: {question}\nGenerate the correct query plan.",
        },
    ]

    completion = openai.ChatCompletion.create(
        model=PLANNING_MODEL,
        temperature=.2,
        functions=[QueryPlan.openai_schema],
        function_call={"name": QueryPlan.openai_schema["name"]},
        messages=messages,
        max_tokens=1000,
    )
    return QueryPlan.from_response(completion)

plan = query_planner(
    "What is the difference in populations of Canada and the Jason's home country?"
)
plan.dict()

{'query_graph': [{'id': 1,
   'question': 'What is the population of Canada?',
   'dependancies': [],
   'node_type': <QueryType.SINGLE_QUESTION: 'SINGLE'>},
  {'id': 2,
   'question': "What is Jason's home country?",
   'dependancies': [],
   'node_type': <QueryType.SINGLE_QUESTION: 'SINGLE'>},
  {'id': 3,
   'question': "What is the population of Jason's home country?",
   'dependancies': [2],
   'node_type': <QueryType.SINGLE_QUESTION: 'SINGLE'>},
  {'id': 4,
   'question': "What is the difference in populations of Canada and Jason's home country?",
   'dependancies': [1, 3],
   'node_type': <QueryType.MERGE_MULTIPLE_RESPONSES: 'MERGE_MULTIPLE_RESPONSES'>}]}

from typing import Any

class RowData(OpenAISchema):
    row: list[Any] = Field(..., description="The values for each row")
    citation: str = Field(
        ..., description="The citation for this row from the original source data"
    )


class Dataframe(OpenAISchema):
    """
    Class representing a dataframe. This class is used to convert
    data into a frame that can be used by pandas.
    """

    name: str = Field(..., description="The name of the dataframe")
    data: List[RowData] = Field(
        ...,
        description="Correct rows of data aligned to column names, Nones are allowed",
    )
    columns: list[str] = Field(
        ...,
        description="Column names relevant from source data, should be in snake_case",
    )

    def to_pandas(self):
        import pandas as pd

        columns = self.columns + ["citation"]
        data = [row.row + [row.citation] for row in self.data]

        return pd.DataFrame(data=data, columns=columns)


class Database(OpenAISchema):
    """
    A set of correct named and defined tables as dataframes
    """

    tables: list[Dataframe] = Field(
        ...,
        description="List of tables in the database",
    )

def dataframe(data: str) -> Database:
    completion = openai.ChatCompletion.create(
        model="gpt-4-0613", # Notice I have to use gpt-4 here, this task is pretty hard
        temperature=0.1,
        functions=[Database.openai_schema],
        function_call={"name": Database.openai_schema["name"]},
        messages=[
            {
                "role": "system",
                "content": """Map this data into a dataframe a
                nd correctly define the correct columns and rows""",
            },
            {
                "role": "user",
                "content": f"{data}",
            },
        ],
        max_tokens=1000,
    )
    return Database.from_response(completion)

dfs = dataframe("""My name is John and I am 25 years old. I live in
New York and I like to play basketball. His name is
Mike and he is 30 years old. He lives in San Francisco
and he likes to play baseball. Sarah is 20 years old
and she lives in Los Angeles. She likes to play tennis.
Her name is Mary and she is 35 years old.
She lives in Chicago.

On one team 'Tigers' the captain is John and there are 12 players.
On the other team 'Lions' the captain is Mike and there are 10 players.
""")
for df in dfs:
   print(df.to_pandas())


    People
    Name  Age           City Favorite Sport
    0   John   25       New York     Basketball
    1   Mike   30  San Francisco       Baseball
    2  Sarah   20    Los Angeles         Tennis
    3   Mary   35        Chicago           None

    Teams
    Team Name Captain  Number of Players
    0    Tigers    John                 12
    1     Lions    Mike                 10