从代码构建模型

# If running in a Jupyter or Databricks notebook cell, uncomment the following line:
# %%writefile "./basic.py"

import pandas as pd
from typing import List, Dict
from mlflow.pyfunc import PythonModel
from mlflow.models import set_model


class BasicModel(PythonModel):
    def exponential(self, numbers):
        return {f"{x}": 2**x for x in numbers}

    def predict(self, context, model_input) -> Dict[str, float]:
        if isinstance(model_input, pd.DataFrame):
            model_input = model_input.to_dict()[0].values()
        return self.exponential(model_input)


# Specify which definition in this script represents the model instance
set_model(BasicModel())

import mlflow

mlflow.set_experiment("Basic Model From Code")

model_path = "basic.py"

with mlflow.start_run():
    model_info = mlflow.pyfunc.log_model(
        python_model=model_path,  # Define the model as the path to the script that was just saved
        artifact_path="arithmetic_model",
        input_example=[42.0, 24.0],
    )

my_model = mlflow.pyfunc.load_model(model_info.model_uri)
my_model.predict([2.2, 3.1, 4.7])

# or, with a Pandas DataFrame input
my_model.predict(pd.DataFrame([5.0, 6.0, 7.0]))

# If running in a Jupyter or Databricks notebook cell, uncomment the following line:
# %%writefile "./calculator.py"


def add(x, y):
    return x + y

# If running in a Jupyter or Databricks notebook cell, uncomment the following line:
# %%writefile "./math_model.py"

from mlflow.pyfunc import PythonModel
from mlflow.models import set_model

from calculator import add


class AddModel(PythonModel):
    def predict(self, context, model_input, params=None):
        return add(model_input["x"], model_input["y"])


set_model(AddModel())

import mlflow

mlflow.set_experiment("Arithmetic Model From Code")

model_path = "math_model.py"

with mlflow.start_run():
    model_info = mlflow.pyfunc.log_model(
        python_model=model_path,  # The model is defined as the path to the script containing the model definition
        artifact_path="arithmetic_model",
        code_paths=[
            "calculator.py"
        ],  # dependency definition included for the model to successfully import the implementation
    )

my_model_from_code = mlflow.pyfunc.load_model(model_info.model_uri)
my_model_from_code.predict({"x": 42, "y": 9001})

# If running in a Jupyter or Databricks notebook cell, uncomment the following line:
# %%writefile "./mfc.py"

import os
from operator import itemgetter

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

import mlflow


def get_region(input_data):
    default = "Virginia, USA"
    if isinstance(input_data[0], dict):
        return input_data[0].get("content").get("region", default)
    return default


def get_area(input_data):
    default = "5000 square feet"
    if isinstance(input_data[0], dict):
        return input_data[0].get("content").get("area", default)
    return default


prompt = PromptTemplate(
    template="You are a highly accomplished landscape designer that provides suggestions for landscape design decisions in a particular"
    " geographic region. Your goal is to suggest low-maintenance hardscape and landscape options that involve the use of materials and"
    " plants that are native to the region mentioned. As part of the recommendations, a general estimate for the job of creating the"
    " project should be provided based on the square footage estimate. The region is: {region} and the square footage estimate is:"
    " {area}. Recommendations should be for a moderately sophisticated suburban housing community within the region.",
    input_variables=["region", "area"],
)

model = ChatOpenAI(model="gpt-4o", temperature=0.95, max_tokens=4096)

chain = (
    {
        "region": itemgetter("messages") | RunnableLambda(get_region),
        "area": itemgetter("messages") | RunnableLambda(get_area),
    }
    | prompt
    | model
    | StrOutputParser()
)

mlflow.models.set_model(chain)

import mlflow

mlflow.set_experiment("Landscaping")

chain_path = "./mfc.py"

input_example = {
    "messages": [
        {
            "role": "user",
            "content": {
                "region": "Austin, TX, USA",
                "area": "1750 square feet",
            },
        }
    ]
}

with mlflow.start_run():
    info = mlflow.langchain.log_model(
        lc_model=chain_path,  # Defining the model as the script containing the chain definition and the set_model call
        artifact_path="chain",
        input_example=input_example,
    )

# Load the model and run inference
landscape_chain = mlflow.langchain.load_model(model_uri=info.model_uri)

question = {
    "messages": [
        {
            "role": "user",
            "content": {
                "region": "Raleigh, North Carolina USA",
                "area": "3850 square feet",
            },
        },
    ]
}

response = landscape_chain.invoke(question)