camel.datagen package

camel.datagen package#

Subpackages#

Submodules#

camel.datagen.cot_datagen module#

class camel.datagen.cot_datagen.AgentResponse(*, score: float)[source]#

Bases: BaseModel

Model for structured agent responses.

A Pydantic model class that represents structured responses from agents, including a similarity score that measures the quality of the response.

Parameters:: score (float) – A similarity score between 0 and 1 that compares the current answer to the correct answer. Must be within the range [0, 1].

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'score': FieldInfo(annotation=float, required=True, description='Similarity score between 0 and 1 \n comparing current answer to correct answer', metadata=[typing.Annotated[float, None, Interval(gt=None, ge=0, lt=None, le=1), None, None]])}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

score: float#

class camel.datagen.cot_datagen.CoTDataGenerator(chat_agent: ChatAgent | None = None, *, generator_agent: ChatAgent | None = None, verifier_agent: ChatAgent | None = None, golden_answers: Dict[str, str], search_limit: int = 100)[source]#

Bases: object

Class for generating and managing data through chat agent interactions.

This module implements a sophisticated Chain of Thought data generation system that combines several key algorithms to produce high-quality reasoning paths. Methods implemented:

Monte Carlo Tree Search (MCTS)
Binary Search Error Detection
Dual-Agent Verification System
Solution Tree Management

Parameters:

chat_agent (Optional[ChatAgent]) – Optional single agent for both tasks (legacy mode). (default:None)
generator_agent (Optional[ChatAgent]) – Optional specialized agent for answer generation. (default:None)
verifier_agent (Optional[ChatAgent]) – Optional specialized agent for answer verification. (default:None)
golden_answers (Dict[str, str]) – Dictionary containing pre-defined correct answers for validation and comparison. Required for answer verification.
search_limit (int) – Maximum number of search iterations allowed. (default:100)

binary_search_error(question: str, solution: str) → int[source]#

Use binary search to locate the first error in the solution. This method splits the solution into sentences using both English and Chinese sentence delimiters and performs binary search to find the first error.

Parameters:

question (str) – The question being solved.
solution (str) – The complete solution to analyze.

Returns:

The position of the first error found in the solution.: Returns -1. If no errors are found (all sentences are correct).

Return type:

int

export_solutions(filepath: str = 'solutions.json') → None[source]#

Export the solution process and results to a JSON file. Exports the solution tree, golden answers,

and export timestamp to a JSON file.

The exported data includes: - solutions: The solution tree

with intermediate steps

golden_answers: The reference answers used for verification
export_time: ISO format timestamp of the export

Parameters:

filepath (str, optional) – Path where the JSON file will be saved. (default:'solutions.json')

Returns:

The method writes to a file and logs the result but does not: return any value.

Return type:

None

get_answer(question: str, context: str = '') → str[source]#

Get an answer from the chat agent for a given question.

Parameters:

question (str) – The question to ask.
context (str) – Additional context for the question. (default:"")

Returns:

The generated answer.

Return type:

str

import_qa_from_json(data: str | Dict[str, str]) → bool[source]#

Import question and answer data from either a JSON file or a dictionary.

Parameters:

data (Union[str, Dict[str, str]]) –

Either a path to a JSON file containing QA pairs or a dictionary of question-answer pairs. If a string is provided, it’s treated as a file path. The expected format is: {“question1”: “answer1”,

”question2”: “answer2”, …}

Returns:

True if import was successful, False otherwise.

Return type:

bool

monte_carlo_tree_search(question: str, partial_solution: str = '') → float[source]#

Perform Monte Carlo Tree Search to find the best solution.

Process: a. Selection: Choose promising partial solutions based on previous scores b. Expansion: Generate new solution steps using the generator agent c. Simulation: Evaluate solution quality using similarity scores d. Backpropagation: Update solution tree with new findings

Parameters:

question (str) – The question to solve.
partial_solution (str) – The current partial solution. (default:"")

Returns:

The similarity score between the current: solution and golden answer.

Return type:

float

solve(question: str) → str[source]#

Solve a question using a multi-step approach.

The solution process follows these steps: 1. Try to solve directly - if correct, return the solution 2. If not correct, use Monte Carlo Tree Search to find a good solution 3. If the solution isn’t perfect, use binary search to locate errors 4. Generate a new solution based on the correct part

Parameters:: question (str) – The question to solve.
Returns:: The best solution found.
Return type:: str

verify_answer(question: str, answer: str) → bool[source]#

Verify if a generated answer is semantically equivalent to the golden answer for a given question.

Parameters:

question (str) – The question being answered.
answer (str) – The answer to verify.

Returns:

True if the answer matches the golden answer based on: semantic equivalence (meaning the core content and meaning are the same, even if the exact wording differs). False in the following cases: - If the provided question doesn’t exist in the golden answers - If the answer’s meaning differs from the golden answer

Return type:

bool

class camel.datagen.cot_datagen.VerificationResponse(*, is_correct: bool)[source]#

Bases: BaseModel

Model for structured verification responses.

A Pydantic model class that represents verification results from agents, indicating whether an answer is correct or not.

Parameters:: is_correct (bool) – Boolean indicating if the answer is correct.

is_correct: bool#

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'is_correct': FieldInfo(annotation=bool, required=True, description='Boolean indicating if the answer is correct')}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

camel.datagen.self_improving_cot module#

class camel.datagen.self_improving_cot.AgentTraceEvaluation(*, correctness: float, clarity: float, completeness: float, feedback: str)[source]#

Bases: BaseModel

clarity: float#

completeness: float#

correctness: float#

feedback: str#

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'clarity': FieldInfo(annotation=float, required=True), 'completeness': FieldInfo(annotation=float, required=True), 'correctness': FieldInfo(annotation=float, required=True), 'feedback': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class camel.datagen.self_improving_cot.ProblemResult(*, id: str | None = None, type: str | None = None, problem: str, solution: str | None = None, final_trace: str, agent_evaluate_success: bool | None = None, boxed_answer_success: bool = False, improvement_history: List[TraceIteration])[source]#

Bases: BaseModel

agent_evaluate_success: bool | None#

boxed_answer_success: bool#

final_trace: str#

id: str | None#

improvement_history: List[TraceIteration]#

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'agent_evaluate_success': FieldInfo(annotation=Union[bool, NoneType], required=False, default=None), 'boxed_answer_success': FieldInfo(annotation=bool, required=False, default=False), 'final_trace': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'improvement_history': FieldInfo(annotation=List[TraceIteration], required=True), 'problem': FieldInfo(annotation=str, required=True), 'solution': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'type': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

problem: str#

solution: str | None#

type: str | None#

class camel.datagen.self_improving_cot.RewardTraceEvaluation(*, feedback: str, **data)[source]#

Bases: BaseModel

class Config[source]#

Bases: object

extra = 'allow'#

feedback: str#

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'feedback': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class camel.datagen.self_improving_cot.SelfImprovingCoTPipeline(reason_agent: ChatAgent, problems: List[Dict], max_iterations: int = 3, score_threshold: float | Dict[str, float] = 0.7, rejection_sampling_n: int | None = None, evaluate_agent: ChatAgent | None = None, reward_model: BaseRewardModel | None = None, output_path: str | None = None, few_shot_examples: str | None = None, batch_size: int | None = None, max_workers: int | None = None, solution_pattern: str = '\\\\boxed{(.*?)}', trace_pattern: str | None = None)[source]#

Bases: object

Pipeline for generating self-taught reasoning traces using the self-improving methodology.

This implements the STaR paper’s approach of: 1. Initial reasoning trace generation 2. Self-evaluation 3. Feedback-based improvement 4. Iterative refinement

EVALUATION_TEMPLATE = 'Please evaluate this reasoning trace and \nprovide scores and feedback in valid JSON format.\n\nProblem: {problem}\n\n{solution}\n\nReasoning Trace:\n{trace}\n\nEvaluate for:\n1. Correctness (Is each step logically sound?)\n2. Clarity (Is the explanation clear and well-structured?)\n3. Completeness (Are all necessary steps included?)\n\nRespond ONLY with a JSON object in this exact format:\n{{\n "correctness": <score between 0 and 1>,\n "clarity": <score between 0 and 1>,\n "completeness": <score between 0 and 1>,\n "feedback": "<specific feedback for improvement>"\n}}'#

IMPROVEMENT_TEMPLATE = 'Based on this feedback, generate an \nimproved reasoning trace:\nProblem: {problem}\n\n{solution}\n\nPrevious Trace:\n{trace}\n\nFeedback:\n{feedback}\n\nGenerate a new, improved reasoning trace that addresses the feedback.'#

REASONING_TEMPLATE = "Let's solve this step by step:\nProblem: {problem}\n1. First, let's understand what we're asked\n2. Let's break this down into parts\n3. Let's solve each part systematically\n4. Finally, let's verify our solution\n\n{few_shot_examples}\n\nPlease show your complete reasoning process."#

clean_json(data)[source]#

evaluate_trace(problem: str, trace: str, solution: str | None = None) → Dict[str, Any][source]#

Evaluate the quality of a reasoning trace.

Parameters:

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: None)

Returns:

Evaluation results containing:

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

For Agent self-evaluation, the scores will include: - correctness: Score for logical correctness - clarity: Score for clarity of explanation - completeness: Score for completeness of reasoning

For reward model evaluation, the scores will depend on the model’s evaluation dimensions.

Return type:

Dict[str, Any]

generate(rationalization: bool = False) → List[Dict[str, Any]][source]#

Execute the self-improving cot pipeline on all problems.

Process problems and return results. If output_path is specified, also save results to file.

Parameters:: rationalization (bool, optional) – Whether to use rationalization. (default: False)
Returns:: List of processed results
Return type:: List[Dict[str, Any]]

generate_reasoning_trace(problem: str) → str[source]#

Generate initial reasoning trace for a given problem.

Parameters:: problem (str) – The problem text to generate reasoning for.
Returns:: Generated reasoning trace.
Return type:: str

generate_reasoning_trace_rejection(problem: str) → str[source]#

Generate multiple candidate reasoning traces for a problem and select the best one based on evaluation.

Parameters:

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

Returns:

The best candidate trace that meets quality criteria, or the: first candidate if none qualify.

Return type:

str

improve_trace(problem: str, trace: str, feedback: str, solution: str | None = None) → str[source]#

Generate improved reasoning trace based on feedback.

Parameters:

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: None)

Returns:

Improved reasoning trace.

Return type:

str

process_problem(problem: Dict, rationalization: bool = False) → ProblemResult[source]#

Process a single problem through the self-improving cot pipeline.

Parameters:

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

Returns:

Results with final trace and history.

Return type:

ProblemResult

Raises:

ValueError – If the problem format is invalid.

safe_write_json(file_path, data)[source]#

validate_problem_format(problem: Dict) → None[source]#

Validate that a problem dictionary has the required format.

Parameters:: problem (Dict) – Problem dictionary to validate.
Raises:: ValueError – If the problem format is invalid.

class camel.datagen.self_improving_cot.TraceIteration(*, iteration: int, trace: str, evaluation: AgentTraceEvaluation | RewardTraceEvaluation)[source]#

Bases: BaseModel

evaluation: AgentTraceEvaluation | RewardTraceEvaluation#

iteration: int#

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'evaluation': FieldInfo(annotation=Union[AgentTraceEvaluation, RewardTraceEvaluation], required=True), 'iteration': FieldInfo(annotation=int, required=True), 'trace': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

trace: str#

Module contents#

class camel.datagen.CoTDataGenerator(chat_agent: ChatAgent | None = None, *, generator_agent: ChatAgent | None = None, verifier_agent: ChatAgent | None = None, golden_answers: Dict[str, str], search_limit: int = 100)[source]#

Bases: object

Class for generating and managing data through chat agent interactions.

This module implements a sophisticated Chain of Thought data generation system that combines several key algorithms to produce high-quality reasoning paths. Methods implemented:

Monte Carlo Tree Search (MCTS)
Binary Search Error Detection
Dual-Agent Verification System
Solution Tree Management

Parameters:

chat_agent (Optional[ChatAgent]) – Optional single agent for both tasks (legacy mode). (default:None)
generator_agent (Optional[ChatAgent]) – Optional specialized agent for answer generation. (default:None)
verifier_agent (Optional[ChatAgent]) – Optional specialized agent for answer verification. (default:None)
golden_answers (Dict[str, str]) – Dictionary containing pre-defined correct answers for validation and comparison. Required for answer verification.
search_limit (int) – Maximum number of search iterations allowed. (default:100)

binary_search_error(question: str, solution: str) → int[source]#

Parameters:

question (str) – The question being solved.
solution (str) – The complete solution to analyze.

Returns:

The position of the first error found in the solution.: Returns -1. If no errors are found (all sentences are correct).

Return type:

int

export_solutions(filepath: str = 'solutions.json') → None[source]#

Export the solution process and results to a JSON file. Exports the solution tree, golden answers,

and export timestamp to a JSON file.

The exported data includes: - solutions: The solution tree

with intermediate steps

golden_answers: The reference answers used for verification
export_time: ISO format timestamp of the export

Parameters:

filepath (str, optional) – Path where the JSON file will be saved. (default:'solutions.json')

Returns:

The method writes to a file and logs the result but does not: return any value.

Return type:

None

get_answer(question: str, context: str = '') → str[source]#

Get an answer from the chat agent for a given question.

Parameters:

question (str) – The question to ask.
context (str) – Additional context for the question. (default:"")

Returns:

The generated answer.

Return type:

str

import_qa_from_json(data: str | Dict[str, str]) → bool[source]#

Import question and answer data from either a JSON file or a dictionary.

Parameters:

data (Union[str, Dict[str, str]]) –

”question2”: “answer2”, …}

Returns:

True if import was successful, False otherwise.

Return type:

bool

monte_carlo_tree_search(question: str, partial_solution: str = '') → float[source]#

Perform Monte Carlo Tree Search to find the best solution.

Parameters:

question (str) – The question to solve.
partial_solution (str) – The current partial solution. (default:"")

Returns:

The similarity score between the current: solution and golden answer.

Return type:

float

solve(question: str) → str[source]#

Solve a question using a multi-step approach.

Parameters:: question (str) – The question to solve.
Returns:: The best solution found.
Return type:: str

verify_answer(question: str, answer: str) → bool[source]#

Verify if a generated answer is semantically equivalent to the golden answer for a given question.

Parameters:

question (str) – The question being answered.
answer (str) – The answer to verify.

Returns:

True if the answer matches the golden answer based on: semantic equivalence (meaning the core content and meaning are the same, even if the exact wording differs). False in the following cases: - If the provided question doesn’t exist in the golden answers - If the answer’s meaning differs from the golden answer

Return type:

bool

class camel.datagen.SelfImprovingCoTPipeline(reason_agent: ChatAgent, problems: List[Dict], max_iterations: int = 3, score_threshold: float | Dict[str, float] = 0.7, rejection_sampling_n: int | None = None, evaluate_agent: ChatAgent | None = None, reward_model: BaseRewardModel | None = None, output_path: str | None = None, few_shot_examples: str | None = None, batch_size: int | None = None, max_workers: int | None = None, solution_pattern: str = '\\\\boxed{(.*?)}', trace_pattern: str | None = None)[source]#

Bases: object

Pipeline for generating self-taught reasoning traces using the self-improving methodology.

This implements the STaR paper’s approach of: 1. Initial reasoning trace generation 2. Self-evaluation 3. Feedback-based improvement 4. Iterative refinement

EVALUATION_TEMPLATE = 'Please evaluate this reasoning trace and \nprovide scores and feedback in valid JSON format.\n\nProblem: {problem}\n\n{solution}\n\nReasoning Trace:\n{trace}\n\nEvaluate for:\n1. Correctness (Is each step logically sound?)\n2. Clarity (Is the explanation clear and well-structured?)\n3. Completeness (Are all necessary steps included?)\n\nRespond ONLY with a JSON object in this exact format:\n{{\n "correctness": <score between 0 and 1>,\n "clarity": <score between 0 and 1>,\n "completeness": <score between 0 and 1>,\n "feedback": "<specific feedback for improvement>"\n}}'#

IMPROVEMENT_TEMPLATE = 'Based on this feedback, generate an \nimproved reasoning trace:\nProblem: {problem}\n\n{solution}\n\nPrevious Trace:\n{trace}\n\nFeedback:\n{feedback}\n\nGenerate a new, improved reasoning trace that addresses the feedback.'#

REASONING_TEMPLATE = "Let's solve this step by step:\nProblem: {problem}\n1. First, let's understand what we're asked\n2. Let's break this down into parts\n3. Let's solve each part systematically\n4. Finally, let's verify our solution\n\n{few_shot_examples}\n\nPlease show your complete reasoning process."#

clean_json(data)[source]#

evaluate_trace(problem: str, trace: str, solution: str | None = None) → Dict[str, Any][source]#

Evaluate the quality of a reasoning trace.

Parameters:

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: None)

Returns:

Evaluation results containing:

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

For Agent self-evaluation, the scores will include: - correctness: Score for logical correctness - clarity: Score for clarity of explanation - completeness: Score for completeness of reasoning

For reward model evaluation, the scores will depend on the model’s evaluation dimensions.

Return type:

Dict[str, Any]

generate(rationalization: bool = False) → List[Dict[str, Any]][source]#

Execute the self-improving cot pipeline on all problems.

Process problems and return results. If output_path is specified, also save results to file.

Parameters:: rationalization (bool, optional) – Whether to use rationalization. (default: False)
Returns:: List of processed results
Return type:: List[Dict[str, Any]]

generate_reasoning_trace(problem: str) → str[source]#

Generate initial reasoning trace for a given problem.

Parameters:: problem (str) – The problem text to generate reasoning for.
Returns:: Generated reasoning trace.
Return type:: str

generate_reasoning_trace_rejection(problem: str) → str[source]#

Generate multiple candidate reasoning traces for a problem and select the best one based on evaluation.

Parameters:

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

Returns:

The best candidate trace that meets quality criteria, or the: first candidate if none qualify.

Return type:

str

improve_trace(problem: str, trace: str, feedback: str, solution: str | None = None) → str[source]#

Generate improved reasoning trace based on feedback.

Parameters:

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: None)

Returns:

Improved reasoning trace.

Return type:

str

process_problem(problem: Dict, rationalization: bool = False) → ProblemResult[source]#

Process a single problem through the self-improving cot pipeline.

Parameters:

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

Returns:

Results with final trace and history.

Return type:

ProblemResult

Raises:

ValueError – If the problem format is invalid.

reasoning_traces: List[Dict[str, Any]]#

safe_write_json(file_path, data)[source]#

validate_problem_format(problem: Dict) → None[source]#

Validate that a problem dictionary has the required format.

Parameters:: problem (Dict) – Problem dictionary to validate.
Raises:: ValueError – If the problem format is invalid.

class camel.datagen.SelfInstructPipeline(agent: ChatAgent, seed: str, num_machine_instructions: int = 5, data_output_path: str | None = './data_output.json', human_to_machine_ratio: tuple = (6, 2), instruction_filter: InstructionFilter | None = None, filter_config: Dict[str, Dict[str, Any]] | None = None, stop_on_first_failure: bool = False)[source]#

Bases: object

A pipeline to generate and manage machine-generated instructions for tasks, combining human and machine task samples.

Parameters:

agent (ChatAgent) – The agent used to interact and generate instructions.
seed (str) – The path to the human-written instructions.
num_machine_instructions (int) – Number of machine-generated instructions to generate. (default:5)
data_output_path (Optional[str]) – Path to save the generated data. (default:/data_output.json)
human_to_machine_ratio (tuple) – Ratio of human to machine tasks used for instruction generation. (default:(6, 2))
instruction_filter (InstructionFilter) – A filter to validate generated instructions. (default:None)
filter_config (Optional[Dict[str, Dict[str, Any]]]) – configuration for the filter functions registered in FILE_REGISTRY. (default:None)
stop_on_first_failure (bool) – If True, stops checking filters after the first failure.

construct_data()[source]#: Save the machine-generated tasks to the specified output path in JSON format.

generate(timeout_minutes=600)[source]#

Execute the entire pipeline to generate machine instructions and instances.

Parameters:: timeout_minutes (int) – Maximum time in minutes to run the generation process before timing out. (default: 600)

generate_machine_instance(instruction: str, classification: bool) → list[dict][source]#

Generate instances for a given instruction.

Parameters:

instruction (str) – The instruction to create instances for.
classification (bool) – Whether the instruction is a classification task.

Returns:

A list of generated instances in input-output format.

Return type:

List[dict]

generate_machine_instances()[source]#: Generate instances for each machine task based on its classification status.

generate_machine_instruction() → List[source]#

Generate a machine instruction using the agent.

Combines human and machine tasks based on the configured ratio to: create a prompt for instruction generation.

Returns:: The prompt and a machine-generated instruction.
Return type:: List

identify_instruction(instruction: str) → bool[source]#

Determine if the given instruction is a classification task.

Parameters:

instruction (str) – The instruction to classify.

Returns:

True if the instruction is a classification task,: otherwise False.

Return type:

bool

load_seed(path: str)[source]#

Load seed tasks from a file. Defaults to a predefined seed file if no path is provided.

Parameters:: path (str) – Path to the seed file.
Raises:: FileNotFoundError – If the seed file does not exist.

parse_classification_output(generated_text: str) → List[Dict[str, str]][source]#

Parse the generated text for classification tasks into input-output pairs.

Parameters:

generated_text (str) – The raw text generated by the agent for classification tasks.

Returns:

A list of dictionaries with ‘input’ and: ’output’ keys.

Return type:

List[Dict[str, str]]

parse_non_classification_output(generated_text: str) → List[Dict[str, str]][source]#

Parse the generated text for non-classification tasks into input-output pairs.

Parameters:

generated_text (str) – The raw text generated by the agent for non-classification tasks.

Returns:

A list of dictionaries with ‘input’ and: ’output’ keys.

Return type:

List[Dict[str, str]]

sample_human_tasks(count: int) → List[dict][source]#

Sample a specified number of human tasks from the loaded seed.

Parameters:: count (int) – Number of human tasks to sample.
Returns:: A list of sampled human tasks.
Return type:: List[dict]

sample_machine_tasks(count: int) → List[dict][source]#

Sample a specified number of machine tasks.

Parameters:

count (int) – Number of machine tasks to sample.

Returns:

A list of sampled machine tasks, with placeholders if: insufficient tasks are available.

Return type:

List[dict]

camel.datagen package

On this page

camel.datagen package#

Subpackages#

Submodules#

camel.datagen.cot_datagen module#

camel.datagen.self_improving_cot module#

Module contents#