app/workflows/utils/feedback.py

from pydantic import BaseModel, UUID4
from typing import List, Optional
from langsmith import Client
from app.lib.supabase_client import SupabaseClient

class Feedback(BaseModel):
    metric_type: Optional[str]
    metric_score: Optional[float]
    feedback_on: Optional[str]


def post_feedback(run_id: UUID4, feedback: Feedback):
    print("Metric Type: ", feedback.metric_type)
    print("Feedback On: ", feedback.feedback_on)

    client = Client()
    client.create_feedback(
        str(run_id),
        key=feedback.metric_type,
        score=feedback.metric_score,
        source_info={"content": feedback.feedback_on},
        type="api",
    )

## fields of inputs dict: run_id, sub_workflow
class NewFeedback(BaseModel):
    metric_type: str
    metric_score: float
    original_content: Optional[dict]
    modified_content: Optional[dict]

    def post(self, inputs={}):
        run_id = inputs["run_id"]
        sub_workflow = inputs["sub_workflow"]

        self._post_to_supbase(run_id, sub_workflow)
        self._post_to_langsmith(run_id)

    def _post_to_supbase(self, run_id: UUID4, sub_worfklow: str):
        client = SupabaseClient()
        client.post_feedback(
            run_id= run_id,
            sub_workflow= sub_worfklow,
            metric_score=self.metric_score,
            metric_type=self.metric_type,
            original_content=self.original_content,
            modified_content=self.modified_content
        )
        return

    def _post_to_langsmith(self, run_id: UUID4):
        client = Client()
        client.create_feedback(
            str(run_id),
            key=self.metric_type,
            score=self.metric_score,
            source_info={"original_content": self.original_content, "modified_content": self.modified_content},
            type="api",
        )