Spaces:
Runtime error
Runtime error
| 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) | |