(
self,
llm_output: str,
run_manager: AsyncCallbackManagerForChainRun,
)
| 232 | return {self.output_key: answer} |
| 233 | |
| 234 | async def _aprocess_llm_result( |
| 235 | self, |
| 236 | llm_output: str, |
| 237 | run_manager: AsyncCallbackManagerForChainRun, |
| 238 | ) -> Dict[str, str]: |
| 239 | await run_manager.on_text(llm_output, color="green", verbose=self.verbose) |
| 240 | llm_output = llm_output.strip() |
| 241 | text_match = re.search(r"^```text(.*?)```", llm_output, re.DOTALL) |
| 242 | if text_match: |
| 243 | expression = text_match.group(1) |
| 244 | output = self._evaluate_expression(expression) |
| 245 | await run_manager.on_text("\nAnswer: ", verbose=self.verbose) |
| 246 | await run_manager.on_text(output, color="yellow", verbose=self.verbose) |
| 247 | answer = "Answer: " + output |
| 248 | elif llm_output.startswith("Answer:"): |
| 249 | answer = llm_output |
| 250 | elif "Answer:" in llm_output: |
| 251 | answer = "Answer: " + llm_output.split("Answer:")[-1] |
| 252 | else: |
| 253 | raise ValueError(f"unknown format from LLM: {llm_output}") |
| 254 | return {self.output_key: answer} |
| 255 | |
| 256 | def _call( |
| 257 | self, |
no test coverage detected