4. 生成与提示

本节导读：掌握RAG生成核心技术，优化提示词工程提升回答质量

学习目标

• 理解大语言模型在RAG中的作用机制
• 掌握提示词设计和优化的核心技巧
• 学会构建企业级提示词模板
• 实现生成质量的评估和改进

核心概念

生成系统架构

Haystack生成系统采用模块化设计，支持多种LLM集成和提示词模板，确保生成回答的相关性和准确性。

提示词工程核心

提示词工程是RAG系统的灵魂，通过精心设计的提示模板引导模型生成高质量、格式规范的回答。

LLM集成

支持的模型类型

from haystack.components.generators import OpenAIGenerator

# OpenAI模型
openai_generator = OpenAIGenerator(
    model="gpt-4",
    api_key=os.getenv("OPENAI_API_KEY"),
    generation_kwargs={
        "temperature": 0.7,
        "max_tokens": 1000,
        "top_p": 0.9,
        "frequency_penalty": 0.1,
        "presence_penalty": 0.1
    }
)

# 本地模型
from haystack.components.generators import HuggingFaceGenerator

local_generator = HuggingFaceGenerator(
    model="microsoft/DialoGPT-medium",
    generation_kwargs={
        "max_length": 100,
        "num_return_sequences": 1,
        "temperature": 0.8
    }
)
​

模型配置优化

class OptimizedLLMGenerator:
    def __init__(self, model_name, config):
        self.model = self._initialize_model(model_name)
        self.config = config
    
    def _initialize_model(self, model_name):
        """根据模型类型初始化"""
        if "gpt" in model_name.lower():
            return OpenAIGenerator(
                model=model_name,
                api_key=os.getenv("OPENAI_API_KEY"),
                generation_kwargs=self.config
            )
        elif "claude" in model_name.lower():
            return ClaudeGenerator(
                model=model_name,
                api_key=os.getenv("ANTHROPIC_API_KEY"),
                generation_kwargs=self.config
            )
        else:
            return HuggingFaceGenerator(
                model=model_name,
                generation_kwargs=self.config
            )
​

成本控制策略

class CostAwareGenerator:
    def __init__(self, generator, cost_calculator):
        self.generator = generator
        self.cost_calculator = cost_calculator
    
    def run(self, prompt):
        # 估算成本
        estimated_cost = self.cost_calculator.estimate(prompt)
        
        # 检查预算
        if estimated_cost > self.budget:
            # 使用更经济的模型
            return self.fallback_generator.run(prompt)
        
        # 执行生成
        result = self.generator.run(prompt)
        
        # 记录实际成本
        actual_cost = self.cost_calculator.actual(result)
        self.cost_calculator.log(actual_cost)
        
        return result
​

提示词模板设计

基础提示模板

from haystack.components.generators import PromptTemplate

# 简单问答提示
qa_template = PromptTemplate(
    prompt="""请根据以下提供的信息回答用户问题：

信息上下文：
{{context}}

用户问题：{{query}}

请提供准确、完整的回答：""",
    input_variables=["context", "query"]
)

# 多轮对话提示
conversation_template = PromptTemplate(
    prompt="""对话历史：
{{conversation_history}}

当前用户输入：{{user_input}}

请基于对话历史和当前输入，继续对话并保持上下文连贯性：""",
    input_variables=["conversation_history", "user_input"]
)
​

企业级提示模板

class EnterprisePromptTemplate:
    def __init__(self):
        self.templates = {
            "qa": self._create_qa_template(),
            "summary": self._create_summary_template(),
            "analysis": self._create_analysis_template(),
            "code": self._create_code_template()
        }
    
    def _create_qa_template(self):
        return PromptTemplate(
            prompt="""你是一位专业的RAG系统助手，需要根据提供的文档信息回答用户问题。

请遵循以下要求：
1. 严格基于提供的文档上下文回答
2. 如果文档中没有相关信息，明确说明"文档中未找到相关信息"
3. 回答要准确、简洁、有条理
4. 重要概念用专业术语表述
5. 避免编造信息

文档上下文：
{{context}}

用户问题：{{query}}

专业回答：""",
            input_variables=["context", "query"]
        )
    
    def _create_summary_template(self):
        return PromptTemplate(
            prompt="""请对以下文档进行专业摘要，要求：

1. 提取核心观点和关键信息
2. 保持逻辑结构清晰
3. 概括主要结论
4. 控制在200字以内

文档内容：
{{document}}

专业摘要：""",
            input_variables=["document"]
        )
​

动态提示生成

class DynamicPromptGenerator:
    def __init__(self, template_repo):
        self.template_repo = template_repo
        self.query_analyzer = QueryAnalyzer()
    
    def generate_prompt(self, query, context, user_profile=None):
        # 分析查询类型
        query_type = self.query_analyzer.analyze(query)
        
        # 选择基础模板
        base_template = self.template_repo.get_template(query_type["template"])
        
        # 根据用户画像个性化
        if user_profile:
            base_template = self.personalize_template(base_template, user_profile)
        
        # 动态插入上下文信息
        prompt = self._insert_context(base_template, context, query_type)
        
        # 添加约束条件
        prompt = self._add_constraints(prompt, query_type)
        
        return prompt
​

提示词优化技术

提示词链

class PromptChain:
    def __init__(self, steps):
        self.steps = steps
    
    def run(self, initial_query, context):
        current_query = initial_query
        
        for step in self.steps:
            # 为当前步骤生成提示
            prompt = step.generate_prompt(current_query, context)
            
            # 执行生成
            result = step.generator.run(prompt)
            
            # 更新查询或上下文
            current_query = step.update_state(current_query, result)
        
        return current_query
​

提示词微调

from transformers import AutoTokenizer, AutoModelForCausalLM

class PromptTuner:
    def __init__(self, model_name):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
    
    def optimize_prompt(self, original_prompt, examples, target_response):
        """基于示例优化提示词"""
        
        # 构建训练数据
        training_data = []
        for example in examples:
            training_data.append({
                "input": example["prompt"],
                "output": example["response"]
            })
        
        # 添加目标示例
        training_data.append({
            "input": original_prompt,
            "output": target_response
        })
        
        # 训练提示词优化模型
        self._train_model(training_data)
        
        # 生成优化后的提示词
        optimized_prompt = self._generate_optimized_prompt(original_prompt)
        
        return optimized_prompt
​

提示词版本管理

class PromptVersionManager:
    def __init__(self):
        self.versions = {}
        self.current_version = "v1.0"
    
    def save_version(self, prompt_name, prompt_content, metadata):
        """保存提示词版本"""
        version_id = f"{prompt_name}_v{self._get_next_version()}"
        
        self.versions[version_id] = {
            "content": prompt_content,
            "metadata": metadata,
            "timestamp": datetime.now(),
            "version_id": version_id
        }
        
        return version_id
    
    def rollback_version(self, prompt_name, target_version):
        """回滚到指定版本"""
        if target_version in self.versions:
            self.versions[prompt_name] = self.versions[target_version]
            return True
        return False
    
    def get_version_history(self, prompt_name):
        """获取版本历史"""
        versions = []
        for version_id, version_data in self.versions.items():
            if version_id.startswith(prompt_name):
                versions.append({
                    "version": version_id,
                    "timestamp": version_data["timestamp"],
                    "metadata": version_data["metadata"]
                })
        return sorted(versions, key=lambda x: x["timestamp"], reverse=True)
​

生成质量评估

自动化评估指标

class GenerationEvaluator:
    def __init__(self):
        self.metrics = {
            "relevance": RelevanceMetric(),
            "coherence": CoherenceMetric(),
            "fluency": FluencyMetric(),
            "accuracy": AccuracyMetric(),
            "completeness": CompletenessMetric()
        }
    
    def evaluate(self, generated_text, reference_text=None, query=None):
        """评估生成质量"""
        
        scores = {}
        
        # 计算各项指标
        for metric_name, metric in self.metrics.items():
            if reference_text and metric.needs_reference:
                score = metric.calculate(generated_text, reference_text)
            elif query and metric.needs_query:
                score = metric.calculate(generated_text, query)
            else:
                score = metric.calculate(generated_text)
            
            scores[metric_name] = score
        
        # 综合评分
        overall_score = self._calculate_overall_score(scores)
        scores["overall"] = overall_score
        
        return scores
​

人工评估框架

class HumanEvaluationFramework:
    def __init__(self):
        self.evaluators = []
        self.evaluation_criteria = [
            "准确性",
            "相关性", 
            "完整性",
            "可读性",
            "专业性"
        ]
    
    def add_evaluator(self, evaluator):
        """添加评估者"""
        self.evaluators.append(evaluator)
    
    def conduct_evaluation(self, test_cases):
        """执行人工评估"""
        results = []
        
        for case in test_cases:
            case_results = {
                "case_id": case["id"],
                "query": case["query"],
                "generated_response": case["generated_response"],
                "reference_response": case.get("reference_response"),
                "scores": {}
            }
            
            # 每个评估者评分
            for evaluator in self.evaluators:
                scores = evaluator.evaluate(
                    case["generated_response"],
                    case.get("reference_response"),
                    case["query"]
                )
                case_results["scores"][evaluator.name] = scores
            
            results.append(case_results)
        
        return results
​

企业级生成优化

多模型融合

class MultiModelFusion:
    def __init__(self, models):
        self.models = models
        self.weights = self._calculate_model_weights()
    
    def generate(self, prompt):
        """多模型融合生成"""
        results = []
        
        # 并行生成
        for model in self.models:
            result = model.generate(prompt)
            results.append(result)
        
        # 融合结果
        fused_result = self._fuse_results(results)
        
        return fused_result
    
    def _fuse_results(self, results):
        """融合多个模型结果"""
        # 简单投票机制
        vote_counts = {}
        for result in results:
            key = result["answer"]
            vote_counts[key] = vote_counts.get(key, 0) + 1
        
        # 得票最多的答案
        best_answer = max(vote_counts.items(), key=lambda x: x[1])[0]
        
        return {
            "answer": best_answer,
            "confidence": vote_counts[best_answer] / len(results),
            "alternative_answers": self._get_alternatives(vote_counts)
        }
​

实时反馈优化

class RealTimeOptimizer:
    def __init__(self, generator, feedback_collector):
        self.generator = generator
        self.feedback_collector = feedback_collector
    
    def generate_with_feedback(self, prompt, user_id):
        """带实时反馈的生成"""
        
        # 初始生成
        initial_result = self.generator.generate(prompt)
        
        # 收集用户反馈
        feedback = self.feedback_collector.collect(user_id, initial_result)
        
        # 基于反馈优化
        if feedback["needs_improvement"]:
            optimized_result = self._optimize_based_on_feedback(
                initial_result, feedback
            )
        else:
            optimized_result = initial_result
        
        return optimized_result
​

本节小结

本章节深入讲解了Haystack生成系统的核心技术，包括LLM集成、提示词设计和优化技术。通过生成质量评估和实时反馈机制，可以持续改进RAG系统的回答质量。下一章我们将进入实战环节，构建完整的RAG应用系统。

延伸阅读

• 提示词工程最佳实践
• 大语言模型评估方法
• 企业级AI应用指南

关键词：提示词工程, LLM集成, 生成优化, Haystack, 企业级应用
难度：进阶
预计阅读：90分钟