25：Few-Shot到Zero-Shot的自动化转换技巧

class ExampleExtractor:
    def __init__(self):
        """初始化示例提取器"""
        pass
    
    def extract_examples(self, few_shot_prompt):
        """从Few-Shot提示词中提取示例"""
        # 分割提示词，识别示例部分
        lines = few_shot_prompt.strip().split('\n')
        examples = []
        in_example = False
        current_example = {}
        
        for line in lines:
            line = line.strip()
            
            # 识别示例开始
            if line.startswith('示例') or line.startswith('Example'):
                in_example = True
                current_example = {'input': '', 'output': ''}
            # 识别输入部分
            elif in_example and (line.startswith('输入:') or line.startswith('Input:')):
                current_example['input'] = line.split(':', 1)[1].strip()
            # 识别输出部分
            elif in_example and (line.startswith('输出:') or line.startswith('Output:')):
                current_example['output'] = line.split(':', 1)[1].strip()
                examples.append(current_example)
                in_example = False
        
        return examples
    
    def extract_prompt_template(self, few_shot_prompt):
        """提取提示词模板（非示例部分）"""
        lines = few_shot_prompt.strip().split('\n')
        template_lines = []
        in_example = False
        
        for line in lines:
            # 检查是否进入示例部分
            if line.strip().startswith('示例') or line.strip().startswith('Example'):
                in_example = True
                continue
            # 检查是否退出示例部分
            if in_example and (line.strip().startswith('输出:') or line.strip().startswith('Output:')):
                in_example = False
                continue
            # 只添加非示例部分
            if not in_example:
                template_lines.append(line)
        
        return '\n'.join(template_lines).strip()

# 使用示例	extractor = ExampleExtractor()
few_shot_prompt = """
请根据输入生成相应的输出。

示例 1:
输入: 苹果
输出: 水果

示例 2:
输入: 胡萝卜
输出: 蔬菜

示例 3:
输入: 鸡肉
输出: 肉类

现在请处理以下输入：
输入: {input}
"""
examples = extractor.extract_examples(few_shot_prompt)
template = extractor.extract_prompt_template(few_shot_prompt)
print("示例:", examples)
print("模板:", template)

class ExampleAnalyzer:
    def __init__(self):
        """初始化示例分析器"""
        pass
    
    def analyze_examples(self, examples):
        """分析示例，提取模式"""
        if not examples:
            return {}
        
        # 分析输入输出的模式
        input_patterns = []
        output_patterns = []
        
        for example in examples:
            input_text = example.get('input', '')
            output_text = example.get('output', '')
            
            # 提取输入模式
            input_patterns.append(self._extract_pattern(input_text))
            # 提取输出模式
            output_patterns.append(self._extract_pattern(output_text))
        
        # 分析输入输出之间的关系
        relations = self._analyze_relations(examples)
        
        return {
            'input_patterns': input_patterns,
            'output_patterns': output_patterns,
            'relations': relations,
            'example_count': len(examples)
        }
    
    def _extract_pattern(self, text):
        """提取文本模式"""
        # 简单的模式提取，实际应用中可以使用更复杂的NLP技术
        patterns = {
            'length': len(text),
            'words': len(text.split()),
            'contains_number': any(char.isdigit() for char in text),
            'contains_punctuation': any(char in '.,!?:;"\'()[]{}' for char in text),
            'is_uppercase': text.isupper(),
            'is_lowercase': text.islower(),
            'starts_with_uppercase': text[0].isupper() if text else False
        }
        return patterns
    
    def _analyze_relations(self, examples):
        """分析输入输出之间的关系"""
        relations = []
        
        for example in examples:
            input_text = example.get('input', '')
            output_text = example.get('output', '')
            
            # 简单的关系分析，实际应用中可以使用更复杂的NLP技术
            relation = {
                'input_length': len(input_text),
                'output_length': len(output_text),
                'length_ratio': len(output_text) / len(input_text) if len(input_text) > 0 else 0,
                'input_words': len(input_text.split()),
                'output_words': len(output_text.split()),
                'words_ratio': len(output_text.split()) / len(input_text.split()) if len(input_text.split()) > 0 else 0
            }
            relations.append(relation)
        
        return relations

# 使用示例
analyzer = ExampleAnalyzer()
analysis_result = analyzer.analyze_examples(examples)
print("示例分析结果:", analysis_result)

class RuleGenerator:
    def __init__(self):
        """初始化规则生成器"""
        pass
    
    def generate_rules(self, examples):
        """从示例中生成规则"""
        if not examples:
            return []
        
        rules = []
        
        # 分析示例，提取规则
        for i, example in enumerate(examples):
            input_text = example.get('input', '')
            output_text = example.get('output', '')
            
            # 生成规则
            rule = {
                'id': i + 1,
                'input': input_text,
                'output': output_text,
                'rule': self._generate_rule(input_text, output_text)
            }
            rules.append(rule)
        
        # 合并相似规则
        merged_rules = self._merge_rules(rules)
        
        return merged_rules
    
    def _generate_rule(self, input_text, output_text):
        """为单个示例生成规则"""
        # 简单的规则生成，实际应用中可以使用更复杂的NLP技术
        # 这里假设输入是一个实体，输出是其类别
        return f"将'{input_text}'归类为'{output_text}'"
    
    def _merge_rules(self, rules):
        """合并相似规则"""
        if not rules:
            return []
        
        # 简单的规则合并，实际应用中可以使用更复杂的聚类技术
        merged = []
        seen_outputs = set()
        
        for rule in rules:
            output = rule['output']
            if output not in seen_outputs:
                seen_outputs.add(output)
                # 收集具有相同输出的所有输入
                similar_inputs = [r['input'] for r in rules if r['output'] == output]
                # 生成合并后的规则
                merged_rule = {
                    'id': len(merged) + 1,
                    'inputs': similar_inputs,
                    'output': output,
                    'rule': f"将{'、'.join(similar_inputs)}等归类为'{output}'"
                }
                merged.append(merged_rule)
        
        return merged

# 使用示例
generator = RuleGenerator()
rules = generator.generate_rules(examples)
print("生成的规则:", rules)

class RuleGeneralizer:
    def __init__(self):
        """初始化规则泛化器"""
        pass
    
    def generalize_rules(self, rules):
        """泛化规则"""
        if not rules:
            return []
        
        generalized_rules = []
        
        for rule in rules:
            generalized_rule = {
                'id': rule['id'],
                'original_rule': rule['rule'],
                'generalized_rule': self._generalize_rule(rule)
            }
            generalized_rules.append(generalized_rule)
        
        return generalized_rules
    
    def _generalize_rule(self, rule):
        """泛化单个规则"""
        # 简单的规则泛化，实际应用中可以使用更复杂的NLP技术
        # 这里将具体的示例替换为通用描述
        output = rule['output']
        return f"将属于{output}类别的实体识别并归类为'{output}'"

# 使用示例
generalizer = RuleGeneralizer()
generalized_rules = generalizer.generalize_rules(rules)
print("泛化后的规则:", generalized_rules)

class ZeroShotConverter:
    def __init__(self):
        """初始化Zero-Shot转换器"""
        pass
    
    def convert_to_zero_shot(self, few_shot_prompt):
        """将Few-Shot提示词转换为Zero-Shot提示词"""
        # 1. 提取示例和模板
        extractor = ExampleExtractor()
        examples = extractor.extract_examples(few_shot_prompt)
        template = extractor.extract_prompt_template(few_shot_prompt)
        
        # 2. 分析示例
        analyzer = ExampleAnalyzer()
        analysis_result = analyzer.analyze_examples(examples)
        
        # 3. 生成规则
        generator = RuleGenerator()
        rules = generator.generate_rules(examples)
        
        # 4. 泛化规则
        generalizer = RuleGeneralizer()
        generalized_rules = generalizer.generalize_rules(rules)
        
        # 5. 构建Zero-Shot提示词
        zero_shot_prompt = self._build_zero_shot_prompt(template, generalized_rules)
        
        return {
            'original_prompt': few_shot_prompt,
            'zero_shot_prompt': zero_shot_prompt,
            'examples': examples,
            'rules': rules,
            'generalized_rules': generalized_rules,
            'analysis_result': analysis_result
        }
    
    def _build_zero_shot_prompt(self, template, generalized_rules):
        """构建Zero-Shot提示词"""
        # 构建规则描述
        rules_description = "根据以下规则处理输入：\n"
        for rule in generalized_rules:
            rules_description += f"- {rule['generalized_rule']}\n"
        
        # 替换模板中的示例部分
        # 这里简单地在模板前添加规则描述，实际应用中可以更智能地整合
        zero_shot_prompt = rules_description + "\n" + template
        
        return zero_shot_prompt

# 使用示例
converter = ZeroShotConverter()
conversion_result = converter.convert_to_zero_shot(few_shot_prompt)
print("转换结果:")
print("Zero-Shot提示词:", conversion_result['zero_shot_prompt'])

class AdvancedZeroShotConverter:
    def __init__(self, model=None):
        """初始化高级Zero-Shot转换器"""
        self.model = model
    
    def convert_to_zero_shot(self, few_shot_prompt):
        """使用模型辅助将Few-Shot提示词转换为Zero-Shot提示词"""
        # 1. 提取示例和模板
        extractor = ExampleExtractor()
        examples = extractor.extract_examples(few_shot_prompt)
        template = extractor.extract_prompt_template(few_shot_prompt)
        
        if not examples:
            return {
                'original_prompt': few_shot_prompt,
                'zero_shot_prompt': few_shot_prompt,
                'examples': [],
                'error': 'No examples found'
            }
        
        # 2. 使用模型生成通用规则
        rules_prompt = f"""
        从以下示例中提取通用规则，不需要具体示例，只需要通用描述：
        
        {few_shot_prompt}
        
        请输出通用规则，格式为：
        1. 规则1
        2. 规则2
        ...
        """
        
        if self.model:
            rules_response = self.model(rules_prompt)
            rules = self._parse_rules(rules_response)
        else:
            # 回退到简单规则生成
            generator = RuleGenerator()
            rules = generator.generate_rules(examples)
            generalizer = RuleGeneralizer()
            generalized_rules = generalizer.generalize_rules(rules)
            rules = [r['generalized_rule'] for r in generalized_rules]
        
        # 3. 构建Zero-Shot提示词
        zero_shot_prompt = self._build_zero_shot_prompt(template, rules)
        
        return {
            'original_prompt': few_shot_prompt,
            'zero_shot_prompt': zero_shot_prompt,
            'examples': examples,
            'rules': rules
        }
    
    def _parse_rules(self, rules_response):
        """解析模型生成的规则"""
        lines = rules_response.strip().split('\n')
        rules = []
        
        for line in lines:
            line = line.strip()
            if line and (line.startswith('1.') or line.startswith('2.') or line.startswith('3.') or line.startswith('- ')):
                # 去除序号和前缀
                rule = line.lstrip('1234567890. -')
                rules.append(rule)
        
        return rules
    
    def _build_zero_shot_prompt(self, template, rules):
        """构建Zero-Shot提示词"""
        # 构建规则描述
        rules_description = "请根据以下规则处理输入：\n"
        for i, rule in enumerate(rules, 1):
            rules_description += f"{i}. {rule}\n"
        
        # 替换模板中的示例部分
        zero_shot_prompt = rules_description + "\n" + template
        
        return zero_shot_prompt

# 使用示例
from langchain.chat_models import ChatOpenAI

model = ChatOpenAI(model="gpt-4-turbo")
advanced_converter = AdvancedZeroShotConverter(model)
advanced_conversion_result = advanced_converter.convert_to_zero_shot(few_shot_prompt)
print("高级转换结果:")
print("Zero-Shot提示词:", advanced_conversion_result['zero_shot_prompt'])

class PerformanceEvaluator:
    def __init__(self, model):
        """初始化性能评估器"""
        self.model = model
    
    def evaluate(self, few_shot_prompt, zero_shot_prompt, test_cases):
        """评估Zero-Shot提示词的性能"""
        few_shot_results = []
        zero_shot_results = []
        
        # 测试Few-Shot提示词
        for test_case in test_cases:
            test_prompt = few_shot_prompt.replace('{input}', test_case['input'])
            result = self.model(test_prompt)
            few_shot_results.append({
                'input': test_case['input'],
                'expected': test_case['expected'],
                'actual': result,
                'correct': self._is_correct(result, test_case['expected'])
            })
        
        # 测试Zero-Shot提示词
        for test_case in test_cases:
            test_prompt = zero_shot_prompt.replace('{input}', test_case['input'])
            result = self.model(test_prompt)
            zero_shot_results.append({
                'input': test_case['input'],
                'expected': test_case['expected'],
                'actual': result,
                'correct': self._is_correct(result, test_case['expected'])
            })
        
        # 计算评估指标
        metrics = self._calculate_metrics(few_shot_results, zero_shot_results)
        
        return {
            'few_shot_results': few_shot_results,
            'zero_shot_results': zero_shot_results,
            'metrics': metrics
        }
    
    def _is_correct(self, actual, expected):
        """判断结果是否正确"""
        # 简单的正确性判断，实际应用中可以使用更复杂的评估方法
        return expected.lower() in actual.lower()
    
    def _calculate_metrics(self, few_shot_results, zero_shot_results):
        """计算评估指标"""
        # 计算准确率
        few_shot_correct = sum(1 for r in few_shot_results if r['correct'])
        zero_shot_correct = sum(1 for r in zero_shot_results if r['correct'])
        accuracy = (zero_shot_correct / few_shot_correct) * 100 if few_shot_correct > 0 else 0
        
        # 计算召回率
        recall = (zero_shot_correct / len(zero_shot_results)) * 100 if len(zero_shot_results) > 0 else 0
        
        # 计算泛化能力（假设测试用例中包含未见过的输入）
        # 这里简化处理，实际应用中需要区分见过和未见过的输入
        generalization = accuracy
        
        # 计算简洁性
        # 这里需要传入原始的few_shot_prompt和zero_shot_prompt
        # 简化处理，假设长度比为0.5
        conciseness = 0.5
        
        # 计算一致性
        # 简化处理，假设一致性为100%
        consistency = 100.0
        
        return {
            'accuracy': accuracy,
            'recall': recall,
            'generalization': generalization,
            'conciseness': conciseness,
            'consistency': consistency
        }

# 使用示例
test_cases = [
    {'input': '香蕉', 'expected': '水果'},
    {'input': '西红柿', 'expected': '蔬菜'},
    {'input': '牛肉', 'expected': '肉类'},
    {'input': '橙子', 'expected': '水果'},
    {'input': '土豆', 'expected': '蔬菜'}
]

evaluator = PerformanceEvaluator(model)
evaluation_result = evaluator.evaluate(
    few_shot_prompt,
    advanced_conversion_result['zero_shot_prompt'],
    test_cases
)
print("评估结果:", evaluation_result['metrics'])

# 情感分析的Few-Shot提示词
sentiment_few_shot = """
请分析以下文本的情感倾向，输出正面、负面或中性。

示例 1:
输入: 这部电影非常精彩，演员表演出色，剧情紧凑，我非常喜欢。
输出: 正面

示例 2:
输入: 这个餐厅的服务态度很差，食物也不好吃，非常失望。
输出: 负面

示例 3:
输入: 今天天气一般，不冷不热，适合出门。
输出: 中性

现在请分析以下文本：
输入: {input}
"""

# 转换为Zero-Shot
converter = AdvancedZeroShotConverter(model)
conversion_result = converter.convert_to_zero_shot(sentiment_few_shot)
zero_shot_prompt = conversion_result['zero_shot_prompt']
print("情感分析Zero-Shot提示词:", zero_shot_prompt)

# 测试
 test_cases = [
    {'input': '这个产品质量很好，性价比高，值得购买。', 'expected': '正面'},
    {'input': '服务态度恶劣，下次再也不来了。', 'expected': '负面'},
    {'input': '今天的会议内容一般，没有太多新意。', 'expected': '中性'}
]

evaluation_result = evaluator.evaluate(sentiment_few_shot, zero_shot_prompt, test_cases)
print("情感分析评估结果:", evaluation_result['metrics'])

# 实体识别的Few-Shot提示词
ner_few_shot = """
请从以下文本中识别出人名、地名和组织名。

示例 1:
输入: 张三在北京大学工作，住在北京。
输出: 人名：张三；地名：北京大学、北京；组织名：无

示例 2:
输入: 李四是阿里巴巴的员工，来自上海。
输出: 人名：李四；地名：上海；组织名：阿里巴巴

示例 3:
输入: 王五在腾讯工作，毕业于清华大学。
输出: 人名：王五；地名：清华大学；组织名：腾讯

现在请识别以下文本：
输入: {input}
"""

# 转换为Zero-Shot
conversion_result = converter.convert_to_zero_shot(ner_few_shot)
zero_shot_prompt = conversion_result['zero_shot_prompt']
print("实体识别Zero-Shot提示词:", zero_shot_prompt)

# 测试
test_cases = [
    {'input': '赵六在百度工作，来自广州。', 'expected': '人名：赵六；地名：广州；组织名：百度'},
    {'input': '钱七是华为的工程师，毕业于浙江大学。', 'expected': '人名：钱七；地名：浙江大学；组织名：华为'}
]

evaluation_result = evaluator.evaluate(ner_few_shot, zero_shot_prompt, test_cases)
print("实体识别评估结果:", evaluation_result['metrics'])

# 问答系统的Few-Shot提示词
qa_few_shot = """
请根据以下上下文回答问题。

示例 1:
上下文: 北京是中国的首都，也是中国的政治、文化和国际交往中心。
问题: 中国的首都是哪里？
答案: 北京

示例 2:
上下文: 苹果是一种常见的水果，富含维生素C和纤维素。
问题: 苹果富含什么？
答案: 维生素C和纤维素

示例 3:
上下文: 爱因斯坦是著名的物理学家，提出了相对论。
问题: 爱因斯坦提出了什么理论？
答案: 相对论

现在请回答以下问题：
上下文: {context}
问题: {question}
答案:
"""

# 转换为Zero-Shot
conversion_result = converter.convert_to_zero_shot(qa_few_shot)
zero_shot_prompt = conversion_result['zero_shot_prompt']
print("问答系统Zero-Shot提示词:", zero_shot_prompt)

# 测试
test_cases = [
    {
        'input': '上下文: 上海是中国的经济中心，也是国际化大都市。\n问题: 中国的经济中心是哪里？',
        'expected': '上海'
    },
    {
        'input': '上下文: 比特币是一种数字货币，由中本聪发明。\n问题: 比特币是谁发明的？',
        'expected': '中本聪'
    }
]

# 适配测试用例格式
adapted_test_cases = []
for test_case in test_cases:
    # 提取上下文和问题
    parts = test_case['input'].split('\n')
    context = parts[0].split(': ', 1)[1]
    question = parts[1].split(': ', 1)[1]
    # 构建完整输入
    full_input = f"上下文: {context}\n问题: {question}"
    adapted_test_cases.append({
        'input': full_input,
        'expected': test_case['expected']
    })

# 构建适配的Few-Shot和Zero-Shot提示词
adapted_few_shot = qa_few_shot
adapted_zero_shot = zero_shot_prompt

evaluation_result = evaluator.evaluate(adapted_few_shot, adapted_zero_shot, adapted_test_cases)
print("问答系统评估结果:", evaluation_result['metrics'])

from langchain.chat_models import ChatOpenAI

# 初始化模型
model = ChatOpenAI(model="gpt-4-turbo")

# 初始化转换器
converter = AdvancedZeroShotConverter(model)

# 定义Few-Shot提示词
few_shot_prompt = """
请根据输入生成相应的输出。

示例 1:
输入: 示例输入1
输出: 示例输出1

示例 2:
输入: 示例输入2
输出: 示例输出2

现在请处理以下输入：
输入: {input}
"""

# 转换为Zero-Shot
conversion_result = converter.convert_to_zero_shot(few_shot_prompt)
zero_shot_prompt = conversion_result['zero_shot_prompt']

# 测试
 test_input = "测试输入"
 test_prompt = zero_shot_prompt.replace('{input}', test_input)
 result = model(test_prompt)
print(f"输入: {test_input}")
print(f"输出: {result}")

def batch_convert(few_shot_prompts):
    """批量转换Few-Shot提示词为Zero-Shot提示词"""
    results = []
    converter = AdvancedZeroShotConverter(model)
    
    for i, prompt in enumerate(few_shot_prompts):
        print(f"转换提示词 {i+1}/{len(few_shot_prompts)}")
        result = converter.convert_to_zero_shot(prompt)
        results.append(result)
    
    return results

# 批量转换示例
few_shot_prompts = [
    """请分析以下文本的情感倾向，输出正面、负面或中性。

示例 1:
输入: 这部电影非常精彩。
输出: 正面

现在请分析：
输入: {input}
""",
    """请从以下文本中识别出人名。

示例 1:
输入: 张三在北京大学工作。
输出: 张三

现在请识别：
输入: {input}
"""
]

batch_results = batch_convert(few_shot_prompts)
for i, result in enumerate(batch_results):
    print(f"\n提示词 {i+1} 转换结果:")
    print(result['zero_shot_prompt'])

# 安装依赖
pip install langchain openai

# 配置环境变量
export OPENAI_API_KEY=your-api-key

# 运行示例
python zero_shot_conversion.py