Spaces:
Sleeping
Sleeping
File size: 7,883 Bytes
e6603b1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | from langchain import callbacks
from langchain import hub
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.tools.tavily_search import TavilyAnswer
from langchain_core.messages import SystemMessage
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, PromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field, UUID4
from typing import List, Optional
import os
import pprint
class TilCrew:
def kickoff(self, inputs={}):
print("Human Message:")
pprint.pp(inputs)
self.content = inputs["content"]
# self._gather_facts()
self._gather_feedback()
return self._final_call_on_feedback()
def _final_call_on_feedback(self):
final_results = []
for feedback in self.feedback_results:
print("Final analysis of:")
pprint.pp(feedback)
result = {
"til": feedback.get('til', ""),
"feedback": "not_ok",
}
if feedback["factuality_categorization"] != 'High':
result["feedback_criteria"] = "factuality_feedback"
result["reason"] = feedback["factuality_reason"]
final_results = final_results + [result]
continue
if feedback["insightful_categorization"] != 'High':
result["feedback_criteria"] = "insightful_feedback"
result["reason"] = feedback["insightful_reason"]
final_results = final_results + [result]
continue
if feedback["simplicity_categorization"] == 'Low':
result["feedback_criteria"] = "simplicity_feedback"
result["reason"] = feedback["simplicity_reason"]
result["suggestion"] = feedback["final_suggestion"]
final_results = final_results + [result]
continue
if feedback["grammatical_categorization"] == 'Low':
result["feedback_criteria"] = "grammatical_feedback"
result["reason"] = feedback["grammatical_reason"]
result["suggestion"] = feedback["final_suggestion"]
final_results = final_results + [result]
continue
result["feedback"] = "ok"
final_results = final_results + [result]
response = {"feedback": final_results, "run_id": self.run_id }
print("Final Results:")
pprint.pp(response)
return response
def _gather_feedback(self):
feedback_chain = self._build_feedback_chain()
pprint.pp("Analysing the TIL.....")
with callbacks.collect_runs() as cb:
self.feedback_results = feedback_chain.invoke({"til_content": self.content})['tils']
self.run_id = cb.traced_runs[0].id
print("Run ID: ", self.run_id)
print("Feedback: ")
pprint.pp(self.feedback_results)
# Deprecated: Not using this as we are getting similar results by using or without using this
def _gather_facts(self):
facts_prompt = PromptTemplate.from_template("What are the facts on the topics mentioned the following user's TILs: {content}")
tools = [TavilyAnswer()]
llm = ChatOpenAI(model=os.environ['OPENAI_MODEL'], temperature=0.2)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
self.facts = agent_executor.invoke({"input": facts_prompt.format(content=self.content)})['output']
print("Gathered Facts: ")
pprint.pp(self.facts)
def _build_feedback_chain(self):
feedback_parser = JsonOutputParser(pydantic_object=TilFeedbackResults)
feedback_prompt = ChatPromptTemplate.from_messages([
SystemMessage(
"You are a 'Personal TIL Reviewer' who works in a Product Engineering Services company. "
"You are an expert in writing TILs which are Insightful, Factually correct, Easy to read and grammatically correct."
"Your goal is to review user's TILs and categorize their correctness as High, Medium, or Low based on the following metrics:"
"1. Is the TIL insightful?"
"2. Is the TIL factually correct and accurate?"
"3. Is the TIL written in simple english?"
"4. Is the TIL grammatically correct?\n\n"
"The criteria to use for assessing if they are insightful or not are:\n"
"* They TIL shouldn't just be a outright statement, it should contain even the reason on why the statement is true."
"* It should showcase the understanding of the user on the subject.\n\n"
"The criteria to use for assessing if they are factual or not are:\n"
"* They are related to facts."
"* You are able to find a source which agrees to the fact from reputable websites.\n\n"
"Give reason for your assessment in one or two sentences for each metric and And also rewrite the TIL if you were given the option to write it. "
"Evaluate each TIL in the context of all the user's TILs."
f"Formatting Instructions: {feedback_parser.get_format_instructions()}"
),
HumanMessagePromptTemplate.from_template("{til_content}")
])
print("Prompt: ")
pprint.pp(feedback_prompt, width=80)
llm = ChatOpenAI(model=os.environ['OPENAI_MODEL'], temperature=0.2)
analysis_chain = (feedback_prompt | llm | feedback_parser).with_config({
"tags": ["til"], "run_name": "Analysing TIL",
"metadata" : {
"versoin": "v1.0.0",
"growth_activity": "til",
"env": os.environ["ENV"],
"model": os.environ["OPENAI_MODEL"],
}
})
return analysis_chain
class TilFeedbackResult(BaseModel):
til: str = Field(description="TIL as exactly captured by the user without any modifications.")
insightful_categorization: str = Field(
description="TIL categorization as High/Medium/Low based on correctness on the insightful metric.")
insightful_reason: str = Field(description="Reason for your assessment in one or two sentences on insightful metric for the user.")
factuality_categorization: str = Field(
description="TIL categorization as High/Medium/Low based on correctness on the factuality metric.")
factuality_reason: str = Field(description="Reason for your assessment in one or two sentences on factuality metric for the user.")
simplicity_categorization: str = Field(
description="TIL categorization as High/Medium/Low based on correctness on the simplicity metric.")
simplicity_reason: str = Field(description="Reason for your assessment in one or two sentences on simplicity metric for the user.")
grammatical_categorization: str = Field(
description="TIL categorization as High/Medium/Low based on correctness on the grammatical metric.")
grammatical_reason: str = Field(description="Reason for your assessment in one or two sentences on grammatical metric for the user.")
final_suggestion: str = Field(
description="Rewrite the TIL if you were given the option to write it which should score High on all the metrics.")
class TilFeedbackResults(BaseModel):
tils: List[TilFeedbackResult]
class TilFinalFeedback(BaseModel):
til: str
feedback: str
feedback_criteria: Optional[str] = None
reason: Optional[str] = None
suggestion: Optional[str] = None
class TilFeedbackResponse(BaseModel):
run_id: UUID4
feedback: List[TilFinalFeedback]
|