camel.datagen 包

本页内容

camel.datagen 包#

子包#

子模块#

camel.datagen.cot_datagen 模块#

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

基类：BaseModel

用于结构化代理响应的模型。

一个Pydantic模型类，用于表示来自代理的结构化响应，包括衡量响应质量的相似度分数。

Parameters:: score (float) – 一个介于0到1之间的相似度分数，用于比较当前答案与正确答案的匹配程度。必须在[0, 1]范围内。

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

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)[来源]#

基类：object

用于通过聊天代理交互生成和管理数据的类。

该模块实现了一个复杂的思维链数据生成系统，结合了多种关键算法来产生高质量的推理路径。实现的方法包括：

蒙特卡洛树搜索 (MCTS)
二分查找错误检测
双代理验证系统
解决方案树管理

Parameters:

chat_agent (可选[ChatAgent]) – 用于两个任务的可选单一代理(传统模式)。(默认值:None)
generator_agent (可选[ChatAgent]) – 用于答案生成的可选专用代理。(默认值:None)
verifier_agent (可选[ChatAgent]) - 用于答案验证的可选专用代理。(默认值:None)
golden_answers (Dict[str, str]) – 包含预定义正确答案的字典，用于验证和比较。这是答案验证所必需的。
search_limit (int) - 允许的最大搜索迭代次数。 (默认值:100)

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

使用二分查找来定位解决方案中的第一个错误。该方法使用中英文句子分隔符将解决方案分割成句子，并通过二分查找来找到第一个错误。

Parameters:

question (str) – 需要解决的问题。
solution (str) – 要分析的完整解决方案。

Returns:

解决方案中发现的第一个错误的位置。: 如果没有发现错误（所有句子都正确），则返回-1。

Return type:

int

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

将解决方案过程和结果导出到JSON文件。导出解决方案树、标准答案，

并将时间戳导出到JSON文件。

导出的数据包括： - solutions: 解决方案树

包含中间步骤

golden_answers: 用于验证的参考答案
export_time: 导出时间的ISO格式时间戳

Parameters:

filepath (str, optional) – JSON文件将被保存的路径。 (默认值:'solutions.json')

Returns:

该方法写入文件并记录结果但不: 返回任何值。

Return type:

无

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

从聊天代理获取给定问题的答案。

Parameters:

question (str) – 要询问的问题。
context (str) – 问题的附加上下文。 (默认值:"")

Returns:

生成的答案。

Return type:

字符串

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

从JSON文件或字典中导入问答数据。

Parameters:

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

可以是一个包含问答对的JSON文件路径，或者一个问答对的字典。如果提供的是字符串，则会被视为文件路径。预期格式为： {"question1": "answer1",

"question2": "answer2", ...}

Returns:

如果导入成功则为True，否则为False。

Return type:

布尔值

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

执行蒙特卡洛树搜索以找到最佳解决方案。

流程: a. 选择: 根据之前的评分选择有潜力的部分解决方案 b. 扩展: 使用生成器代理生成新的解决步骤 c. 模拟: 使用相似度评分评估解决方案质量 d. 反向传播: 用新发现更新解决方案树

Parameters:

question (str) – 需要解决的问题。
partial_solution (str) – 当前的部分解决方案。 (默认值:"")

Returns:

当前解决方案与标准答案之间的相似度得分: 解决方案与黄金答案。

Return type:

浮点数

solve(question: str) → str[来源]#

使用多步骤方法解决问题。

解决方案流程遵循以下步骤： 1. 尝试直接解决 - 如果正确，返回解决方案 2. 如果不正确，使用蒙特卡洛树搜索寻找一个好的解决方案 3. 如果解决方案不完美，使用二分查找定位错误 4. 基于正确的部分生成新的解决方案

Parameters:: 问题 (str) – 需要解决的问题。
Returns:: 找到的最佳解决方案。
Return type:: 字符串

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

验证生成的答案在语义上是否等同于给定问题的标准答案。

Parameters:

question (str) – 正在回答的问题。
answer (str) – 需要验证的答案。

Returns:

如果答案与标准答案在语义上匹配则为True: （即核心内容和含义相同，即使措辞不完全一致）。在以下情况下为False： - 如果所提问题在标准答案中不存在 - 如果答案的含义与标准答案不同

Return type:

布尔值

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

基类：BaseModel

结构化验证响应的模型。

一个表示代理验证结果的Pydantic模型类，用于指示答案是否正确。

Parameters:: is_correct (bool) - 布尔值，表示答案是否正确。

is_correct: bool#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

camel.datagen.self_improving_cot 模块#

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

基类：BaseModel

clarity: float#

completeness: float#

correctness: float#

feedback: str#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

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])[来源]#

基类：BaseModel

agent_evaluate_success: bool | None#

boxed_answer_success: bool#

final_trace: str#

id: str | None#

improvement_history: List[TraceIteration]#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

problem: str#

solution: str | None#

type: str | None#

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

基类：BaseModel

class Config[来源]#

基类：object

extra = 'allow'#

feedback: str#

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

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)[来源]#

基类：object

用于生成自学习推理轨迹的流水线采用自我改进的方法论。

这实现了STaR论文提出的方法： 1. 初始推理轨迹生成 2. 自我评估 3. 基于反馈的改进 4. 迭代优化

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)[来源]#

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

评估推理轨迹的质量。

Parameters:

problem (str) – 需要评估的原始问题文本。
trace (str) – 要评估的推理轨迹。
solution (Optional[str]) – 问题的解决方案，如果提供的话。 (默认值: None)

Returns:

评估结果包含：

分数：评估维度及其得分的字典
反馈：详细的改进建议

对于智能体自评，分数将包括： - 正确性：逻辑正确性的评分 - 清晰度：解释清晰度的评分 - 完整性：推理完整性的评分

对于奖励模型评估，分数将取决于模型的评估维度。

Return type:

字典[字符串, 任意类型]

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

在所有问题上执行自我改进的cot流程。

处理问题并返回结果。如果指定了output_path，同时将结果保存到文件。

Parameters:: 合理化 (bool, 可选) – 是否使用合理化。 (默认: False)
Returns:: 已处理结果列表
Return type:: List[Dict[str, Any]]

generate_reasoning_trace(problem: str) → str[来源]#

为给定问题生成初始推理轨迹。

Parameters:: problem (str) – 需要生成推理的问题文本。
Returns:: 生成的推理轨迹。
Return type:: 字符串

generate_reasoning_trace_rejection(problem: str) → str[来源]#

为问题生成多个候选推理轨迹，并根据评估选择最佳的一个。

Parameters:

problem (str) – 用于生成推理轨迹的问题文本。

Returns:

符合质量标准的最佳候选跟踪，或者: 如果没有符合条件的则选择第一个候选。

Return type:

字符串

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

基于反馈生成改进的推理轨迹。

Parameters:

problem (str) – 原始问题文本。
trace (str) - 当前的推理追踪记录。
feedback (str) - 用于改进追踪的反馈信息。
solution (Optional[str]) – 问题的解决方案，如果提供了的话。 (默认值: None)

Returns:

改进的推理追踪。

Return type:

字符串

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

通过自我改进的cot管道处理单个问题。

Parameters:

problem (Dict) – 包含问题文本的问题字典。
合理化 (bool, 可选) – 是否使用合理化。 (默认: False)

Returns:

包含最终追踪记录和历史的结果。

Return type:

ProblemResult

Raises:

ValueError – 如果问题格式无效。

safe_write_json(file_path, data)[来源]#

validate_problem_format(problem: Dict) → None[来源]#

验证问题字典是否符合所需的格式。

Parameters:: problem (Dict) – 要验证的问题字典。
Raises:: ValueError - 如果问题格式无效。

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

基类：BaseModel

evaluation: AgentTraceEvaluation | RewardTraceEvaluation#

iteration: int#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

trace: str#

模块内容#

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)[来源]#

基类：object

用于通过聊天代理交互生成和管理数据的类。

该模块实现了一个复杂的思维链数据生成系统，结合了多种关键算法来产生高质量的推理路径。实现的方法包括：

蒙特卡洛树搜索 (MCTS)
二分查找错误检测
双代理验证系统
解决方案树管理

Parameters:

chat_agent (可选[ChatAgent]) – 用于两个任务的可选单一代理(传统模式)。(默认值:None)
generator_agent (可选[ChatAgent]) – 用于答案生成的可选专用代理。(默认值:None)
verifier_agent (可选[ChatAgent]) - 用于答案验证的可选专用代理。(默认值:None)
golden_answers (Dict[str, str]) – 包含预定义正确答案的字典，用于验证和比较。这是答案验证所必需的。
search_limit (int) - 允许的最大搜索迭代次数。 (默认值:100)

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

使用二分查找来定位解决方案中的第一个错误。该方法使用中英文句子分隔符将解决方案分割成句子，并通过二分查找来找到第一个错误。

Parameters:

question (str) – 需要解决的问题。
solution (str) – 要分析的完整解决方案。

Returns:

在解决方案中发现的第一个错误的位置。: 如果没有发现错误（所有句子都正确），则返回-1。

Return type:

int

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

将解决方案过程和结果导出到JSON文件。导出解决方案树、黄金答案，

并将时间戳导出到JSON文件。

导出的数据包括： - solutions: 解决方案树

包含中间步骤

golden_answers: 用于验证的参考答案
export_time: 导出时间的ISO格式时间戳

Parameters:

filepath (str, optional) – JSON文件将被保存的路径。 (默认值:'solutions.json')

Returns:

该方法写入文件并记录结果但不: 返回任何值。

Return type:

无

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

从聊天代理获取给定问题的答案。

Parameters:

question (str) – 要询问的问题。
context (str) – 问题的附加上下文。 (默认:"")

Returns:

生成的答案。

Return type:

字符串

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

从JSON文件或字典中导入问答数据。

Parameters:

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

可以是一个包含问答对的JSON文件路径，或者一个问答对的字典。如果提供的是字符串，则会被视为文件路径。预期格式为： {"question1": "answer1",

"question2": "answer2", ...}

Returns:

如果导入成功则为True，否则为False。

Return type:

布尔值

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

执行蒙特卡洛树搜索以找到最佳解决方案。

Parameters:

question (str) – 需要解决的问题。
partial_solution (str) - 当前的局部解决方案。 (默认值:"")

Returns:

当前解决方案与标准答案之间的相似度得分: 解决方案与黄金答案。

Return type:

浮点数

solve(question: str) → str[来源]#

使用多步骤方法解决问题。

Parameters:: 问题 (str) – 需要解决的问题。
Returns:: 找到的最佳解决方案。
Return type:: 字符串

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

验证生成的答案在语义上是否等同于给定问题的标准答案。

Parameters:

question (str) – 需要回答的问题。
answer (str) – 需要验证的答案。

Returns:

如果答案与标准答案在语义上匹配则为True: （即核心内容和含义相同，即使措辞不完全一致）。在以下情况下为False： - 如果所提问题在标准答案中不存在 - 如果答案的含义与标准答案不同

Return type:

布尔值

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)[来源]#

基类：object

用于生成自学习推理轨迹的流水线采用自我改进的方法论。

这实现了STaR论文提出的方法： 1. 初始推理轨迹生成 2. 自我评估 3. 基于反馈的改进 4. 迭代优化

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)[来源]#

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

评估推理轨迹的质量。

Parameters:

problem (str) – 需要评估的原始问题文本。
trace (str) – 要评估的推理轨迹。
solution (可选[str]) – 问题的解决方案，如果提供的话。 (默认值: None)

Returns:

评估结果包含：

分数：评估维度及其得分的字典
反馈：详细的改进建议

对于智能体自评，分数将包括： - 正确性：逻辑正确性的评分 - 清晰度：解释清晰度的评分 - 完整性：推理完整性的评分

对于奖励模型评估，分数将取决于模型的评估维度。

Return type:

字典[字符串, 任意类型]

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

在所有问题上执行自我改进的cot流程。

处理问题并返回结果。如果指定了output_path，同时将结果保存到文件。

Parameters:: 合理化 (bool, 可选) – 是否使用合理化。 (默认: False)
Returns:: 已处理结果列表
Return type:: List[Dict[str, Any]]

generate_reasoning_trace(problem: str) → str[来源]#

为给定问题生成初始推理轨迹。

Parameters:: problem (str) – 需要生成推理的问题文本。
Returns:: 生成的推理轨迹。
Return type:: 字符串

generate_reasoning_trace_rejection(problem: str) → str[来源]#

为问题生成多个候选推理轨迹，并根据评估选择最佳的一个。

Parameters:

problem (str) – 用于生成推理轨迹的问题文本。

Returns:

符合质量标准的最佳候选跟踪，或者: 如果没有符合条件的则选择第一个候选。

Return type:

字符串

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

基于反馈生成改进的推理轨迹。

Parameters:

problem (str) – 原始问题文本。
trace (str) - 当前的推理追踪记录。
feedback (str) - 用于改进跟踪的反馈。
solution (Optional[str]) – 问题的解决方案，如果提供的话。 (默认值: None)

Returns:

改进的推理追踪。

Return type:

字符串

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

通过自我改进的cot管道处理单个问题。

Parameters:

problem (Dict) – 包含问题文本的问题字典。
rationalization (bool, optional) – 是否使用rationalization。 (默认: False)

Returns:

包含最终追踪记录和历史的结果。

Return type:

ProblemResult

Raises:

ValueError - 如果问题格式无效。

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

safe_write_json(file_path, data)[来源]#

validate_problem_format(problem: Dict) → None[来源]#

验证问题字典是否符合所需的格式。

Parameters:: problem (Dict) – 要验证的问题字典。
Raises:: ValueError – 如果问题格式无效。

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)[来源]#

基类：object

一个用于生成和管理机器生成任务指令的流水线，结合了人类和机器的任务样本。

Parameters:

agent (ChatAgent) – 用于交互和生成指令的代理。
seed (str) – 人类编写指令的路径。
num_machine_instructions (int) – 要生成的机器生成指令的数量。(默认值：5)
data_output_path (可选[str]) – 保存生成数据的路径。 (默认值:/data_output.json)
human_to_machine_ratio (tuple) – 用于指令生成的人类任务与机器任务的比例。(默认:(6, 2))
instruction_filter (InstructionFilter) – 用于验证生成指令的过滤器。(默认值:None)
filter_config (Optional[Dict[str, Dict[str, Any]]]) - 用于配置在FILE_REGISTRY中注册的过滤函数。 (default:None)
stop_on_first_failure (bool) - 如果为True，则在第一次失败后停止检查过滤器。

construct_data()[来源]#: 将机器生成的任务以JSON格式保存到指定的输出路径。

generate(timeout_minutes=600)[来源]#

执行整个流水线以生成机器指令和实例。

Parameters:: timeout_minutes (int) – 生成过程在超时前运行的最大时间（分钟）。(默认: 600)

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

为给定指令生成实例。

Parameters:

instruction (str) – 用于创建实例的指令。
classification (bool) - 该指令是否为分类任务。

Returns:

以输入-输出格式生成的实例列表。

Return type:

列表[字典]

generate_machine_instances()[来源]#: 根据每台机器任务的分类状态生成实例。

generate_machine_instruction() → List[来源]#

使用代理生成机器指令。

Combines human and machine tasks based on the configured ratio to: 创建一个用于指令生成的提示。

Returns:: 提示词和机器生成的指令。
Return type:: 列表

identify_instruction(instruction: str) → bool[来源]#

判断给定的指令是否属于分类任务。

Parameters:

instruction (str) – 需要分类的指令。

Returns:

如果指令是分类任务则为True，: 否则为False。

Return type:

布尔值

load_seed(path: str)[来源]#

从文件加载种子任务。如果没有提供路径，则默认使用预定义的种子文件。

Parameters:: path (str) – 种子文件的路径。
Raises:: FileNotFoundError – 如果种子文件不存在。

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

将分类任务生成的文本解析为输入-输出对。

Parameters:

generated_text (str) – 由代理生成的原始文本，用于分类任务。

Returns:

一个包含'input'和: 'output'键的字典列表。

Return type:

List[Dict[str, str]]

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

将生成文本解析为输入-输出对，用于非分类任务。

Parameters:

generated_text (str) – 由agent生成的原始文本，用于非分类任务。

Returns:

一个包含'input'和: 'output'键的字典列表。

Return type:

List[Dict[str, str]]

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

从加载的种子中采样指定数量的人工任务。

Parameters:: count (int) – 要采样的人类任务数量。
Returns:: 一份抽样的人类任务列表。
Return type:: 列表[字典]

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

对指定数量的机器任务进行采样。

Parameters:

count (int) – 要采样的机器任务数量。

Returns:

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

Return type:

列表[字典]