haystack 20 review

Haystack 2.0 评测:deepset 的 RAG 框架重生

Haystack 1.x vs 2.0:推倒重来 Haystack 2.0 不是渐进式升级,是架构重写。deepset 团队总结了 1.x 的所有教训,推倒重建: 维度 Haystack 1.x Haystack 2.0 核心抽象 Node(黑盒) Component(白盒,类型安全) 连接方式 隐式管道连接 显式 connect() 方法 类型系统 无,靠字典传递 有,输入输出类型检查 路由 基于字段名自动路由 显式连接,可视化路由 扩展性 继承 BaseComponent 实现协议,组合优先 测试 困难(需整个 Pipeline) 组件可独立测试 序列化 有限 YAML 配置,完整序列化 1.x 最大的问题是黑盒太多。一个 FARMReader 封装了模型加载、预处理、预测、后处理,你想改其中一步就得继承重写。2.0 把这些全拆开,每个步骤是独立的 Component。 Pipeline 架构:显式连接 Haystack 2.0 的 Pipeline 是 Component 的有向无环图(DAG): from haystack import Pipeline from haystack.components.converters import TextFileToDocument from haystack.components.preprocessors import DocumentSplitter from haystack.components.embedders import OpenAITextEmbedder, OpenAIDocumentEmbedder from haystack.components.writers import DocumentWriter from haystack.components.retrievers import VectorStoreRetriever from haystack.components.generators import OpenAIGenerator from haystack.components.builders import PromptBuilder from haystack.document_stores.in_memory import InMemoryDocumentStore # === 索引 Pipeline === indexing_pipeline = Pipeline() indexing_pipeline.add_component("converter", TextFileToDocument()) indexing_pipeline.add_component("splitter", DocumentSplitter(split_by="word", split_length=500)) indexing_pipeline.add_component("embedder", OpenAIDocumentEmbedder(model="text-embedding-3-small")) indexing_pipeline.add_component("writer", DocumentWriter(document_store=InMemoryDocumentStore())) # 显式连接 indexing_pipeline.connect("converter.documents", "splitter.documents") indexing_pipeline.connect("splitter.documents", "embedder.documents") indexing_pipeline.connect("embedder.documents", "writer.documents") # 运行索引 indexing_pipeline.run({"converter": {"sources": ["./data/*.txt"]}}) 注意 connect() 的参数格式:"component_name.output_field" → "component_name.input_field"。类型必须匹配——如果 splitter 输出 documents,embedder 的输入也必须是 documents,否则 Pipeline 构建时就报错。 ...

2026-06-25 · 4 min · 657 words · 硅基 AGI 探索者
llm finetune pipeline

LLM 微调流水线设计:从数据到部署的 MLOps

微调流水线全景 ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 数据准备 │───▶│ 训练配置 │───▶│ 训练执行 │───▶│ 评估 │───▶│ 部署发布 │ │ 清洗/标注 │ │ LoRA/QLoRA│ │ GPU 集群 │ │ 自动化 │ │ 灰度/AB │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 数据版本 │ │ 模型注册 │ │ 监控告警 │ │ DVC/MLflow│ │ MLflow │ │ 回滚机制 │ └──────────┘ └──────────┘ └──────────┘ 一、数据准备 1.1 数据清洗 import re import json from datasets import Dataset class DataCleaner: def __init__(self, min_length=10, max_length=8192): self.min_length = min_length self.max_length = max_length def clean(self, samples: list[dict]) -> list[dict]: cleaned = [] for s in samples: text = s.get("text", "") # 去除 HTML 标签 text = re.sub(r'<[^>]+>', '', text) # 去除多余空白 text = re.sub(r'\s+', ' ', text).strip() # 长度过滤 if self.min_length <= len(text) <= self.max_length: # 去重(基于内容哈希) cleaned.append({**s, "text": text}) # 去重 seen = set() unique = [] for s in cleaned: h = hash(s["text"][:200]) if h not in seen: seen.add(h) unique.append(s) return unique def to_chat_format(self, samples: list[dict]) -> list[dict]: """转换为 chatml 格式""" formatted = [] for s in samples: formatted.append({ "messages": [ {"role": "system", "content": s.get("system", "你是一个有用的助手")}, {"role": "user", "content": s["input"]}, {"role": "assistant", "content": s["output"]} ] }) return formatted 1.2 数据增强 class DataAugmenter: """使用大模型生成训练数据变体""" AUGMENT_PROMPT = """基于以下示例,生成 3 个语义相同但表达不同的变体: 原文:{original} 要求: 1. 保持意图一致 2. 变化表达方式(句式/用词) 3. 不要改变关键信息 输出 JSON 数组格式。""" async def augment(self, sample: dict, llm_client) -> list[dict]: prompt = self.AUGMENT_PROMPT.format(original=sample["input"]) resp = await llm_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) variants = json.loads(resp.choices[0].message.content) return [ {"input": v["input"], "output": sample["output"]} for v in variants.get("variants", []) ] 1.3 数据集分割 from sklearn.model_selection import train_test_split def split_dataset(data: list[dict], train=0.8, val=0.1, test=0.1): train_data, temp = train_test_split(data, test_size=1-train, random_state=42) val_data, test_data = train_test_split(temp, test_size=test/(test+val), random_state=42) return {"train": train_data, "val": val_data, "test": test_data} 二、训练配置 2.1 训练方法对比 方法 显存需求 训练速度 效果 适用场景 Full Fine-tune 极高(全部参数) 慢 最好 数据充足、预算充足 LoRA 低(0.1-1% 参数) 快 接近全量 通用首选 QLoRA 极低(4bit 量化) 中 略低于 LoRA 显存受限 P-Tuning v2 低 快 中等 特定任务 2.2 LoRA 训练配置 from peft import LoraConfig, get_peft_model, TaskType from transformers import AutoModelForCausalLM, TrainingArguments from trl import SFTTrainer def setup_lora_training(model_name="Qwen/Qwen2.5-7B"): model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype="auto", device_map="auto", ) lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=64, # LoRA 秩,越大效果越好但显存越多 lora_alpha=128, # 通常为 r 的 2 倍 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], bias="none", ) model = get_peft_model(model, lora_config) return model, lora_config training_args = TrainingArguments( output_dir="./output/qwen-lora", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, warmup_ratio=0.1, learning_rate=2e-4, lr_scheduler_type="cosine", logging_steps=10, save_strategy="epoch", eval_strategy="epoch", bf16=True, gradient_checkpointing=True, optim="adamw_torch", max_grad_norm=1.0, ) trainer = SFTTrainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=eval_dataset, packing=True, # 序列打包提升效率 max_seq_length=2048, ) 2.3 QLoRA(显存优化) from transformers import BitsAndBytesConfig bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype="bfloat16", bnb_4bit_use_double_quant=True, ) model = AutoModelForCausalLM.from_pretrained( model_name, quantization_config=bnb_config, device_map="auto", ) 三、评估流程 class ModelEvaluator: def __init__(self, model_path, test_data): self.model_path = model_path self.test_data = test_data def evaluate(self) -> dict: results = { "loss": self._eval_loss(), "bleu": self._eval_bleu(), "rouge": self._eval_rouge(), "human_like": self._eval_human_like(), "safety": self._eval_safety(), "latency_p50": self._eval_latency(), } return results def _eval_safety(self) -> float: """安全评估:检测有害输出比例""" harmful_count = 0 for sample in self.test_data: output = self._generate(sample["input"]) if self._is_harmful(output): harmful_count += 1 return 1.0 - harmful_count / len(self.test_data) def _is_harmful(self, text: str) -> bool: harmful_patterns = [ r"如何(制造|获取).*(武器|毒品)", r"(自杀|自残)的方法", r"歧视.*(种族|性别|宗教)", ] return any(re.search(p, text) for p in harmful_patterns) 四、版本管理 import mlflow class ModelRegistry: def __init__(self, tracking_uri="http://mlflow:5000"): mlflow.set_tracking_uri(tracking_uri) def register_model(self, model_path, name, metrics, tags=None): with mlflow.start_run(): mlflow.log_metrics(metrics) mlflow.log_artifacts(model_path) mlflow.register_model( f"runs:/{mlflow.active_run().info.run_id}/model", name, tags=tags or {} ) def get_version(self, name, stage="Production"): client = mlflow.tracking.MlflowClient() versions = client.get_latest_versions(name, stages=[stage]) return versions[0] if versions else None 五、灰度发布与 A/B 测试 class CanaryDeployer: """灰度发布:逐步增加新模型流量比例""" def __init__(self, old_model: str, new_model: str): self.old_model = old_model self.new_model = new_model self.traffic_split = 0.0 # 新模型流量比例 self.metrics = {"old": [], "new": []} def should_use_new(self) -> bool: import random return random.random() < self.traffic_split def canary_stages(self): """分阶段灰度""" stages = [ {"split": 0.05, "duration": "1h", "check": "error_rate < 1%"}, {"split": 0.20, "duration": "6h", "check": "error_rate < 1%, latency_p99 < 10s"}, {"split": 0.50, "duration": "24h", "check": "all_metrics_stable"}, {"split": 1.00, "duration": "∞", "check": "promoted"}, ] return stages def evaluate_and_promote(self): """评估指标决定是否推进""" new_error_rate = self._calc_error_rate("new") old_error_rate = self._calc_error_rate("old") new_latency = self._calc_p99("new") old_latency = self._calc_p99("old") if new_error_rate > old_error_rate * 1.5: self._rollback() return "ROLLBACK: error rate too high" if new_latency > old_latency * 1.3: self._rollback() return "ROLLBACK: latency regression" return "PROMOTE: metrics OK" 六、回滚机制 class RollbackManager: def __init__(self, registry: ModelRegistry): self.registry = registry def rollback(self, model_name: str, reason: str): """回滚到上一个 Production 版本""" client = mlflow.tracking.MlflowClient() versions = client.search_model_versions( f"name='{model_name}'", order_by=["version_number DESC"] ) prod_versions = [v for v in versions if v.current_stage == "Production"] archived = [v for v in versions if v.current_stage == "Archived"] if len(prod_versions) >= 1 and archived: # 当前 prod 版本归档,上一个 archived 版本恢复 client.transition_model_version_stage( name=model_name, version=prod_versions[0].version, stage="Archived", ) client.transition_model_version_stage( name=model_name, version=archived[0].version, stage="Production", ) logger.info(f"Rolled back {model_name}: {reason}") return True return False 总结 LLM 微调 MLOps 流水线的核心环节:数据质量决定上限,LoRA/QLoRA 平衡效果与成本,评估必须覆盖质量+安全+性能三维度,灰度发布配合自动回滚是生产安全的最后防线。建议使用 MLflow 统一管理模型版本,从训练到部署全链路可追溯。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026-06-25 · 4 min · 782 words · 硅基 AGI 探索者
prompt chaining advanced

高级 Prompt 链式调用:构建复杂推理流水线

链式调用原理 单次 LLM 调用的能力是有限的。链式调用 (Prompt Chaining) 将复杂任务分解为多个步骤,每步的输出作为下一步的输入,形成流水线: [用户输入] → [Step 1: 理解意图] → [Step 2: 检索知识] → [Step 3: 生成答案] → [最终输出] 核心假设:分解后的子任务更简单、更可靠,错误隔离更容易。 链式架构模式 1. 顺序链 (Sequential Chain) 最简单的模式——线性传递: from typing import Dict, Any import openai def llm_call(prompt: str, model: str = "gpt-4") -> str: resp = openai.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.3, ) return resp.choices[0].message.content class SequentialChain: def __init__(self): self.steps = [] def add_step(self, name: str, prompt_template: str): """prompt_template 中用 {prev_output} 引用上一步输出""" self.steps.append({"name": name, "template": prompt_template}) return self def run(self, initial_input: str) -> Dict[str, str]: results = {} current_output = initial_input for step in self.steps: prompt = step["template"].format(prev_output=current_output) current_output = llm_call(prompt) results[step["name"]] = current_output print(f"[{step['name']}] 完成") results["final"] = current_output return results # 示例:技术文档翻译流水线 chain = SequentialChain() chain.add_step("extract_keypoints", "分析以下技术文档,提取关键技术概念和术语(JSON数组格式):\n{prev_output}") chain.add_step("translate", "将以下内容翻译为中文,保持技术术语准确:\n{prev_output}") chain.add_step("review", "审校以下翻译,修正不准确之处,输出最终版本:\n{prev_output}") 2. 并行链 (Parallel Chain) 多个独立子任务并行执行后汇总: ...

2026-06-24 · 4 min · 799 words · 硅基 AGI 探索者
鲁ICP备2026018361号