第 1 章
全景与编排
场景还原
凌晨两点，数据团队的批处理任务在第 42351 份 PDF 上停了。这份文件是一个损坏的扫描件，解析器抛了异常，然后整个脚本退出。前面四个小时的处理全废了，日志里只有一条 KeyError，没有告诉你它卡在哪一步、已经处理了多少、哪些文件是好的。
这是「手工编排」的典型死法：把 ingest、parse、extract、build 四个模块用一段脚本串起来，每步的输入输出靠变量传递，错误靠 try/except 兜底，顺序靠人脑保证。当数据源从一个变成三十个、当同一条事实有多个来源互相打架、当老板要求「每一句结论都能指回它的来源」时，这段脚本会以各种方式崩掉。
本章回答的是这门课的第一个问题：数据从入口到知识库，编排层怎么保证步骤顺序、状态管理与失败处理。答案不在任何一段魔法代码里，而在三个明确的对象上：一个总入口 Semantica、一个显式的 Pipeline 对象、一个执行它的 ExecutionEngine。我们先从数据结构的底子看起。
逐行精读
先看全景。整条链路从总入口出发，落到四个对象的接力：
步骤先有状态，再有流水线
编排层最核心的设计，是把「一步」做成一个带状态的数据对象。三个类一次看清：
45class StepStatus(Enum):46 """Pipeline step status."""4748 PENDING = "pending"49 RUNNING = "running"50 COMPLETED = "completed"51 FAILED = "failed"52 SKIPPED = "skipped"535455@dataclass56class PipelineStep:57 """Pipeline step definition."""5859 name: str60 step_type: str61 config: Dict[str, Any] = field(default_factory=dict)62 dependencies: List[str] = field(default_factory=list)63 handler: Optional[Callable] = None64 status: StepStatus = StepStatus.PENDING65 result: Any = None66 error: Optional[Exception] = None67 delta_mode: bool = False68 base_version_id: Optional[str] = None69 target_version_id: Optional[str] = None707172@dataclass73class Pipeline:74 """Pipeline definition."""7576 name: str77 steps: List[PipelineStep] = field(default_factory=list)78 config: Dict[str, Any] = field(default_factory=dict)79 metadata: Dict[str, Any] = field(default_factory=dict)PipelineStep 有十一个字段，可以分成三组。第一组是身份：name（这一步叫什么）、step_type（它属于哪一类，供重试策略按类别查找）、config（这步自己的配置）。第二组是图结构：dependencies 是一个字符串列表，存的是一步之前必须先跑完的步骤名，这就是有向无环图（DAG）的边。第三组是运行时状态：status 从 PENDING 出发，result 和 error 在跑完后回填，delta_mode 和两个 version id 留给了增量处理，后面会讲。
Pipeline 本身只有四个字段：name、steps 列表、config、metadata。它不包含任何执行逻辑，纯粹是一份「步骤清单加配置」的声明。执行逻辑在 ExecutionEngine 里，这是后面要看到的解耦点。
两个细节值得停一下。config 和 dependencies 的默认值写的是 field(default_factory=dict)、field(default_factory=list)，这个写法避开了 Python 的经典坑：直接写 = {} 会让所有 PipelineStep 实例共享同一个字典，改一个步骤的 config 会污染同类的其他实例。handler 字段是 Optional[Callable]，它是步骤与业务逻辑之间的唯一注入口：框架不知道、也刻意不去知道 handler 内部在做什么，它只负责在正确的时间调用它，并把前一步的输出、本步配置、执行选项一起传进去。这个「只认签名、不认实现」的约定，是后面多源摄取、抽取、建图各章能独立演进的前提。
用 add_step 和 connect_steps 搭 DAG
PipelineBuilder 提供两条方法，把「加一步」和「连一条边」拆开：
119 def add_step(self, step_name: str, step_type: str, **config) -> "PipelineStep":120 """121 Add step to pipeline.122123 Args:124 step_name: Step name/identifier125 step_type: Step type/category126 **config: Step configuration127128 Returns:129 Created PipelineStep object130 """131 delta_mode = config.pop("delta_mode", False)132 base_version_id = config.pop("base_version_id", None)133 target_version_id = config.pop("target_version_id", None)134135 step = PipelineStep(136 name=step_name,137 step_type=step_type,138 config=config,139 dependencies=config.get("dependencies", []),140 handler=config.get("handler"),141 delta_mode = delta_mode,142 base_version_id=base_version_id,143 target_version_id=target_version_id,144 )145146 self.steps.append(step)147 self.logger.debug(f"Added step: {step_name} ({step_type}) | Delta Mode: {delta_mode}")148149 return step输入是 step_name、step_type 和一个 **config。函数先 pop 出 delta_mode、base_version_id、target_version_id 三个键：它们被提升为 PipelineStep 的显式字段，而其余键（比如 handler、dependencies）留在 config 字典里，再经 config.get 取出来。这一步的输入是一堆字符串和可调用对象，输出是一个挂到 self.steps 上的 PipelineStep，同时把对象返回出去，方便链式调用。
pop 的先后顺序在这里是实质性的：三个增量键先从 **config 里取走，剩下的才装进 step.config。如果不先 pop，delta_mode 会连同 handler 一起留在 config 里，等执行到 _execute_step 时被原样展开成 handler 的关键字参数，一个无关的布尔值就这样漏进了业务函数。顺带一提，141 行里 delta_mode = delta_mode 等号两边多了空格，和其余字段的紧凑风格不一致，这是一处肉眼可见的复制粘贴痕迹，不影响语义，但说明这份代码是逐字段手工构造的，没有走 dataclass 的自动装配。
connect_steps 负责补边：
151 def connect_steps(152 self, from_step: str, to_step: str, **options153 ) -> "PipelineBuilder":154 """155 Connect pipeline steps.156157 Args:158 from_step: Source step name159 to_step: Target step name160 **options: Connection options161162 Returns:163 Self for method chaining164 """165 # Find target step and add dependency166 target_step = next((s for s in self.steps if s.name == to_step), None)167 if target_step:168 if from_step not in target_step.dependencies:169 target_step.dependencies.append(from_step)170 else:171 raise ValidationError(f"Target step not found: {to_step}")172173 return self输入是两个步骤名，输出是 self。它做两件事：先在已有步骤里按名字找目标步骤，找不到直接抛 ValidationError（拼错步骤名在建图阶段就暴露）；找到后，把 from_step 追加进目标步骤的 dependencies 列表，且用 not in 去重。注意依赖的方向：connect_steps("ingest", "parse") 表示 parse 依赖 ingest，即 ingest 先跑。边存在下游节点的 dependencies 字段上，这在后面拓扑排序时要反复用到。
这里有个容易读反的方向约定：connect_steps(from_step, to_step) 里的 from_step 是先执行的步骤，to_step 是后执行的步骤，依赖记在 to_step.dependencies 里。为什么不省掉手动连边、让 add_step 按加入顺序自动推导先后？因为步骤的先后不总能从名字或加入顺序推出来，一张 DAG 可以有并行分支，两个步骤之间也可能根本没有先后约束。把「加节点」和「连边」拆成两个 API，换来的是用户对图的完全控制权。
build 时先验证，再产出 Pipeline
add_step 只负责堆积，真正「封箱」发生在 build：
204 try:205 # Validate pipeline structure206 self.progress_tracker.update_tracking(207 tracking_id, message="Validating pipeline structure..."208 )209 validation_result = self.validator.validate_pipeline(self)210 if not validation_result.valid:211 errors = validation_result.errors212 raise ValidationError(f"Pipeline validation failed: {errors}")213214 self.progress_tracker.update_tracking(215 tracking_id, message="Creating pipeline object..."216 )217 pipeline = Pipeline(218 name=name,219 steps=list(self.steps),220 config=self.pipeline_config,221 metadata={222 "step_count": len(self.steps),223 "parallelism": self.pipeline_config.get("parallelism", 1),224 },225 )226227 self.logger.info(f"Built pipeline: {name} with {len(self.steps)} steps")228 self.progress_tracker.stop_tracking(229 tracking_id,230 status="completed",231 message=f"Built pipeline: {name} with {len(self.steps)} steps",232 )233 return pipeline234235 except Exception as e:236 self.progress_tracker.stop_tracking(237 tracking_id, status="failed", message=str(e)238 )239 raisebuild 的输入是流水线名字，输出是一个不可再变的 Pipeline。关键在 209 到 212 行：先调用 self.validator.validate_pipeline(self)，只要 valid 为假，就把 errors 拼进异常消息直接抛掉。这里的 validator 是 PipelineValidator 实例，在 PipelineBuilder.__init__ 里创建，负责检查重复步骤名、缺失的依赖、环。也就是说，一张结构错误的图根本走不到执行阶段。
验证通过后，steps=list(self.steps) 做了一次浅拷贝，把 builder 内部的列表复制进 Pipeline，同时把 step_count 和 parallelism 写进 metadata。此后 builder 再 add_step 也不会影响已经产出的 Pipeline。
执行引擎：状态机与资源边界
构建好了流水线，轮到 ExecutionEngine。先看它管理的两类对象：
51class PipelineStatus(Enum):52 """Pipeline execution status."""5354 PENDING = "pending"55 RUNNING = "running"56 PAUSED = "paused"57 COMPLETED = "completed"58 FAILED = "failed"59 STOPPED = "stopped"606162@dataclass63class ExecutionResult:64 """Pipeline execution result."""6566 success: bool67 output: Any68 metadata: Dict[str, Any] = field(default_factory=dict)69 metrics: Dict[str, Any] = field(default_factory=dict)70 errors: List[str] = field(default_factory=list)PipelineStatus 有六个取值，比 StepStatus 多了 PAUSED 和 STOPPED：暂停和停止是整条流水线级别的控制，单步只需要 pending、running、completed、failed、skipped。ExecutionResult 是执行器对外的统一出口：success 是一个布尔结论，output 是最后一步的输出，metrics 装 steps_executed、steps_failed 之类的计数，errors 是失败信息列表。这个结构让「执行失败」本身也成为一个可返回、可序列化的值，调用方拿到的永远是对象，异常只在最外层兜底。
六种状态之间的实际流转如下：
需要说明：PENDING 在枚举里存在，但引擎的 pipeline_status 字典在置 RUNNING 前没有条目，只有 get_progress 在 execution_engine.py 490 行把缺省值回退到 PENDING。所以图里 PENDING 到 RUNNING 的箭头是声明上的起点，实际代码第一步就把状态写成 RUNNING。
execute_pipeline 是主流程：
177 # Set status178 with self.pipeline_lock:179 self.pipeline_status[pipeline_id] = PipelineStatus.RUNNING180 self.running_pipelines[pipeline_id] = pipeline181182 # Allocate resources183 resources = self.resource_scheduler.allocate_resources(pipeline, **options)184185 try:186 # Execute steps187 result = self._execute_steps(pipeline, data, **options)188189 # Collect metrics190 execution_time = time.time() - start_time191 metrics = {192 "execution_time": execution_time,193 "steps_executed": len(194 [s for s in pipeline.steps if s.status == StepStatus.COMPLETED]195 ),196 "steps_failed": len(197 [s for s in pipeline.steps if s.status == StepStatus.FAILED]198 ),199 }200201 # Update status202 with self.pipeline_lock:203 if metrics["steps_failed"] == 0:204 self.pipeline_status[pipeline_id] = PipelineStatus.COMPLETED205 else:206 self.pipeline_status[pipeline_id] = PipelineStatus.FAILED207208 # Update progress tracking209 self.progress_tracker.stop_tracking(210 pipeline_tracking_id,211 status="completed" if metrics["steps_failed"] == 0 else "failed",212 message=f"Executed {metrics['steps_executed']} steps in {execution_time:.2f}s",213 )214215 # Clear pipeline context when pipeline completes216 self.progress_tracker.clear_pipeline_context(pipeline_id)217218 return ExecutionResult(219 success=metrics["steps_failed"] == 0,220 output=result,221 metadata={222 "pipeline_id": pipeline_id,223 "execution_time": execution_time,224 },225 metrics=metrics,226 )227228 finally:229 # Release resources230 self.resource_scheduler.release_resources(resources)231232 except Exception as e:233 self.progress_tracker.stop_tracking(234 pipeline_tracking_id, status="failed", message=str(e)235 )236 # Clear pipeline context on failure237 self.progress_tracker.clear_pipeline_context(pipeline_id)238 self.logger.error(f"Pipeline execution failed: {e}")239 with self.pipeline_lock:240 self.pipeline_status[pipeline_id] = PipelineStatus.FAILED241242 return ExecutionResult(success=False, output=None, errors=[str(e)])输入是 Pipeline 和一份 data，输出是 ExecutionResult。177 到 180 行把流水线登记进两个字典，都用 pipeline_lock 这把线程锁保护，因为暂停、恢复、停止会从别的线程改这些字典。183 行先分配资源，try/finally 保证 228 到 230 行无论成败都释放资源。
真正的执行在 187 行 _execute_steps，它返回最后一步的输出。之后 191 到 199 行不信任任何计数器，而是直接扫描 pipeline.steps 里每个 step 的 status 字段来数成功与失败的步数：因为 PipelineStep 自己带状态，统计就从「推算」变成了「读回」。203 到 206 行据此把流水线状态置为 COMPLETED 或 FAILED。
最外层的 except 兜住的是那些连 _execute_steps 都没正常返回的情况，比如资源分配阶段就抛了错。它同样返回一个 ExecutionResult(success=False, ...)，把异常信息塞进 errors 列表，函数签名承诺的返回类型始终兑现。
逐步执行：状态流转与重试
_execute_steps 是引擎的循环体：
253 for step_idx, step in enumerate(sorted_steps):254 if self.pipeline_status.get(pipeline.name) == PipelineStatus.STOPPED:255 break256257 # Wait if paused258 while self.pipeline_status.get(pipeline.name) == PipelineStatus.PAUSED:259 time.sleep(0.1)260261 # Track step execution262 step_tracking_id = self.progress_tracker.start_tracking(263 module="pipeline",264 submodule=step.step_type or step.name,265 message=f"Step {step_idx + 1}/{total_steps}: {step.name}",266 )267268 try:269 # Execute step270 step.status = StepStatus.RUNNING271 step_result = self._execute_step(step, current_data, **options)272 step.status = StepStatus.COMPLETED273 step.result = step_result274 current_data = step_result275276 self.progress_tracker.stop_tracking(277 step_tracking_id,278 status="completed",279 message=f"Completed step: {step.name}",280 )281282 except Exception as e:283 step.status = StepStatus.FAILED284 step.error = e285286 # Retry loop respecting max_retries from the policy287 retry_policy = self.failure_handler.get_retry_policy(step.step_type)288 max_retries = retry_policy.max_retries if retry_policy else 0289 retry_count = 0290 success = False291292 while retry_count < max_retries:293 recovery_result = self.failure_handler.handle_step_failure(step, e)294 if not recovery_result.get("retry", False):295 break296 retry_delay = recovery_result.get("retry_delay", 0.0)297 if retry_delay > 0:298 time.sleep(retry_delay)299 self.progress_tracker.update_tracking(300 step_tracking_id,301 status="running",302 message=f"Retrying step: {step.name} (attempt {retry_count + 1})",303 )304 step.status = StepStatus.RUNNING305 try:306 step_result = self._execute_step(step, current_data, **options)307 step.status = StepStatus.COMPLETED308 step.result = step_result309 current_data = step_result310 success = True311 break312 except Exception as retry_e:313 step.status = StepStatus.FAILED314 step.error = retry_e315 e = retry_e316 retry_count += 1317318 if success:319 self.progress_tracker.stop_tracking(320 step_tracking_id,321 status="completed",322 message=f"Retry successful: {step.name}",323 )324 else:325 self.progress_tracker.stop_tracking(326 step_tracking_id, status="failed", message=str(e)327 )328 raise e329330 return current_data输入是排好序的步骤列表和初始数据，输出是最后一步的结果，也就是整条流水线的 output。每个循环开始先查两条控制信号：253 行看是否被 STOPPED，是就 break 退出整个循环；257 到 259 行看是否被 PAUSED，是就以 0.1 秒为周期自旋等待。暂停和停止在代码里都是「每步之间检查一次」，所以它们只在步与步的边界生效，正在执行的那一步不会被中途打断。
正常路径是 268 到 274 行：置 RUNNING，调用 _execute_step，置 COMPLETED，把结果同时写进 step.result 和局部变量 current_data。current_data 是步骤之间的数据通道：上一步的输出就是下一步的输入，这是一个纯线性的数据流。
异常路径是 282 行往后。先置 FAILED、记下 step.error，然后按 step.step_type 从 failure_handler 取重试策略。max_retries 为 0 时跳过整个重试循环，直接落到 328 行 raise e，把错误向上抛。重试循环里，每次先问 handle_step_failure 返回的 retry 是否为真，再按 retry_delay 睡一觉，重新置 RUNNING 再跑一次。重试成功就 break，失败就把错误换成最新的 retry_e 并计数。重试次数耗尽后，324 到 328 行停止跟踪并 raise e，这个异常最终被 execute_pipeline 最外层的 except 接住，转成 ExecutionResult(success=False, errors=[...])。
拓扑排序：顺序从哪里来
循环依赖是编排层最容易翻车的地方，_execute_steps 开跑前先做一次 Kahn 拓扑排序：
400 step_map = {step.name: step for step in steps}401 in_degree = {step.name: len(step.dependencies) for step in steps}402403 # Find steps with no dependencies404 self.progress_tracker.update_tracking(405 tracking_id, message="Finding entry points..."406 )407 queue = [step for step in steps if in_degree[step.name] == 0]408 sorted_steps = []409410 # Perform topological sort411 self.progress_tracker.update_tracking(412 tracking_id, message="Performing topological sort..."413 )414 while queue:415 step = queue.pop(0)416 sorted_steps.append(step)417418 # Update in-degrees of dependent steps419 for other_step in steps:420 if step.name in other_step.dependencies:421 in_degree[other_step.name] -= 1422 if in_degree[other_step.name] == 0:423 queue.append(other_step)424425 # Check for cycles426 self.progress_tracker.update_tracking(427 tracking_id, message="Checking for circular dependencies..."428 )429 if len(sorted_steps) != len(steps):430 raise ValidationError("Circular dependency detected in pipeline")输入是步骤列表，输出是排好序的列表。in_degree 是每个步骤的入度，等于它的 dependencies 长度。入度为 0 的步骤先进队列，随后每弹出一个步骤，就把所有依赖它的步骤入度减一，减到 0 再入队。这一步依赖「依赖边存在下游节点的 dependencies 字段」这个约定。
循环检测藏在 429 到 430 行：如果排序结果的长度不等于原列表长度，说明有环，直接抛 ValidationError。这是环的第二道防线，第一道在 build 阶段的 PipelineValidator 里，两道防线的判断逻辑互相独立。
增量模式：拦截在 handler 之前
_execute_step 是真正调用 handler 的地方，但 handler 前还有一段拦截：
341 if getattr(step, "delta_mode", False):342 self.logger.info(f"Executing step '{step.name}' in incremental delta mode.")343 344 version_manager = options.get("version_manager") or self.config.get("version_manager")345 triplet_store = options.get("triplet_store") or self.config.get("triplet_store")346 347 if not version_manager or not triplet_store:348 raise ProcessingError(349 f"Step '{step.name}' requires 'version_manager' and 'triplet_store' "350 f"in execution options for delta processing."351 )352 353 if not step.base_version_id or not step.target_version_id:354 raise ValidationError(355 f"Step '{step.name}' in delta_mode requires 'base_version_id' "356 f"and 'target_version_id' to be set."357 )358 359360 base_snap = version_manager.get_version(step.base_version_id)361 target_snap = version_manager.get_version(step.target_version_id)362 363 if not base_snap:364 raise ValidationError(f"Base version '{step.base_version_id}' not found in storage.")365 if not target_snap:366 raise ValidationError(f"Target version '{step.target_version_id}' not found in storage.")367 368 base_uri = base_snap.get("graph_uri")369 target_uri = target_snap.get("graph_uri")370 371 if not base_uri or not target_uri:372 raise ValidationError(373 "Both base and target snapshots must contain a 'graph_uri' "374 "to compute native store deltas."375 )376 377 self.logger.debug(f"Computing delta between {base_uri} and {target_uri}")378 delta_result = triplet_store.compute_delta(base_uri, target_uri, **options)379 380 data = delta_result381 382 if step.handler:383 return step.handler(data, **step.config, **options)384 else:385 return data方法 docstring 里写明了这段拦截的目的：「If delta_mode is enabled for the step, this intercepts the execution to compute the delta between the base and target versions, passing only the changes (added/removed triples) to the handler.」。翻译过来：开了 delta_mode 的步骤，执行器会先算出 base 与 target 两个版本之间的差集，只把新增和删除的三元组交给 handler。version_manager 和 triplet_store 从 options 或引擎配置里取，缺一个就抛 ProcessingError；两个 version id 没设就抛 ValidationError。最终 382 到 385 行是兜底：有 handler 就调用它并把 data、步骤配置、执行选项一起传进去，没有 handler 就把 data 原样返回，这一步退化为透传。
序列化：把图存下来，但存不下函数
PipelineSerializer 把一张图变成 JSON 或字典：
388 def serialize_pipeline(389 self, pipeline: Pipeline, format: str = "json", **options390 ) -> Union[str, Dict[str, Any]]:391 """392 Serialize pipeline to specified format.393394 Args:395 pipeline: Pipeline object396 format: Serialization format397 **options: Additional options398399 Returns:400 Serialized pipeline401 """402 pipeline_data = {403 "name": pipeline.name,404 "steps": [405 {406 "name": step.name,407 "type": step.step_type,408 "config": step.config,409 "dependencies": step.dependencies,410 "delta_mode": getattr(step, "delta_mode", False),411 "base_version_id": getattr(step, "base_version_id", None),412 "target_version_id": getattr(step, "target_version_id", None),413 }414 for step in pipeline.steps415 ],416 "config": pipeline.config,417 "metadata": pipeline.metadata,418 }419420 if format == "json":421 import json422423 return json.dumps(pipeline_data, indent=2, default=str)424 else:425 return pipeline_data输入是 Pipeline，输出是 JSON 字符串或字典。它把 name、每个 step 的 name/type/config/dependencies、三个增量字段、config、metadata 全部展开成纯数据。注意它没有序列化 handler：Python 函数对象不能 json.dumps。官方文档也点明了这一点，序列化后需要把 handler 重新注册回恢复出来的步骤上才能执行。
总入口：编排器如何把散件拼起来
回到最外层。Semantica 类是本框架的总入口，它自己几乎不干活，负责把下面各模块的 lazy property 串起来。build_knowledge_base 的源循环是编排的核心：
359 for idx, source in enumerate(validated_sources, 1):360 try:361 # Track file processing362 file_str = str(source)363 file_tracking_id = self.progress_tracker.start_tracking(364 file=file_str,365 module="core",366 submodule="build_knowledge_base",367 message=f"Processing {Path(file_str).name if file_str else 'source'}",368 pipeline_id=pipeline_id,369 )370 try:371 result = self.run_pipeline(pipeline, source)372 results.append(result)373 self.progress_tracker.stop_tracking(374 file_tracking_id, status="completed"375 )376 except Exception as e:377 self.progress_tracker.stop_tracking(378 file_tracking_id, status="failed", message=str(e)379 )380 self.logger.error(f"Failed to process source {source}: {e}")381 if kwargs.get("fail_fast", False):382 raise ProcessingError(383 f"Failed to process source {source}: {e}"384 )⋯ # ... 空行省略 ...386 # Update overall progress387 if idx % update_interval == 0 or idx == total_sources:388 self.progress_tracker.update_progress(389 overall_tracking_id,390 processed=idx,391 total=total_sources,392 message=f"Processing sources... {idx}/{total_sources}"393 )394 except Exception as e:395 self.logger.error(f"Failed to process source {source}: {e}")396 if kwargs.get("fail_fast", False):397 raise ProcessingError(f"Failed to process source {source}: {e}")输入是验证过的源列表，输出是 results 列表。外层 for 逐个源处理，每份源先用 progress_tracker.start_tracking 开一条跟踪记录，再调 run_pipeline。关键在 376 到 384 行：单份源失败时，先记一条 failed 跟踪，再记日志，只有当调用方传了 fail_fast=True 才向上抛。默认情况下，一份坏文件只是让 results 里少一条成功记录，循环继续。
这个循环的语义与场景还原里的凌晨两点事故直接对应：如果那套手工脚本换成这里的 fail_fast 默认关闭，第 42351 份 PDF 只会被标记失败，其余文件照常处理，最后 statistics 里的 success_rate 会诚实反映丢了多少。
入口之前的守卫：源校验
进循环之前，_validate_sources 先把不能处理的源过滤掉：
711 validated_sources = []712713 for source in sources:714 # Convert string to Path for easier handling715 source_path = Path(source) if isinstance(source, str) else source716717 # Check if source is valid:718 # 1. File path exists on filesystem719 # 2. URL starts with http:// or https://720 is_valid_file = source_path.exists()721 is_valid_url = isinstance(source, str) and source.startswith(722 ("http://", "https://")723 )724725 if is_valid_file or is_valid_url:726 validated_sources.append(source)727 else:728 self.logger.warning(729 f"Source not found or invalid: {source}. " "Skipping this source."730 )731732 # Ensure we have at least one valid source733 if not validated_sources:734 error_msg = (735 f"No valid sources provided. "736 f"Checked {len(sources)} source(s), all were invalid."737 )738 raise ProcessingError(error_msg)739740 return validated_sources输入是原始源列表，输出是过滤后的列表。有效性的判据只有两条：文件路径真实存在，或者字符串以 http://、https:// 开头。无效的源记一条 warning 就跳过。但 733 到 738 行设了底线：如果过滤完一个都不剩，抛 ProcessingError，拒绝空跑。这是「宽容单点、严拒全无」的边界策略。
流水线从哪来：_create_pipeline 的降级路径
build_knowledge_base 里的 pipeline 由 _create_pipeline 造出：
742 def _create_pipeline(self, pipeline_config: Dict[str, Any]) -> Any:743 """744 Create processing pipeline from configuration.745746 This method creates a pipeline instance from the provided configuration.747 The actual pipeline creation is delegated to the pipeline module.748749 Args:750 pipeline_config: Pipeline configuration dictionary751752 Returns:753 Pipeline object or configuration dict (if pipeline module not available)754 """755 try:756 # Use the lazy property757 builder = self.pipeline_builder758759 if not pipeline_config:760 builder.add_step("default_step", "default")761 return builder.build("default_pipeline")762763 steps_config = pipeline_config.get("steps")764765 if isinstance(steps_config, list) and steps_config and isinstance(766 steps_config[0], str767 ):768 converted_steps = [769 {"name": name, "type": name, "config": {}}770 for name in steps_config771 ]772 normalized_config: Dict[str, Any] = {773 "name": pipeline_config.get("name", "default_pipeline"),774 "steps": converted_steps,775 }776 if "parallelism" in pipeline_config:777 normalized_config["parallelism"] = pipeline_config["parallelism"]778 return builder.build_pipeline(normalized_config)779780 if "steps" in pipeline_config:781 return builder.build_pipeline(pipeline_config)782783 builder.add_step("default_step", "default")784 return builder.build("default_pipeline")785 except (ImportError, OSError, ProcessingError):786 self.logger.debug("Pipeline module not available, using config directly")787 return pipeline_config输入是流水线配置字典，输出是 Pipeline 或配置字典。它接受三种形状：空配置、steps 是字符串列表（如 ["ingest", "parse"]）、steps 是完整字典列表。前两种都会归一化成 build_pipeline 能吃的结构。最外层 except 捕获 ImportError、OSError、ProcessingError：pipeline 模块不可用时，它不崩，而是把配置字典原样返回，让下游的 run_pipeline 走「对象没有 execute 方法」的降级分支。
整条调用链画成时序图：
设计决策分析
先给出结论：编排器解耦、状态挂在步骤上、执行器只管状态机、序列化只存数据不存函数，这四个决策合起来，才让「可解释、可审计」成为可能。
第一，Semantica 不直接 import 各模块，而是通过 embedding_generator、pipeline_builder、graph_builder 这类 lazy property 按需导入。orchestrator.py 的模块 docstring 里写明，这个类协调所有组件并管理整体执行流。这种解耦的直接收益是「部分模块装不上，框架还能站起来」：_initialize_modules 的注释说 import 失败只记日志不停止初始化，_create_pipeline 的 except 把 ImportError 降级成返回配置字典。代价是错误可能被推迟到真正调用模块时才暴露。
第二，把 status/result/error 放进 PipelineStep，执行器就不需要维护一份并行的状态表。execute_pipeline 数成功失败时直接扫 pipeline.steps 的字段，get_progress 也是读步骤状态算百分比。状态跟着对象走，审计时拿到的 Pipeline 本身就带着每一步的结局。
第三，执行器对外返回 ExecutionResult 而不抛出「正常失败」。文档 docs/reference/pipeline.md 给出了这份设计要解决的直接问题：
28## Why Use a Pipeline?2930You could wire Semantica modules together with plain Python code. Pipelines add:3132- **Retry and failure handling** — A single bad document doesn't crash a 10,000-document run.33- **Parallelism** — Run extraction across multiple workers with one parameter.34- **Progress tracking** — tqdm console bar or WebSocket streaming to Explorer.35- **Reproducibility** — Save the exact pipeline configuration to YAML and replay on any machine.36- **Delta mode** — On re-runs, only process documents that changed since the last run.37- **Validation** — Catch misconfigured steps and dependency cycles before they fail mid-run.第一条「Retry and failure handling」就是场景还原里那个问题的官方回答：单份坏文档不该拖垮一万份的整批运行。这里的措辞用了 doesn't crash，说明设计目标明确是「失败被装进结果里，运行不中断」。
第四，序列化只存数据不存函数。docs/reference/pipeline.md 在第 296 行点明原因：序列化的流水线只捕获名字、类型、配置，不捕获 handler 函数，因为 callable 无法序列化。这意味着「可复现」有个前提：函数要能靠 step_type 重新找到并注册。这既是限制，也是它和 GraphRAG 的差异起点，下面横向对比会展开。
第五个决策藏得更深：环的检测做了两遍。build 阶段由 PipelineValidator.check_dependencies 用 DFS 找环，执行阶段由 _topological_sort 用 Kahn 入度再查一次。两遍检测的算法不同，防的是「有人绕过 builder 直接手搓 Pipeline」。验证的时机因此落在 build：构建期图还在变，边还没连完，只有封箱那一刻才有完整结构可验。配置统一用 dict 这种宽松结构，为 lazy property 的按需初始化留了口子：各模块的配置切片可以用 self.config.get("pipeline", {}) 这种取值直接下传，缺省的模块不会被强制构造。
边界条件剖析
如果 pipeline 模块导入失败，build_knowledge_base 会怎样？
_create_pipeline 的 except (ImportError, OSError, ProcessingError) 在 785 到 787 行返回原始配置字典。随后 build_knowledge_base 在 338 行用 hasattr(pipeline, 'steps') 判断：字典没有 steps 属性，跳过进度注册。接着 run_pipeline 在 orchestrator.py 454 行起判断 isinstance(pipeline, Pipeline) 为假，走 pipeline.execute(data) 分支；如果连 execute 方法都没有，在 orchestrator.py 498 到 500 行抛 ProcessingError("Pipeline must be a Pipeline object or have execute() method")。所以降级不是无限降，最后一层守卫是「必须有 execute 方法」。
如果某个步骤没有 handler，会怎样？
_execute_step 的 382 到 385 行：if step.handler 为假时 return data，这一步变成透传。PipelineValidator.validate_step 在 pipeline_validator.py 252 行只会给出一条 warning「Step X has no handler」，不阻止构建。于是「只有名字没有动作的步骤」能合法存在于图中，把输入原样传给下一步。这在本章源码里是有意保留的柔性，_create_pipeline 在空配置时造的 default_step 正是这种透传步骤。
如果所有源都无效，会怎样？
_validate_sources 在 733 到 738 行抛 ProcessingError，消息明确写出「No valid sources provided ... all were invalid」。这个异常从 build_knowledge_base 的最外层 except（orchestrator.py 444 行起）接住，重新包装成 ProcessingError("Failed to build knowledge base: ...") 向上抛。结果：一次也不跑，不会产出空图再假装成功。
如果两个步骤互相依赖，会怎样？
build 阶段 PipelineValidator.check_dependencies 在 pipeline_validator.py 334 行报「Circular dependency detected」，build 在 210 到 212 行把它转成 ValidationError 抛出。即使有人绕过 build 直接构造了带环的 Pipeline 丢给执行器，_topological_sort 在 429 到 430 行还会再抛一次同样的异常。两道防线，第一道在构建期，第二道在执行期。
如果 delta 步骤没传 version_manager 或 triplet_store，会怎样？
_execute_step 在 344 到 345 行从 options 或引擎配置里取这两个对象，347 到 351 行只要有一个是空值就抛 ProcessingError，消息写明 delta 处理需要这两个执行期资源。如果 version id 没设，353 到 357 行抛 ValidationError。也就是说，delta 模式把「配置错误」和「资源缺失」分成了两类异常：缺 version id 是构建问题，缺 version_manager 是执行问题。这两类在 build_knowledge_base 里都不会被静默吞掉，最终以 ExecutionResult(success=False, errors=[...]) 的形式回到调用方。
横向对比
对比对象是 GraphRAG 的 indexing 入口。两边回答的是同一个问题：怎么把「数据到图」的整条流程串起来。Semantica 的答案是显式 Pipeline 对象，GraphRAG 的答案是配置驱动的工作流列表。
GraphRAG 的 Pipeline 是一个极薄的容器：
11class Pipeline:12 """Encapsulates running workflows."""1314 def __init__(self, workflows: list[Workflow]):15 self.workflows = workflows1617 def run(self) -> Generator[Workflow]:18 """Return a Generator over the pipeline workflows."""19 yield from self.workflows2021 def names(self) -> list[str]:22 """Return the names of the workflows in the pipeline."""23 return [name for name, _ in self.workflows]2425 def remove(self, name: str) -> None:26 """Remove a workflow from the pipeline by name."""27 self.workflows = [w for w in self.workflows if w[0] != name]它没有步骤状态、没有依赖字段、没有验证，只有一个 (name, function) 二元组列表和一个 run 生成器。顺序就是列表顺序，没有拓扑排序可言。remove 方法是它唯一的「编排」操作，用于在运行前把某个工作流摘掉。
Pipeline 从哪来？PipelineFactory.create_pipeline 从配置或预设列表拼出来：
39 @classmethod40 def create_pipeline(41 cls,42 config: GraphRagConfig,43 method: IndexingMethod | str = IndexingMethod.Standard,44 ) -> Pipeline:45 """Create a pipeline generator."""46 workflows = config.workflows or cls.pipelines.get(method, [])47 logger.info("Creating pipeline with workflows: %s", workflows)48 return Pipeline([(name, cls.workflows[name]) for name in workflows])输入是 config 和一个 method，输出是 Pipeline。config.workflows 优先，否则按 method 查预设。预设列表在文件底部注册：
84PipelineFactory.register_pipeline(85 IndexingMethod.Standard, ["load_input_documents", *_standard_workflows]86)87PipelineFactory.register_pipeline(88 IndexingMethod.Fast, ["load_input_documents", *_fast_workflows]89)90PipelineFactory.register_pipeline(91 IndexingMethod.StandardUpdate,92 ["load_update_documents", *_standard_workflows, *_update_workflows],93)94PipelineFactory.register_pipeline(95 IndexingMethod.FastUpdate,96 ["load_update_documents", *_fast_workflows, *_update_workflows],97)四条预设流水线都是「名字列表」。执行时 GraphRAG 只是逐个 await：
131 for name, workflow_function in pipeline.run():132 last_workflow = name133 context.callbacks.workflow_start(name, None)134135 with WorkflowProfiler() as profiler:136 result = await workflow_function(config, context)137138 context.callbacks.workflow_end(name, result)139 yield PipelineRunResult(140 workflow=name, result=result.result, state=context.state, error=None141 )142 context.stats.workflows[name] = profiler.metrics143 await _dump_stats_json(context)144 if result.stop:145 logger.info("Halting pipeline at workflow request")146 break147148 context.stats.total_runtime = time.time() - start_time149 logger.info("Indexing pipeline complete.")150 await _dump_stats_json(context)151 await _dump_context_json(context)152153 except Exception as e:154 logger.exception("error running workflow %s", last_workflow)155 yield PipelineRunResult(156 workflow=last_workflow, result=None, state=context.state, error=e157 )循环里没有 step.status、没有 retry_count、没有 dependencies。状态被放进了外部的 context.state 字典和存储里的 context.json、stats.json，用 _dump_stats_json 落盘，步骤对象本身不携带状态。异常处理也只有一个外层 except：任何工作流抛错，记录 last_workflow 后 yield 一个带 error 的 PipelineRunResult，没有按工作流的重试。
GraphRAG 为什么可以没有步骤状态、依赖和重试？因为它的输入是单语料文档集合，标准流程是固定的九步，顺序由预设列表写死，不存在「三十种源、每步可替换」的爆炸组合。它用 config.workflows 的名字列表加 PipelineFactory 的名字到函数映射，补上了「可配置」的需求；用存储里的 context.json 补上了「可恢复」的需求。检索证据是 graphrag/index/workflows/factory.py 的 register、register_pipeline 和 run_pipeline.py 的 state = json.loads(state_json) if state_json else {}（run_pipeline.py 47 到 49 行）。
对比的落点：Semantica 把「步骤顺序、状态、失败处理」做成显式的、可序列化的、可单步检查的对象；GraphRAG 把它们压成「配置里的一个列表」加「存储里的一个状态字典」。前者为多源、可插拔、可审计服务，后者为单语料、固定流程、快速上手服务。两者没有谁更对，只有各自要解决的问题不同。
把两边放到审计场景里再压一次。Semantica 想回答「这条边是哪个源、哪一步、什么时候产出的」，所以它必须让每一步在对象上留下 status/result/error，还要把整张图序列化成可回放的配置。GraphRAG 想回答「这批文档的索引跑完了没有、哪些工作流花了多久」，所以它把状态塞进存储里的 stats.json 和 context.json，用 WorkflowProfiler 记录每个工作流的耗时。一个是对象内状态，一个是存储内状态：前者随图走、可被单步检查；后者随运行走、可被跨进程恢复。两条路都指向同一个目标，即让一次处理从黑盒变成可查询的记录。
互动演示设计
形态：模拟器。
一句话结论：把每一步做成「带状态的小工位」，再把工位排成一条只许单向流动的传送带，顺序、失败、暂停就都有了可观察的落点。
舞台元素与比喻：一条传送带上有四个工位，分别是 ingest、parse、extract、build_kg。每个工位门口挂一块状态牌（PENDING / RUNNING / COMPLETED / FAILED）。传送带入口有个质检员（_validate_sources），出口有个会计（ExecutionResult）。工位之间用绳子相连，绳子方向就是 dependencies。
分步动画：
- 质检员检查每个包裹，坏的记 warning 丢一边，全坏就罢工。字幕：「宽容单点，严拒全无，全无时抛 ProcessingError」。
- 一张图纸（
Pipeline）被画出来，质检员先验图：有没有重名工位、有没有缺绳子、有没有绕成圈。字幕：「build 先验证，环在 210 行就抛 ValidationError」。 - 包裹进入传送带，第一个工位亮 RUNNING，做完亮 COMPLETED，结果递给下一个工位。字幕：「current_data 是步骤之间的数据通道」。
- 第三个工位突然 FAILED。字幕：「先记 step.error，再按 step_type 取重试策略」。
- 重试小车倒回一段距离，工位重新亮 RUNNING，再试一次。字幕：「retry_count 到 max_retries 才 raise e」。
- 出口会计数完成工位数，打印 ExecutionResult。字幕：「失败也返回对象，调用方拿到的永远是 ExecutionResult」。
读者可操作项：把第三个工位的 handler 换成 lambda data, **kw: 1/0，跑一遍 execute_pipeline，观察 result.success 为 False 且 result.errors 非空；再给这个 step_type 配一个 RetryPolicy(max_retries=2)，观察工位重试两次后仍失败。
逻辑轨迹面板伪代码（右侧为真实行号）：
build_knowledge_base(sources) # orchestrator.py:281
_validate_sources(sources) # orchestrator.py:693
_create_pipeline(config) # orchestrator.py:742
for source in validated_sources: # orchestrator.py:359
run_pipeline(pipeline, source) # orchestrator.py:454
ExecutionEngine.execute_pipeline # execution_engine.py:113
_topological_sort(steps) # execution_engine.py:387
_execute_steps # execution_engine.py:244
_execute_step # execution_engine.py:332可迁移结论
值得抄的第一条：给步骤挂状态字段。PipelineStep 的 status/result/error 让执行器免于维护并行状态表，让统计直接从字段读回、不再靠推算，让审计能拿到带结局的图。最小成本形态是三个字段加一个枚举，几十行内能抄走。
值得抄的第二条：执行器对外返回结果对象，把正常失败也装进同一个对象。ExecutionResult 把 success、output、metrics、errors 打包，失败成了可序列化的值。这一条不依赖 Python：任何工作流系统都能用「返回带 errors 的结果」替代「异常中断」，让批处理在单点失败时继续。
值得抄的第三条：资源用 try/finally 包住，状态登记用锁保护。execute_pipeline 在 228 到 230 行释放资源，177 到 180 行在锁内改状态，这两处是并发安全的最小样板。
哪些是过度设计：ResourceScheduler 在本章锚点里基本是空壳，Semantica._allocate_resources 在 orchestrator.py 938 到 940 行直接返回一个占位字典，注释自己写明「Placeholder for resource allocation logic」。ParallelismManager 有线程池、进程池两套执行器，但 ExecutionEngine._execute_steps 仍是顺序 for 循环，并行要显式调用 ParallelismManager.execute_parallel 才生效。这两处说明：框架把「未来可能用」的部件先摆出来了，抄的时候可以只留接口、不抄实现。
最小成本形态：一个 (name, handler, deps) 列表、一个 Kahn 拓扑排序函数、一个「置 RUNNING → 调 handler → 置 COMPLETED / FAILED」的循环，加上「失败记 errors、不中断」的兜底，就是本章编排层能落地的全部。
把「值得抄」和「过度设计」的分界线再划清楚一点：状态字段、结果对象、拓扑排序、try/finally 释放，这四样是编排层的骨架，缺一样都会在规模上来时翻车。ResourceScheduler、ParallelismManager、FailureHandler 的 severity 分类，这三样是可选的增强，单机、几十个源、线性流程的团队可以先不碰。判断标准只有一个：你的流程里有没有「多步依赖」和「单点失败要隔离」这两件事，有就先抄骨架，没有就继续用一段脚本加 try/except。
思考题
- 动手验证：在你自己的临时目录新建
cycle_demo.py（不要写进 semantica 仓库），写入下面内容，运行python3 cycle_demo.py，观察抛出的ValidationError里errors列表包含什么。
from semantica.pipeline import PipelineBuilder
b = PipelineBuilder()
b.add_step("a", "demo")
b.add_step("b", "demo")
b.connect_steps("a", "b")
b.connect_steps("b", "a")
b.build("cycle")再删掉一行 connect_steps，重跑，观察 build 是否成功。这个现象说明环在构建期就被哪一行拦下？
-
_execute_steps里，STOPPED用break退出，PAUSED用while自旋等待。为什么停止能直接跳出、暂停却只能自旋？如果你要支持「暂停后立刻打断正在执行的一步」，现在的实现差在哪（提示：看_execute_step里 handler 调用有没有被检查）？ -
serialize_pipeline不存handler。如果你拿到一份序列化后的 JSON，要让它重新可执行，最小需要补什么步骤？GraphRAG 的Pipeline存的是(name, function)二元组，它序列化时会不会遇到同样的函数问题？为什么它的名字到函数映射（PipelineFactory.workflows）能作为序列化的替代？ -
GraphRAG 的
_run_pipeline只有一个外层except，任何工作流失败都会让后续工作流停下来（yield 一个带 error 的结果后不再执行）。Semantica 的单步失败在默认fail_fast=False时会让整条流水线继续。这两种语义分别适合什么规模、什么输入类型的任务？如果你给 GraphRAG 加「单工作流重试」，需要改run_pipeline.py的哪几行？