Invoke the chain on a single input :param chain_input: The input for the chain :return: A dict with the defined json schema
(self, chain_input: dict)
| 55 | self.callback = get_dummy_callback |
| 56 | |
| 57 | def invoke(self, chain_input: dict) -> dict: |
| 58 | """ |
| 59 | Invoke the chain on a single input |
| 60 | :param chain_input: The input for the chain |
| 61 | :return: A dict with the defined json schema |
| 62 | """ |
| 63 | with self.callback() as cb: |
| 64 | try: |
| 65 | result = self.chain.invoke(chain_input) |
| 66 | if self.parser_func is not None: |
| 67 | result = self.parser_func(result) |
| 68 | except Exception as e: |
| 69 | if e.http_status == 401: |
| 70 | raise e |
| 71 | else: |
| 72 | logging.error('Error in chain invoke: {}'.format(e.user_message)) |
| 73 | result = None |
| 74 | self.accumulate_usage += cb.total_cost |
| 75 | return result |
| 76 | |
| 77 | async def retry_operation(self, tasks): |
| 78 | """ |
no outgoing calls
no test coverage detected