|
from typing import Dict, Any |
|
|
|
from aiflows.base_flows import CircularFlow |
|
from aiflows.utils import logging |
|
from abc import ABC, abstractmethod |
|
|
|
|
|
logging.set_verbosity_debug() |
|
log = logging.get_logger(__name__) |
|
|
|
|
|
class ContentWriterFlow(CircularFlow, ABC): |
|
"""This is an abstract class for writing content (plan, code) |
|
The ContentWriterFlow is made of a controller and a branching executor. |
|
Each time the controller is called, the controller decides whether to write content |
|
or to finish. If the content writer executor is called, the executor will write content |
|
in an interactive way, finally, the user is able to give feedback to the content, so that |
|
the controller can decide whether to write content again or to finish. |
|
|
|
*Configuration Parameters*: |
|
- `name` |
|
- `description` |
|
- `max_round` |
|
- `subflows_config`: |
|
- `Controller` (dict): The controller that decides whether to write content or to finish. |
|
- `Executor` (dict): A branching flow, we configure the specific executor in the subflows of the executor. |
|
- `early_exit_key`: The key of the early exit variable in the output payload of the executor. |
|
- `topology`: The topology of the subflows, this describes the I/O interface instances. |
|
|
|
*Input Interface*: |
|
- `goal` |
|
|
|
*Output Interface*: |
|
- `answer` |
|
- `status` |
|
""" |
|
@abstractmethod |
|
def _on_reach_max_round(self): |
|
""" |
|
should update flow state dictionary about the output variables and status. |
|
""" |
|
pass |
|
|
|
@abstractmethod |
|
@CircularFlow.output_msg_payload_processor |
|
def detect_finish_or_continue(self, output_payload: Dict[str, Any], src_flow) -> Dict[str, Any]: |
|
""" |
|
1. Writing content to file; |
|
2. Finish and early exit. |
|
""" |
|
pass |
|
|
