Integrating Azure OpenAI

from langchain.prompts import ChatPromptTemplate
from langchain.pydantic_v1 import BaseModel, Field
from langchain.utils.openai_functions import convert_pydantic_to_openai_function
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain.tools import StructuredTool
from langchain_openai import AzureChatOpenAI
from typing import List


####################
# LLM
####################

llm = AzureChatOpenAI(deployment_name="gpt-4", temperature=0.0)


####################
# Data Model
####################

class Idea(BaseModel):
    id: str = Field(description="Unique identifier for the idea")
    content: str = Field(description="Description of the new business idea")


class Classification(BaseModel):
    class_definition: str = Field(description="Classification label")
    reason: str = Field(description="Concise explanation (around 20 words)")


####################
# Function
####################

def generate_prompt(idea: Idea, class_definition: dict) -> str:
    template_sys = f"""
    ## Class Definition:
    {class_definition['definition']}
    """

    template_usr = f"""
    ## Input Data: New Business Idea Information
    ```json
    {{"id": "{idea.id}", "content": "{idea.content}"}}
    ```

    ## Task:
    - Thoroughly read the [Input Data: New Business Idea Information].
    - Based on the [Class Definition], select the class that best represents the characteristic of the new business idea.
    - Output the class and the reason for the classification (around 20 words).
    """

    return ChatPromptTemplate.from_messages([("system", template_sys), ("user", template_usr)]).as_str()


def eval_idea(idea: Idea, class_definition: dict) -> Classification:
    prompt = generate_prompt(idea, class_definition)
    output_func = convert_pydantic_to_openai_function(Classification)
    llm_func = llm.bind(functions=[output_func], function_call={"name": output_func["name"]})
    parser = JsonOutputFunctionsParser()

    chain = prompt | llm_func | parser

    result = chain.invoke()
    return result


def classify_ideas(ideas: List[Idea], categories: List[str]) -> List[Classification]:
    filtered_definitions = [d for d in list_class_definition if d["category"] in categories]
    results = []
    for idea in ideas:
        for class_definition in filtered_definitions:
            classification = eval_idea(idea, class_definition)
            classification.id = idea.id
            results.append(classification)
    return results


####################
# Structured Tool
####################

classify_idea_tool = StructuredTool.from_function(
    func=classify_ideas,
    name="classify_idea_tool",
    description="Classify new business ideas based on characteristic",
    args_schema=List[Idea],
    return_schema=List[Classification],
)