Async output processor responsible for distributing engine outputs to corresponding request queues
| 42 | |
| 43 | |
| 44 | class AsyncOutputProcessor: |
| 45 | """Async output processor responsible for distributing engine outputs to corresponding request queues""" |
| 46 | |
| 47 | def __init__(self, data_processor=None): |
| 48 | """ |
| 49 | Args: |
| 50 | data_processor: The data processor created by InputPreprocessor, |
| 51 | used to post-process RequestOutput (decode token_ids, reasoning, tools, etc.). |
| 52 | """ |
| 53 | self.data_processor = data_processor |
| 54 | |
| 55 | def _process_output( |
| 56 | self, |
| 57 | response_item: RequestOutput | Dict[str, Any], |
| 58 | stream: bool = True, |
| 59 | enable_thinking: bool = False, |
| 60 | include_stop_str_in_output: bool = False, |
| 61 | ) -> Dict[str, Any]: |
| 62 | """Process a single response dict via data_processor.process_response_dict. |
| 63 | |
| 64 | This mirrors the behavior of ChatResponseProcessor in the OpenAI serving |
| 65 | path: operate on a dict representation and return a dict. On any error |
| 66 | we fall back to the original dict and ensure ``outputs.text`` exists to |
| 67 | avoid cascading failures. |
| 68 | """ |
| 69 | |
| 70 | try: |
| 71 | processed = self.data_processor.process_response_dict( |
| 72 | response_item, |
| 73 | stream=stream, |
| 74 | enable_thinking=enable_thinking, |
| 75 | include_stop_str_in_output=include_stop_str_in_output, |
| 76 | ) |
| 77 | # Some processors may return None when there is no valid text. |
| 78 | if processed is None: |
| 79 | outputs = response_item.get("outputs") or {} |
| 80 | if "text" not in outputs: |
| 81 | outputs["text"] = "" |
| 82 | response_item["outputs"] = outputs |
| 83 | return response_item |
| 84 | return processed |
| 85 | except Exception: |
| 86 | outputs = response_item.get("outputs") or {} |
| 87 | if "text" not in outputs: |
| 88 | outputs["text"] = "" |
| 89 | response_item["outputs"] = outputs |
| 90 | return response_item |
| 91 | |
| 92 | |
| 93 | class EngineServiceClient: |