Camel.datagen.self improving cot

​

SelfImprovingCoTPipeline

class SelfImprovingCoTPipeline:

1. Initial reasoning trace generation
2. Self-evaluation
3. Feedback-based improvement
4. Iterative refinement

​

init

def __init__(
    self,
    reason_agent: ChatAgent,
    problems: List[Dict],
    max_iterations: int = 3,
    score_threshold: Union[float, Dict[str, float]] = 0.7,
    rejection_sampling_n: Optional[int] = None,
    evaluate_agent: Optional[ChatAgent] = None,
    reward_model: Optional[BaseRewardModel] = None,
    output_path: Optional[str] = None,
    few_shot_examples: Optional[str] = None,
    batch_size: Optional[int] = None,
    max_workers: Optional[int] = None,
    solution_pattern: str = '\\\\boxed{(.*?)}',
    trace_pattern: Optional[str] = None
):

• reason_agent (ChatAgent): The chat agent used for generating and improving reasoning traces.
• problems (List[Dict]): List of problem dictionaries to process.
• max_iterations (int, optional): Maximum number of improvement iterations. If set to 0, the pipeline will generate an initial trace without any improvement iterations. (default: :obj:3)
• score_threshold (Union[float, Dict[str, float]], optional): Quality threshold. Can be either a single float value applied to average score, or a dictionary mapping score dimensions to their thresholds. For example: {"correctness": 0.8, "coherence": 0.7}. If using reward model and threshold for a dimension is not specified, will use the default value 0.7. (default: :obj:0.7)
• rejection_sampling_n (int, optional): Specifies the number of samples to be drawn using the rejection sampling method, where samples are accepted or rejected based on a predefined condition to achieve a desired distribution. (default: :obj: None)
• evaluate_agent (Optional[ChatAgent]): The chat agent used for evaluating reasoning traces. (default: :obj:None)
• reward_model (BaseRewardModel, optional): Model used to evaluate reasoning traces. If None, uses Agent self-evaluation. (default: :obj:None)
• output_path (str, optional): Output path for saving traces. If None, results will only be returned without saving to file. (default: :obj:None)
• few_shot_examples (str, optional): Examples to use for few-shot generation. (default: :obj:None)
• batch_size (int, optional): Batch size for parallel processing. (default: :obj:None)
• max_workers (int, optional): Maximum number of worker threads. (default: :obj:None)
• solution_pattern (str, optional): Regular expression pattern with one capture group to extract answers from solution text. (default: :obj:r'\\boxed{(.*?)}')
• trace_pattern (str, optional): Regular expression pattern with one capture group to extract answers from trace text. If None, uses the same pattern as solution_pattern. (default: :obj:None)

​

safe_write_json

def safe_write_json(self, file_path, data):

​

clean_json

def clean_json(self, data):

​

_check_score_threshold

def _check_score_threshold(self, scores: Dict[str, float]):

• scores (Dict[str, float]): Dictionary of scores for different dimensions.

​

_generate_feedback

def _generate_feedback(self, scores: Dict[str, float]):

• scores (Dict[str, float]): Dictionary of scores for different dimensions.

​

generate_reasoning_trace

def generate_reasoning_trace(self, problem: str):

• problem (str): The problem text to generate reasoning for.

​

evaluate_trace

def evaluate_trace(
    self,
    problem: str,
    trace: str,
    solution: Optional[str] = None
):

• problem (str): The original problem text to evaluate against.
• trace (str): The reasoning trace to evaluate.
• solution (Optional[str]): The solution to the problem, if provided. (default: :obj:None)

• scores: Dict of evaluation dimensions and their scores
• feedback: Detailed feedback for improvement

• correctness: Score for logical correctness
• clarity: Score for clarity of explanation
• completeness: Score for completeness of reasoning

​

generate_reasoning_trace_rejection

def generate_reasoning_trace_rejection(self, problem: str):

• problem (str): The problem text for generating a reasoning trace.

​

improve_trace

def improve_trace(
    self,
    problem: str,
    trace: str,
    feedback: str,
    solution: Optional[str] = None
):

• problem (str): The original problem text.
• trace (str): The current reasoning trace.
• feedback (str): Feedback for improving the trace.
• solution (Optional[str]): The solution to the problem, if provided. (default: :obj:None)

​

validate_problem_format

def validate_problem_format(self, problem: Dict):

• problem (Dict): Problem dictionary to validate.

​

_check_boxed_answers

def _check_boxed_answers(self, solution: str, trace: str):

• solution (str): The problem solution string.
• trace (str): The reasoning trace string.

​

process_problem

def process_problem(self, problem: Dict, rationalization: bool = False):

• problem (Dict): Problem dictionary containing the problem text.
• rationalization (bool, optional): Whether to use rationalization. (default: :obj:False)

​

generate

def generate(self, rationalization: bool = False):

• rationalization (bool, optional): Whether to use rationalization. (default: :obj:False)