Spaces:
Runtime error
Runtime error
File size: 3,990 Bytes
0413dd6 c22b03b 0413dd6 c22b03b 3d8743c c22b03b e25475d c22b03b e25475d c22b03b e25475d 0413dd6 3d8743c b49d827 e25475d 3d8743c c22b03b e25475d b49d827 0413dd6 c22b03b d9efaab c22b03b e25475d bb6a634 0413dd6 b49d827 0413dd6 c22b03b 0413dd6 c22b03b d9efaab b49d827 d9efaab 5dc54c6 d9efaab c22b03b 3d8743c b49d827 3d8743c | 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 | from langchain import hub, callbacks
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field, UUID4
from typing import List
from pydantic import BaseModel
from typing import List, Optional
import os
class Expectation(BaseModel):
expectation: str = Field(
description="The learning outcome that the course designer has identified for the learner to demonstrate upon successful completion of the module.")
check_question: str = Field(
description="Targeted question that the course designer have developed to assess the learner's understanding of the learning outcomes.")
class Expectations(BaseModel):
expectations: List[Expectation]
class Response(BaseModel):
run_id: UUID4
expectations: List[Expectation]
class Inputs(BaseModel):
course: str
module: str
concepts: List[str]
existing_expectations: Optional[List[Expectation]]
class SuggestExpectations:
def kickoff(self, inputs={}):
self.course = inputs["course"]
self.module = inputs["module"]
self.existing_expectations = inputs["existing_expectations"]
self.concepts = inputs["concepts"]
llm_response = self._get_suggestions()
return {
"run_id": self.run_id,
"expectations": llm_response["expectations"]
}
def _get_suggestions(self):
parser = JsonOutputParser(pydantic_object=Expectations)
chain = self._build_chain()
# Existing Expectations
existing_expectations = []
for expectation in self.existing_expectations:
existing_expectations.append(f"""
Learning Outcome: {expectation.expectation}
Check Question: {expectation.check_question}
""")
existing_expectations_str = ""
if len(existing_expectations) > 0:
existing_expectations_str = "Here are existing Learning Outcomes & Check Questions for the module, don't repeat these learning outcomes:\n ```"
existing_expectations_str += "\n".join(existing_expectations)
existing_expectations_str += "\n```"
with callbacks.collect_runs() as cb:
llm_response = chain.invoke({
"course": self.course, "module": self.module, "concepts": "* " + ("\n* ".join(self.concepts)),
"format_instructions": parser.get_format_instructions(),
"existing_expectations": existing_expectations_str
})
self.run_id = cb.traced_runs[0].id
return llm_response
def _build_chain(self):
parser = JsonOutputParser(pydantic_object=Expectations)
prompt = hub.pull("course_learn_suggest_expectations")
llm = ChatOpenAI(model=os.environ['OPENAI_MODEL'], temperature=0.2)
chain = (prompt | llm | parser).with_config({
"tags": ["course_learn", "suggest_expectations"], "run_name": "Suggest Module Expectations",
"metadata": {
"version": "v1.0.0",
"growth_activity": "course_learn",
"env": os.environ["ENV"],
"model": os.environ["OPENAI_MODEL"],
}
})
return chain
# Example usage
# suggester = SuggestExpectations()
# response = suggester.kickoff(inputs={
# "course": "SQL",
# "module": "Query Optimization Techniques",
# "concepts": [
# "Watch this video https://youtu.be/BHwzDmr6d7s?si=sfFYnd73y9w0EjGB to understand SQL execution order and some optimization techniques.",
# "Watch this video https://youtu.be/FoznjTU929c?si=6M3xUIUwAxE6EbKS to understand SQL explain command usage.",
# "Go over these articles https://intellipaat.com/blog/sql-optimization-techniques/ & https://www.thoughtspot.com/data-trends/data-modeling/optimizing-sql-queries to understand various query optimization techniques."
# ]
# })
# print(response)
|