Generate a refined prompt for the reasoning task based on the user's input and the JSON schema. Args: state (dict): The current state of the graph. The input keys will be used to fetch the correct data from the state. Returns
(self, state: dict)
| 54 | self.output_schema = node_config.get("schema") |
| 55 | |
| 56 | def execute(self, state: dict) -> dict: |
| 57 | """ |
| 58 | Generate a refined prompt for the reasoning task based |
| 59 | on the user's input and the JSON schema. |
| 60 | |
| 61 | Args: |
| 62 | state (dict): The current state of the graph. The input keys will be used |
| 63 | to fetch the correct data from the state. |
| 64 | |
| 65 | Returns: |
| 66 | dict: The updated state with the output key containing the generated answer. |
| 67 | |
| 68 | Raises: |
| 69 | KeyError: If the input keys are not found in the state, indicating |
| 70 | that the necessary information for generating an answer is missing. |
| 71 | """ |
| 72 | |
| 73 | self.logger.info(f"--- Executing {self.node_name} Node ---") |
| 74 | |
| 75 | user_prompt = state["user_prompt"] |
| 76 | |
| 77 | self.simplefied_schema = transform_schema(self.output_schema.schema()) |
| 78 | |
| 79 | if self.additional_info is not None: |
| 80 | prompt = PromptTemplate( |
| 81 | template=TEMPLATE_REASONING_WITH_CONTEXT, |
| 82 | partial_variables={ |
| 83 | "user_input": user_prompt, |
| 84 | "json_schema": str(self.simplefied_schema), |
| 85 | "additional_context": self.additional_info, |
| 86 | }, |
| 87 | ) |
| 88 | else: |
| 89 | prompt = PromptTemplate( |
| 90 | template=TEMPLATE_REASONING, |
| 91 | partial_variables={ |
| 92 | "user_input": user_prompt, |
| 93 | "json_schema": str(self.simplefied_schema), |
| 94 | }, |
| 95 | ) |
| 96 | |
| 97 | output_parser = StrOutputParser() |
| 98 | |
| 99 | chain = prompt | self.llm_model | output_parser |
| 100 | refined_prompt = chain.invoke({}) |
| 101 | |
| 102 | state.update({self.output[0]: refined_prompt}) |
| 103 | return state |
nothing calls this directly
no test coverage detected