Self-Refine扩展指南：如何添加新的任务类型

Self-Refine扩展指南：如何添加新的任务类型

【免费下载链接】self-refineLLMs can generate feedback on their work, use it to improve the output, and repeat this process iteratively.项目地址: https://gitcode.com/gh_mirrors/se/self-refine

想要让Self-Refine项目支持你的自定义任务吗？这篇完整指南将带你一步步了解如何为这个强大的AI自我优化框架添加新的任务类型。Self-Refine的核心思想是让语言模型能够自我生成反馈并迭代改进输出，现在你可以轻松扩展它来处理任何你感兴趣的任务！

为什么需要扩展Self-Refine？

Self-Refine项目已经内置了多个任务类型，包括：

• 首字母缩略词生成(Acronym Generation)
• 数学问题求解(GSM-8k)
• 代码可读性改进(Code Readability)
• 通用句子生成(Commongen)
• 情感反转(Sentiment Reversal)
• 程序效率提升(PIE)

但你可能需要处理特定领域的任务，比如创意写作优化、技术文档改进、代码审查建议等。通过添加新的任务类型，你可以让Self-Refine框架适应你的具体需求！

Self-Refine任务架构解析

每个Self-Refine任务都遵循相同的三阶段架构：

1. 初始化阶段 (Init)

负责生成任务的初始输出。对应的文件位于src/你的任务名/task_init.py

2. 反馈阶段 (Feedback)

评估当前输出并生成改进建议。对应的文件位于src/你的任务名/feedback.py

3. 迭代阶段 (Iterate)

基于反馈生成改进后的输出。对应的文件位于src/你的任务名/task_iterate.py

图：Self-Refine的三阶段迭代优化流程

快速开始：创建你的第一个任务

步骤1：创建任务目录结构

首先，在src/目录下创建你的任务文件夹：

mkdir src/your_task_name

步骤2：实现初始化组件

创建src/your_task_name/task_init.py文件：

import pandas as pd from src.utils import Prompt from prompt_lib.backends import openai_api class YourTaskInit(Prompt): def __init__(self, prompt_examples: str, engine: str) -> None: super().__init__( question_prefix="输入: ", # 根据你的任务调整 answer_prefix="输出: ", # 根据你的任务调整 intra_example_sep="\n\n", inter_example_sep="\n\n###\n\n", ) self.engine = engine self.setup_prompt_from_examples_file(prompt_examples) def setup_prompt_from_examples_file(self, examples_path: str) -> str: # 加载示例数据，构建提示词 examples_df = pd.read_json(examples_path, orient="records", lines=True) prompt_parts = [] for _, row in examples_df.iterrows(): prompt_parts.append(f"输入: {row['input']}\n\n输出: {row['output']}") self.prompt = self.inter_example_sep.join(prompt_parts) self.prompt = self.prompt + self.inter_example_sep def __call__(self, input_text: str) -> str: query = f"{self.prompt}{self.question_prefix}{input_text}{self.intra_example_sep}{self.answer_prefix}" output = openai_api.OpenaiAPIWrapper.call( prompt=query, engine=self.engine, max_tokens=300, stop_token="###", temperature=0.7, ) return openai_api.OpenaiAPIWrapper.get_first_response(output).strip()

步骤3：实现反馈组件

创建src/your_task_name/feedback.py文件：

from src.utils import Prompt from prompt_lib.backends import openai_api class YourTaskFeedback(Prompt): def __init__(self, engine: str, prompt_examples: str) -> None: super().__init__( question_prefix="", answer_prefix="", intra_example_sep="\n\n", inter_example_sep="\n\n### END ###\n\n", engine=engine, ) self.setup_prompt_from_examples_file(prompt_examples) def setup_prompt_from_examples_file(self, examples_path: str) -> str: with open(examples_path, "r") as f: self.prompt = f.read() def __call__(self, input_text: str, current_output: str): query = self.make_query(input_text, current_output) output = openai_api.OpenaiAPIWrapper.call( prompt=query, engine=self.engine, max_tokens=600, stop_token="### END", temperature=0.7, ) feedback = openai_api.OpenaiAPIWrapper.get_first_response(output) return {"feedback": feedback, "improved_output": ""} def make_query(self, input_text: str, current_output: str): return f"{self.prompt}输入: {input_text}\n\n当前输出: {current_output}\n\n反馈: "

步骤4：实现迭代组件

创建src/your_task_name/task_iterate.py文件：

from src.utils import Prompt from prompt_lib.backends import openai_api class YourTaskIterate(Prompt): def __init__(self, engine: str, prompt_examples: str) -> None: super().__init__( question_prefix="", answer_prefix="", intra_example_sep="\n\n", inter_example_sep="\n\n### END ###\n\n", engine=engine, ) self.setup_prompt_from_examples_file(prompt_examples) def __call__(self, input_text: str, current_output: str, feedback: str): query = self.make_query(input_text, current_output, feedback) output = openai_api.OpenaiAPIWrapper.call( prompt=query, engine=self.engine, max_tokens=300, stop_token="### END", temperature=0.7, ) improved_output = openai_api.OpenaiAPIWrapper.get_first_response(output) return improved_output.strip() def make_query(self, input_text: str, current_output: str, feedback: str): return f"{self.prompt}输入: {input_text}\n\n当前输出: {current_output}\n\n反馈: {feedback}\n\n改进后的输出: "

图：任务组件之间的数据流和交互关系

步骤5：创建主运行文件

创建src/your_task_name/run.py文件：

from src.your_task_name.task_init import YourTaskInit from src.your_task_name.feedback import YourTaskFeedback from src.your_task_name.task_iterate import YourTaskIterate from src.utils import retry_parse_fail_prone_cmd @retry_parse_fail_prone_cmd def iterative_your_task(input_text: str, max_attempts: int = 5): # 初始化组件 task_init = YourTaskInit( prompt_examples="data/prompt/your_task_name/init.jsonl", engine="gpt-3.5-turbo" ) # 反馈组件 task_feedback = YourTaskFeedback( prompt_examples="data/prompt/your_task_name/feedback.txt", engine="gpt-3.5-turbo" ) # 迭代组件 task_iterate = YourTaskIterate( prompt_examples="data/prompt/your_task_name/iterate.txt", engine="gpt-3.5-turbo" ) n_attempts = 0 current_output = None while n_attempts < max_attempts: if n_attempts == 0: current_output = task_init(input_text) else: feedback_result = task_feedback(input_text, current_output) current_output = task_iterate(input_text, current_output, feedback_result["feedback"]) n_attempts += 1 return current_output if __name__ == "__main__": result = iterative_your_task("你的输入文本", max_attempts=3) print(f"最终输出: {result}")

步骤6：准备提示词数据

在data/prompt/your_task_name/目录下创建三个文件：

1. init.jsonl- 初始化示例数据
2. feedback.txt- 反馈提示词模板
3. iterate.txt- 迭代提示词模板

示例的init.jsonl文件格式：

{"input": "示例输入1", "output": "示例输出1"} {"input": "示例输入2", "output": "示例输出2"}

实战案例：创建创意写作优化任务

让我们通过一个具体例子来演示如何创建一个创意写作优化任务：

1. 定义任务目标

• 输入: 一段创意写作草稿
• 输出: 优化后的创意写作
• 反馈: 对写作风格、情节连贯性、角色发展等方面的建议

2. 创建提示词模板

data/prompt/creative_writing/feedback.txt:

你是一个专业的创意写作编辑。请分析以下写作草稿，并提供具体的改进建议。 写作草稿: {input_text} 当前版本: {current_output} 请从以下方面提供反馈： 1. 情节连贯性 2. 角色发展 3. 语言表达 4. 情感张力 反馈:

3. 实现核心逻辑

在src/creative_writing/feedback.py中：

def __call__(self, input_text: str, current_output: str): query = self.make_query(input_text, current_output) output = openai_api.OpenaiAPIWrapper.call( prompt=query, engine=self.engine, max_tokens=800, # 创意写作需要更多tokens stop_token="### END", temperature=0.7, ) feedback = openai_api.OpenaiAPIWrapper.get_first_response(output) # 解析反馈，提取具体建议 return self.parse_feedback(feedback)

图：创意写作任务的自我优化流程示例

高级技巧：优化你的任务实现

1. 自定义停止条件

在run.py中，你可以添加智能停止条件：

def should_stop(feedback: str, current_output: str, iteration: int) -> bool: # 当反馈表示输出已经很好时停止 if "excellent" in feedback.lower() or "perfect" in feedback.lower(): return True # 达到最大迭代次数时停止 if iteration >= max_attempts: return True # 连续多次没有改进时停止 if no_improvement_for > 2: return True return False

2. 添加评估指标

为你的任务添加量化评估：

def evaluate_output(input_text: str, output: str, feedback: str) -> dict: return { "coherence_score": calculate_coherence(output), "creativity_score": calculate_creativity(output), "improvement_potential": estimate_improvement(feedback) }

3. 集成外部工具

如果你的任务需要特定工具，可以在组件中集成：

class SpecializedTaskInit(Prompt): def __init__(self, prompt_examples: str, engine: str, external_tool: Any): super().__init__(...) self.external_tool = external_tool def __call__(self, input_text: str): # 使用外部工具预处理输入 processed_input = self.external_tool.preprocess(input_text) # 生成初始输出 return super().__call__(processed_input)

调试和测试你的任务

1. 单元测试

为每个组件创建测试：

def test_task_init(): task = YourTaskInit(prompt_examples="test_data.jsonl", engine="gpt-3.5-turbo") result = task("测试输入") assert isinstance(result, str) assert len(result) > 0

2. 集成测试

测试整个迭代流程：

def test_full_iteration(): result = iterative_your_task("测试输入", max_attempts=3) assert isinstance(result, str) # 验证输出质量 assert quality_check(result) > threshold

3. 性能监控

添加日志记录来监控任务性能：

import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def iterative_your_task(input_text: str, max_attempts: int = 5): logger.info(f"开始处理任务: {input_text[:50]}...") for i in range(max_attempts): logger.info(f"迭代 {i+1}/{max_attempts}") # ... 任务逻辑 logger.info(f"任务完成，最终输出长度: {len(result)}") return result

常见问题解答

Q: 如何选择合适的提示词模板？

A: 参考现有任务的提示词设计，如src/acronym/和src/gsm/目录中的示例。关键是保持一致的格式和清晰的结构。

Q: 如何处理复杂的输出格式？

A: 可以在反馈组件中添加输出解析逻辑，确保模型返回结构化的反馈和建议。

Q: 如何优化迭代效率？

A: 考虑添加早期停止机制，当连续多次迭代没有明显改进时提前终止。

Q: 我的任务需要特殊的数据预处理怎么办？

A: 在初始化组件中添加预处理步骤，或者在单独的preprocess.py文件中实现。

最佳实践总结

1. 保持一致性: 遵循项目现有的命名约定和代码结构
2. 模块化设计: 每个组件只负责单一职责
3. 充分的测试: 为每个组件编写单元测试
4. 清晰的文档: 在代码中添加详细的注释
5. 逐步迭代: 先实现基本功能，再逐步添加高级特性

图：遵循最佳实践的任务扩展效果对比

开始你的任务扩展之旅

现在你已经掌握了为Self-Refine添加新任务类型的所有知识！从简单的文本处理任务开始，逐步扩展到复杂的创意和技术任务。记住，Self-Refine的强大之处在于它的灵活性 - 你可以将它应用到几乎任何需要迭代改进的场景。

🚀立即行动: 选择一个你感兴趣的任务，按照本文的步骤开始实现。遇到问题时，可以参考现有任务的实现，或者查看项目的官方文档。

💡小贴士: 从修改现有任务开始是最快的学习方式。尝试复制src/acronym/目录，修改它以处理你的特定任务需求。

通过扩展Self-Refine，你不仅能为项目做出贡献，还能创建出强大的AI工具来解决实际问题。祝你在AI自我优化的探索之旅中取得成功！

【免费下载链接】self-refineLLMs can generate feedback on their work, use it to improve the output, and repeat this process iteratively.项目地址: https://gitcode.com/gh_mirrors/se/self-refine