(
self,
inputs: Dict[str, str],
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
)
| 268 | return self._process_llm_result(llm_output, _run_manager) |
| 269 | |
| 270 | async def _acall( |
| 271 | self, |
| 272 | inputs: Dict[str, str], |
| 273 | run_manager: Optional[AsyncCallbackManagerForChainRun] = None, |
| 274 | ) -> Dict[str, str]: |
| 275 | _run_manager = run_manager or AsyncCallbackManagerForChainRun.get_noop_manager() |
| 276 | await _run_manager.on_text(inputs[self.input_key]) |
| 277 | llm_output = await self.llm_chain.apredict( |
| 278 | question=inputs[self.input_key], |
| 279 | stop=["```output"], |
| 280 | callbacks=_run_manager.get_child(), |
| 281 | ) |
| 282 | return await self._aprocess_llm_result(llm_output, _run_manager) |
| 283 | |
| 284 | @property |
| 285 | def _chain_type(self) -> str: |
nothing calls this directly
no test coverage detected