第 2 章
多源摄取
场景还原
一家中型公司要做知识图谱，原料散在五个地方：HR 给的 CSV、S3 桶里几百份 PDF、一个 Git 仓库、一台 Postgres 的工单库、一个公开的汇率 API。第一版代码是实习生写的，一个 read_source(path) 函数塞了十几层 if/else 判断扩展名，谁加新源就往里再叠一层。
两个月后出了两件事。第一件，营销同事要求「顺手抓一下竞品网站」，实习生直接 requests.get(url)，测试环境抓的是公网，一切正常；上到云主机那天，某个 URL 指向 http://169.254.169.254/latest/meta-data/，云厂商的元数据服务直接把实例的临时凭证吐了出来。第二件，财务要接 Snowflake，于是那个函数又被塞进一段连接串判断，从此改一行要担心五处别的源跟着炸。
这一章读的就是 Semantica 怎么回答这个核心问题：几十种数据源收进一个包，怎么做到加一个新源不动旧源、怎么让 web 摄取不敢乱连内网、又怎么让每个结论指回来源。先交代一个实读结论：Semantica 的 ingest 层没有把 22 个源压成同一种对象，它统一的是入口和分发，不是数据形状。
逐行精读
整条链路的形态先看一张图：
四个源进去，同一个入口判定，同一个注册表分发，出来的对象形状各不相同。
数据对象各归各，统一发生在分发层
先看文件源产出的对象。它是一个纯 dataclass，没有基类、没有接口约束：
43@dataclass44class FileObject:45 """File object representation."""4647 path: str48 name: str49 size: int50 file_type: str51 mime_type: Optional[str] = None52 content: Optional[bytes] = None53 metadata: Dict[str, Any] = field(default_factory=dict)54 ingested_at: datetime = field(default_factory=datetime.now)5556 @property57 def text(self) -> str:58 """59 Get the file content as text.6061 Returns:62 str: Decoded file content or empty string if no content63 """64 if self.content is None:65 return ""6667 if isinstance(self.content, str):68 return self.content6970 try:71 # Try to decode as UTF-872 return self.content.decode("utf-8")73 except UnicodeDecodeError:74 try:75 # Fallback to latin-176 return self.content.decode("latin-1")77 except Exception:78 return ""输入是一个路径加一组选项，输出是 FileObject。content 是原始字节，text 属性按 UTF-8 解码，失败再退到 latin-1，两次都失败返回空串。read_content=False 时 content 保持 None，text 因此返回空串，对象退化成纯元数据载体，适合「扫目录只看有哪些文件」这类场景。metadata 字典由 ingest_file 填充，扩展名、父目录、is_supported 标志都在里面，调用方透传的 **options 也会原样混进去。这里已经埋了第一条设计线索：文件摄取不负责把 PDF 变成结构化文本，它只负责「把字节拿进来 + 判断是什么类型 + 记下从哪来」。解析是下一章的事。
再往里一层是类型检测。FileTypeDetector 用三种手段按顺序判定扩展名、MIME、魔数：
81class FileTypeDetector:82 """83 File type detection and validation.8485 This class identifies file types using multiple detection methods:86 1. File extension analysis87 2. MIME type detection88 3. Magic number (file signature) analysis8990 It supports a wide range of document, image, audio, and video formats.91 """9293 def __init__(self):94 """95 Initialize file type detector.9697 Sets up the detector with all supported file formats and initializes98 the MIME types database for accurate type detection.99 """100 self.logger = get_logger("file_type_detector")101102 # Combine all supported formats into a single list103 self.supported_formats = (104 SUPPORTED_DOCUMENT_FORMATS105 + SUPPORTED_IMAGE_FORMATS106 + SUPPORTED_AUDIO_FORMATS107 + SUPPORTED_VIDEO_FORMATS108 )109110 # Initialize Python's MIME types database111 mimetypes.init()三类格式列表从 semantica/utils/constants.py 拼进来，检测顺序是「扩展名 → MIME → 魔数」，扩展名最快所以排最前。魔数排在最后，原因是它最贵：detect_type 先看扩展名，文件存在但扩展名不可靠才求助 MIME，两种都拿不到、且调用方传了 content 字节时，才去比对文件头签名字典，PDF 头、ZIP/Office 头、Parquet 的 PAR1 都在那张表里。这一层的职责边界同样克制：它只回答「这是什么类型」，不回答「里面写了什么」。
FileIngestor 本体把上面这些拼起来。看它的类头和文档串：
396class FileIngestor:397 """398 File system and cloud storage ingestion handler.399400 This class provides comprehensive file ingestion capabilities from:401 - Local filesystem (files and directories)402 - Cloud storage providers (AWS S3, Google Cloud Storage, Azure Blob)403404 Features:405 - Automatic file type detection406 - Recursive directory scanning407 - File size validation408 - Progress tracking callbacks409 - Batch processing with error handling410411 Example Usage:412 >>> ingestor = FileIngestor()413 >>> files = ingestor.ingest_directory("./documents", recursive=True)414 >>> single_file = ingestor.ingest_file("report.pdf", read_content=True)415 """一个 ingestor 类对应一个源类型，类的内部再按「文件 / 目录 / 云」拆成三个方法，云客户端 CloudStorageIngestor 又单独成类，惰性初始化。ingest() 这个名字在 FileIngestor 里还是目录与单文件的统一别名，输入是目录就转 ingest_directory，是文件就转 ingest_file，都不是才抛错：
460 path = Path(source)461 if path.is_dir():462 return self.ingest_directory(path, **options)463 elif path.is_file():464 return [self.ingest_file(path, **options)]465 else:466 raise ValidationError(f"Path not found: {path}")注意单文件分支包成列表 [FileObject]，目录分支本身也返回列表，所以两条路径的返回形状统一成 List[FileObject]。判断依据只有 Path.is_dir() 和 Path.is_file() 两个真值，没读文件内容。这就是「每源一个文件」的组织形态，本章第三节会展开它换来什么。
可选依赖的懒加载，隔离的第一道闸
semantica/ingest/__init__.py 里最值得读的不是 __all__ 列表，是模块级 __getattr__：
264def __getattr__(name: str) -> Any:265 """Load optional ingestion backends only when callers request them."""266 if name not in _LAZY_EXPORTS:267 raise AttributeError(f"module {__name__!r} has no attribute {name!r}")268269 module_name, attr_name = _LAZY_EXPORTS[name]270 try:271 module = importlib.import_module(module_name, __name__)272 except ModuleNotFoundError as exc:273 message = _OPTIONAL_DEPENDENCY_MESSAGES.get(module_name)274 missing_name = getattr(exc, "name", None)275 if message and missing_name in {"git", "bs4", "pyarrow"}:276 raise ImportError(message) from exc277 raise278279 value = getattr(module, attr_name)280 globals()[name] = value281 return valueWebIngestor、RepoIngestor、ParquetIngestor 这类带重依赖的类都在 _LAZY_EXPORTS 字典里登记成 名字 → (模块, 属性)。只有当你真的写 from semantica.ingest import WebIngestor，Python 才会触发 __getattr__ 去 import web_ingestor 模块。缺依赖时它认 git、bs4、pyarrow 三个名字，把裸的 ModuleNotFoundError 换成带安装提示的 ImportError。效果是：一个只装文件摄取的用户 import 顶层包时，boto3、beautifulsoup4、GitPython 一个都不会被拉进来。这就是「互相污染」的第一层防线，靠的是模块隔离加惰性导入，外加可选依赖报错信息集中登记。
MethodRegistry：一张 task 打头的两层字典
注册机制在 semantica/ingest/registry.py，核心就一个类，全文不含注释约四十行：
52class MethodRegistry:53 """Registry for custom ingestion methods."""5455 _methods: Dict[str, Dict[str, Callable]] = {56 "file": {},57 "web": {},58 "feed": {},59 "stream": {},60 "repo": {},61 "email": {},62 "db": {},63 "api": {},64 "public_api": {},65 "mcp": {},66 "parquet": {},67 "arrow": {},68 "xml": {},69 "ingest": {},70 }7172 @classmethod73 def register(cls, task: str, name: str, method_func: Callable):74 """75 Register a custom ingestion method.7677 Args:78 task: Task type such as "file", "web", "feed", "stream",79 "repo", "email", "db", "api", "public_api", "mcp",80 "parquet", "xml", or "ingest"81 name: Method name82 method_func: Method function83 """84 if task not in cls._methods:85 cls._methods[task] = {}86 cls._methods[task][name] = method_func8788 @classmethod89 def get(cls, task: str, name: str) -> Optional[Callable]:90 """91 Get method by task and name.9293 Args:94 task: Task type such as "file", "web", "feed", "stream",95 "repo", "email", "db", "api", "public_api", "mcp",96 "parquet", "xml", or "ingest"97 name: Method name9899 Returns:100 Method function or None101 """102 return cls._methods.get(task, {}).get(name)结构是一张 task → {name → callable} 的两层字典，14 个 task 槽位在类体里预先开好。register 对未知 task 会先补空字典再写，所以第三方包可以往「file」里塞自己的解析函数。get 用两次 .get() 兜底，查不到就返回 None，不抛异常。list_all、unregister、clear 三个方法体略去，逻辑同构。
这里有个细节值得记：_methods 是类属性，register/get 都是 @classmethod，所以全进程共享同一张表，不需要实例化。文件末尾的 method_registry = MethodRegistry() 只是给习惯 import method_registry 的人留的别名。
IngestConfig：三层回退的配置
注册表解决「谁来干」，配置解决「怎么干」。IngestConfig 把配置来源叠成三层：程序内 set、环境变量、默认值。看 get 的回退链：
160 def get(self, key: str, default: Any = None) -> Any:161 """Get configuration value with fallback chain: config -> env -> default."""162 # Check config first163 if key in self._configs:164 return self._configs[key]165166 # Check environment variables167 env_key = f"INGEST_{key.upper()}"168 value = os.getenv(env_key)169 if value:170 try:171 # Try to convert to appropriate type172 if isinstance(default, int):173 return int(value)174 elif isinstance(default, float):175 return float(value)176 elif isinstance(default, bool):177 return value.lower() in ("true", "1", "yes", "on")178 return value179 except (ValueError, TypeError):180 pass181182 return default类型转换靠 default 的类型来猜：默认值是 int 就把 env 字符串转 int，是 bool 就认 true/1/yes/on 四组真值。_load_env_vars 在构造时还会扫描所有 INGEST_ 前缀变量自动入配置，这套「环境变量即配置」让 ingest 层不需要显式配置文件也能跑起来。文件摄取里的 max_size、web 摄取里的 timeout、respect_robots 都从这里取值，同一个部署可以通过环境变量统一调节各源行为，不必逐个 ingestor 传参。
ingest()：字符串前缀后缀驱动的自动路由
真正的一线入口是 semantica/ingest/methods.py 里的 ingest()。它先靠字符串特征猜源类型：
1334 # Auto-detect source type if not specified1335 if not source_type:1336 if isinstance(sources, (str, Path)):1337 source_str = str(sources)1338 source_str_lower = source_str.lower()1339 if source_str_lower.startswith(("http://", "https://")):1340 # Check if it's a feed URL1341 if any(1342 ext in source_str_lower1343 for ext in [".xml", "/feed", "/rss", "/atom"]1344 ):1345 source_type = "feed"1346 else:1347 source_type = "web"1348 elif source_str_lower.startswith(1349 ("postgresql://", "mysql://", "sqlite://", "oracle://", "mssql://")1350 ):1351 source_type = "db"1352 elif _is_scp_like_repo_source(source_str) or source_str_lower.startswith(1353 ("https://github.com", "https://gitlab.com")1354 ):1355 source_type = "repo"1356 elif source_str_lower.endswith(1357 (".ttl", ".owl", ".rdf", ".jsonld", ".n3", ".nt")1358 ):1359 source_type = "ontology"1360 elif source_str_lower.endswith((".parquet", ".pq")):1361 source_type = "parquet"1362 elif source_str_lower.endswith((".arrow", ".feather", ".ipc")):1363 source_type = "arrow"1364 elif source_str_lower.endswith(".xml"):1365 source_type = "xml"1366 else:1367 source_type = "file"1368 elif (1369 isinstance(sources, list)1370 and sources1371 and all(1372 str(source).lower().endswith((".parquet", ".pq")) for source in sources1373 )1374 ):1375 source_type = "parquet"1376 elif (1377 isinstance(sources, list)1378 and sources1379 and all(1380 str(source).lower().endswith((".arrow", ".feather", ".ipc"))1381 for source in sources1382 )1383 ):1384 source_type = "arrow"1385 elif (1386 isinstance(sources, list)1387 and sources1388 and all(str(source).lower().endswith(".xml") for source in sources)1389 ):1390 source_type = "xml"1391 else:1392 source_type = "file"判定顺序本身就是设计：先看协议前缀（http、数据库连接串、git 仓库地址），再看文件扩展名，最后落到 file。注意 http 分支里先查 feed 特征再落 web，.xml、/feed、/rss、/atom 四个子串任一命中就算 feed，这个优先级后面边界条件要专门挑它。
猜完类型就进入分发：
1394 # Route to appropriate ingestor1395 if source_type == "file":1396 return {"files": ingest_file(sources, method=method or "file", **kwargs)}1397 elif source_type == "web":1398 return {"content": ingest_web(sources, method=method or "url", **kwargs)}1399 elif source_type in {"public_api", "api"}:1400 return {1401 "data": ingest_public_api(sources, method=method or "endpoint", **kwargs)1402 }1403 elif source_type == "feed":1404 return {"feeds": ingest_feed(sources, method=method or "rss", **kwargs)}1405 elif source_type == "stream":每个分支返回一个字典，顶层 key 由源类型决定：文件是 files、web 是 content、API 是 data。这份「返回字典 key 因源而异」的设计，正是上一节说的「不统一数据形状」的佐证。ingest() 把「猜类型」和「查注册表」两件事拆开：猜类型靠字符串规则，查注册表靠 method_registry.get，后者由 get_ingest_method 一行转发完成。
SSRF 防线：先黑白名单，再钉死 IP
web 摄取的安全防线单独成文件 semantica/ingest/ssrf.py，753 行，是 ingest 包里最重的单文件之一。它要防的是场景还原里那类事故：用户给的 URL 把请求引向内网或云元数据端点。第一层是黑名单：
138BLOCKED_NETWORKS = (139 ipaddress.ip_network("0.0.0.0/8"),140 ipaddress.ip_network("10.0.0.0/8"),141 ipaddress.ip_network("100.64.0.0/10"), # CGNAT (RFC 6598) — routable inside carrier/cloud NAT142 ipaddress.ip_network("127.0.0.0/8"),143 ipaddress.ip_network("169.254.0.0/16"), # link-local / cloud metadata144 ipaddress.ip_network("172.16.0.0/12"),145 ipaddress.ip_network("192.168.0.0/16"),146 ipaddress.ip_network("::1/128"),147 ipaddress.ip_network("fc00::/7"),148 ipaddress.ip_network("fe80::/10"),149)150151152def _ip_is_blocked(addr: ipaddress._BaseAddress) -> bool:153 if (154 addr.is_private155 or addr.is_loopback156 or addr.is_link_local157 or addr.is_reserved158 or addr.is_multicast159 or addr.is_unspecified160 ):161 return True162 return any(addr in network for network in BLOCKED_NETWORKS)169.254.0.0/16 这条注释里写着 cloud metadata，正是云主机元数据服务的地址段。判定是双保险：先查 Python ipaddress 自带的六种属性（私有、环回、链路本地、保留、组播、未指定），再对照显式网段列表，任一命中即拦。
第二层是入口校验函数：
205def validate_url_for_request(206 url: str, *, allow_private_ips: bool = False207) -> None:208 """Validate that *url* is safe to fetch over HTTP(S).209210 Args:211 url: Absolute URL to validate.212 allow_private_ips: When True, skip private/loopback/link-local checks213 (for trusted internal deployments).214215 Raises:216 ValidationError: If the scheme is not http/https, the URL is malformed,217 or the host targets a blocked address space.218 """219 if not isinstance(url, str) or not url.strip():220 raise ValidationError("URL must be a non-empty string")221222 parsed = urlparse(url.strip())223 scheme = (parsed.scheme or "").lower()224 if scheme not in ALLOWED_URL_SCHEMES:225 raise ValidationError(226 f"URL scheme '{parsed.scheme}' is not permitted. "227 "Only http and https are allowed."228 )229 if not parsed.netloc:230 raise ValidationError(231 f"Invalid URL format: {url}. "232 "URL must include scheme (http/https) and netloc (domain)."233 )234235 host = parsed.hostname236 if not host:237 raise ValidationError(238 f"Invalid URL format: {url}. "239 "URL must include a hostname."240 )241242 if allow_private_ips:243 return244245 lowered = host.lower().rstrip(".")246 if lowered == "localhost" or lowered.endswith(".localhost"):247 raise ValidationError(f"URL host is not allowed: {host}")248249 try:250 literal_ip = ipaddress.ip_address(host)251 except ValueError:252 literal_ip = None253254 if literal_ip is not None:255 if _ip_is_blocked(literal_ip):256 raise ValidationError(f"URL points to a blocked address: {host}")257 return258259 if _hostname_resolves_to_blocked(host):260 raise ValidationError(261 f"URL host '{host}' resolves to a blocked (private/loopback/"262 "link-local) address"263 )校验顺序值得逐行记：先验 scheme 只允许 http/https，再验 netloc 和 hostname 存在，然后 allow_private_ips 早退（这是给可信内网部署留的后门），接着拦 localhost，接着处理字面 IP，最后对域名做 DNS 解析再查解析结果。域名路径调用 _hostname_resolves_to_blocked，它在共享线程池上跑 socket.getaddrinfo，用 2 秒超时兜底，解析失败或超时直接 raise ValidationError，即「失败即关闭」。
但这还不够。DNS 有 rebinding 的 TOCTOU 窗口：校验时解析出一个公网 IP，真正连接时同一个低 TTL 域名又解析成内网 IP。_resolve_pinned_ips 把「判定用的一次解析」和「连接用的 IP」合并成同一次，再用自定义 HTTPAdapter 把连接钉死在已验 IP 上。真正发请求的循环长这样：
610 try:611 while True:612 _apply_connection_pin(613 active_session,614 current_url,615 current_pinned_ips,616 _orig_http_adapter,617 _orig_https_adapter,618 _had_host_header,619 _orig_host_header,620 )621 response = requester(622 current_method,623 current_url,624 allow_redirects=False,625 **kwargs,626 )627628 if response.status_code not in _REDIRECT_STATUS_CODES:629 return response630631 if redirects_followed >= max_redirects:632 response.close()633 raise ValidationError(634 f"Exceeded maximum redirects ({max_redirects}) while "635 f"fetching '{url}'"636 )637638 location = response.headers.get("Location")639 if not location or not str(location).strip():640 response.close()641 raise ValidationError(642 f"Redirect from '{current_url}' is missing a Location header"643 )644645 next_url = urljoin(current_url, str(location).strip())646 next_host = (urlparse(next_url).hostname or "").lower()647 # A redirect back to the original host inherits the caller's648 # trust in that host (e.g. a same-host path redirect on a649 # private/localhost MCP server). A redirect to a *different*650 # host must not inherit that trust, even if the original host651 # was private/internal — otherwise a compromised or malicious652 # endpoint could redirect into arbitrary private address space653 # (e.g. cloud metadata) the caller never configured.654 hop_allow_private_ips = (655 allow_private_ips656 if next_host and next_host == _original_host657 else redirect_allow_private_ips658 )659 current_pinned_ips = _resolve_pinned_ips(660 next_url, allow_private_ips=hop_allow_private_ips661 )allow_redirects=False 是刻意为之：requests 默认自动跟随重定向，一个已验公网 URL 弹到内网就前功尽弃。所以这里手写循环，每一跳先 urljoin 拼出 Location，再对 next_url 重新跑 _resolve_pinned_ips，重定向计数超上限或 Location 缺失都关闭响应并抛错。跨源跳转时 hop_allow_private_ips 不再继承调用者对原 host 的信任，凭据剥离逻辑随后在同一循环里执行。这一段交互用时序图看更清楚：
还有一个容易被忽略的细节：_make_pinned_adapter 对代理是直接拒绝。_PinnedIPHTTPAdapter.get_connection_with_tls_context 开头检查 requests.utils.select_proxy，只要本次请求要走代理就抛 ValidationError，报错分支在 ssrf.py:378-383，文案写着「a proxy would resolve the host itself and bypass IP pinning」。代理在本进程之外自己做 DNS，等于把 pinning 想堵的 rebinding 窗口重新打开，所以宁可 fail closed 也不放行。同文件里 _get_session_lock（ssrf.py:455）给调用方传入的 session 挂一把私有锁，防止两个线程共享同一个 session 时各自的 mount/restore adapter 动作交错，把 A 请求钉死的 IP 串到 B 请求头上。这两处都在回应同一件事：安全不能靠「调用方别乱用」，要靠代码把乱用的路在结构上堵死。
摄取层溯源：一个只完成了一半的钩子
大纲把 ingest_provenance.py 列为「摄取层就带溯源」的候选入口。实读结果是：mixin 骨架在，但唯一具体子类是死代码。
21class IngestProvenanceMixin:22 """Mixin for ingest provenance tracking."""2324 def __init__(25 self,26 provenance: bool = False,27 agent_id: Optional[str] = None,28 is_automated: bool = True,29 **kwargs,30 ):31 self.provenance = provenance32 self._prov_manager = None33 self._agent_id = agent_id or self.__class__.__name__34 self._is_automated = is_automated3536 if provenance:37 try:38 from semantica.provenance import ProvenanceManager39 self._prov_manager = ProvenanceManager()40 except ImportError:41 self.provenance = Falsemixin 的设计意图清楚：provenance=True 时延迟导入 ProvenanceManager，导入失败就静默降级回 provenance=False。问题出在唯一的落地类：
44class PDFIngestorWithProvenance(IngestProvenanceMixin):45 """PDF ingestor with provenance tracking."""4647 def __init__(48 self,49 provenance: bool = False,50 agent_id: Optional[str] = None,51 is_automated: bool = True,52 **config,53 ):54 from .pdf_ingestor import PDFIngestor5556 IngestProvenanceMixin.__init__(57 self, provenance=provenance, agent_id=agent_id, is_automated=is_automated58 )59 self._ingestor = PDFIngestor(**config)from .pdf_ingestor import PDFIngestor 引用了一个不存在的模块。全仓检索 class PDFIngestor 与 pdf_ingestor 只命中这一个文件和文档里两处示例，semantica/ingest/ 下没有 pdf_ingestor.py。结论：这个类能 import（类体不执行那条语句），但一实例化就抛 ModuleNotFoundError。摄取层的溯源在挂载点上是断的，真正成体系的溯源在 semantica/provenance 模块，属于第八章。这里如实记录，检索关键词为 pdf_ingestor、PDFIngestor、IngestProvenanceMixin。
设计决策分析
先回答「每源一个文件」换来什么。文件摄取依赖 mimetypes、云摄取依赖 boto3/google.cloud/azure.storage，web 摄取依赖 bs4，repo 摄取依赖 GitPython，parquet 摄取依赖 pyarrow。如果把这些塞进一个大类，import 顶层包就会把所有重依赖全部拉起，任何一个装不上，整个 ingest 层报废。拆成每源一文件加 __getattr__ 懒加载，依赖才变成「用到才要」。__init__.py 里 _OPTIONAL_DEPENDENCY_MESSAGES 按模块名登记缺包提示，报错时把 git、bs4、pyarrow 三个名字翻译成人话，这是把「可观测性」建在导入边界的做法。以上是代码结构直接支撑的判断。
再看 SSRF 防线为什么不是一行 if url.startswith("http") 就完事。威胁模型里攻击者手里有整个 URL，所以从 scheme、netloc、hostname、字面 IP、域名解析结果到重定向每一跳都要查。_hostname_resolves_to_blocked 的注释明确写着 fail closed：DNS 解析失败或超时一律拒绝，宁可误杀不可放过。IP pinning 是第二个层次，它承认「先校验后连接」中间存在 DNS rebinding 的时间窗，于是把判定用的解析结果直接喂给连接层。文档 docs/glossary.md 对 SSRF 的解释只有一句「Semantica validates base_url at construction time to prevent SSRF in LLM gateway configurations」，讲的是 LLM 网关那一处，ingest 层的防线范围比这句文档说的更大。这是文档口径落后于代码的一处。
官方 docs/reference/ingest.md 在入口处写得明确：
9- 15+ ingestion adapters: files, web, SQL, Databricks, Snowflake, Kafka, MCP, Git repos, email10- PyArrow Parquet with column selection and partitioned dataset support11- XXE-safe lxml XML with optional XSD schema validation12- `ingest()` unified dispatcher: auto-detects source type from path or URL13- Each ingestor returns its own typed object (`FileObject`, `WebContent`, `TableData`, etc.)「Each ingestor returns its own typed object」这句话是理解整个 ingest 层的关键：官方文档自己就承认不追求统一对象。与之呼应，docs/glossary.md 里 Ingestion 词条写的是「into the pipeline as a unified SourceDocument」，但 SourceDocument 在全仓不是类名，只是 semantica/split/provenance_tracker.py 里一个字符串字段名。两处官方文档对「统一」的口径并不一致，代码偏向后者：统一的是 ingest() 分发和 MethodRegistry 注册，不是数据对象。
还要回答一个反直觉的问题：为什么用字符串前缀后缀猜类型，而不是让调用方显式传 source_type？因为统一入口的卖点是「给一个来源，自动路由」，显式传参只是可选覆盖。代价是误判，本章边界条件会给出确切案例。这个取舍没有文档背书，是推断。
再补一条关于 MethodRegistry 的推断：它选择类属性加 @classmethod，不选实例方法加依赖注入。好处是注册表成为进程内单例，任何模块 import 后都能直接 method_registry.register，插件作者不用先拿到某个对象引用。代价是全局可变状态，测试之间要记得 clear()，并发下不同测试互相污染的风险由使用者自己承担。这个选择适合「注册表本质就是全局的」这个前提，若换成一个库里要跑多套隔离的注册表，类属性形态就不够用。
边界条件剖析
如果 URL 是 https://example.com/feed，会走哪个 ingestor？
答案落在 methods.py:1341 到 1345 的 feed 判定：any(ext in source_str_lower for ext in [".xml", "/feed", "/rss", "/atom"]) 用子串包含判断，/feed 命中，source_type = "feed"，路由到 ingest_feed。注意它是「包含」不是「以这些结尾」：https://example.com/feedback 同样命中。反过来，一个以 .xml 结尾的普通网页（比如站点地图首页）也会被判成 feed，即使它根本不是 RSS/Atom。误判的后果由 ingest_feed 承担，它拿到的可能是一份 HTML。
如果文件超过默认 100MB 上限，会发生什么？
答案落在 file_ingestor.py:592 到 601：
592 # Check file size against limits593 file_size = file_path.stat().st_size594 max_size = FILE_SIZE_LIMITS.get(595 "MAX_DOCUMENT_SIZE", 104857600596 ) # 100MB default597 if file_size > max_size:598 raise ValidationError(599 f"File size {file_size:,} bytes exceeds maximum {max_size:,} bytes "600 f"({file_path.name})"601 )MAX_DOCUMENT_SIZE 缺省 104857600 字节，超了在真正读内容之前就抛 ValidationError，不会把 100MB 字节读进内存再拒。顺序是「存在性 → 是文件 → 大小 → 类型检测 → 读内容」，大小检查排在读内容前面，这正是防内存爆炸的分支。
如果一个域名同时解析出公网 IP 和内网 IP，会放行吗？
答案落在 ssrf.py:198 到 203 的循环：for info in resolved 逐个把解析结果喂给 _ip_is_blocked，任一个命中立即 return True。语义是「所有解析结果必须全部安全」，不是「至少有一个安全即可」。这堵死了「域名有 A 记录指向公网、又有一条 A 记录指向 169.254.169.254」这类双面解析。这是 fail closed 在解析层落到具体行号的体现。文件摄取那条链路的失败分支，用状态图收束一下：
任何一个前置校验失败都直接落到失败终态，不会带着半成品继续往下走。
横向对比
对比对象是 GraphRAG 的输入加载层，commit 6dad6d2，代码在 packages/graphrag-input/。两侧解决同一个问题：把外部原料变成下游可消费的文档流。
GraphRAG 这一侧的输入类型只有六种，用枚举定死：
9class InputType(StrEnum):10 """The input file type for the pipeline."""1112 Csv = "csv"13 """The CSV input type."""14 Text = "text"15 """The text input type."""16 Json = "json"17 """The JSON input type."""18 JsonLines = "jsonl"19 """The JSON Lines input type."""20 MarkItDown = "markitdown"21 """The MarkItDown input type."""22 Parquet = "parquet"23 """The Parquet input type."""2425 def __repr__(self):26 """Get a string representation."""27 return f'"{self.value}"'六种之外加新类型要改这个枚举，再在工厂的 match 语句里补一个 case：
63 match input_strategy:64 case InputType.Csv:65 from graphrag_input.csv import CSVFileReader6667 register_input_reader(InputType.Csv, CSVFileReader)68 case InputType.Text:69 from graphrag_input.text import TextFileReader7071 register_input_reader(InputType.Text, TextFileReader)72 case InputType.Json:73 from graphrag_input.json import JSONFileReader7475 register_input_reader(InputType.Json, JSONFileReader)76 case InputType.JsonLines:77 from graphrag_input.jsonl import JSONLinesFileReader7879 register_input_reader(InputType.JsonLines, JSONLinesFileReader)80 case InputType.MarkItDown:81 from graphrag_input.markitdown import MarkItDownFileReader8283 register_input_reader(InputType.MarkItDown, MarkItDownFileReader)84 case InputType.Parquet:85 from graphrag_input.parquet import ParquetFileReader8687 register_input_reader(InputType.Parquet, ParquetFileReader)88 case _:89 msg = f"InputConfig.type '{input_strategy}' is not registered in the InputReaderFactory. Registered types: {', '.join(input_reader_factory.keys())}."90 raise ValueError(msg)关键差别在数据形状上。GraphRAG 的每一种 reader 都把结果归一成同一个 TextDocument，文本 reader 就是例子：
16class TextFileReader(InputReader):17 """Reader implementation for text files."""1819 def __init__(self, file_pattern: str | None = None, **kwargs):20 super().__init__(21 file_pattern=file_pattern if file_pattern is not None else ".*\\.txt$",22 **kwargs,23 )2425 async def read_file(self, path: str) -> list[TextDocument]:26 """Read a text file into a list of documents.2728 Args:29 - path - The path to read the file from.3031 Returns32 -------33 - output - list with a TextDocument for each row in the file.34 """35 text = await self._storage.get(path, encoding=self._encoding)36 document = TextDocument(37 id=gen_sha512_hash({"text": text}, ["text"]),38 title=str(Path(path).name),39 text=text,40 creation_date=await self._storage.get_creation_date(path),41 raw_data=None,42 )43 return [document]TextDocument 有固定的 id、text、title、creation_date、raw_data 五个字段，id 由 text 的 SHA-512 哈希生成。GraphRAG 之所以可以只支持六种输入，是因为它的下游假设「原料最终都是文本行，一行一个文档」，CSV 用列名把行映射成 TextDocument，parquet、jsonl 同理。它把复杂度从「源类型爆炸」转移到「列映射」，所以输入层可以薄。TextDocument.get（text_document.py:30-50）把这一点落到代码：id、title、text、creation_date 四个标准字段直接返回，其余字段走点号路径从 raw_data 字典里取，CSV 的任意列因此都能被下游按名字访问，输入层不用为每一列生成一个类。
Semantica 走的是另一条路：源类型可以继续涨（22 个 ingestor 模块，注册表 14 个 task 槽位），但每个源产出自己的类型。代价是下游 pipeline 必须分别认识 FileObject、WebContent、TableData，没有一个统一的可遍历文档形状；收益是每个源能保留自己的结构化信息（文件的字节与 MIME、web 的链接表、数据库的行类型），不被压平成「一行文本」。换句话说：GraphRAG 用「统一对象 + 少量源」换薄，Semantica 用「每源对象 + 大量源」换信息的保真。企业多源场景里后者更匹配，因为 PDF 的页数和 DB 的列类型不该在同一层就被抹掉。SSRF 防线 GraphRAG 输入层没有对应物，检索关键词为 ssrf、validate_url、allow_private_ips，在 packages/graphrag-input/ 下无命中，原因是它面向本地文件与受信存储，web 抓取根本不在它的输入路径里。
另一个对照点是执行模型。GraphRAG 的 InputReader._iterate_files（input_reader.py:41-77）是异步生成器，async for doc in reader 逐文件产出，加载和下游消费可以流水线式重叠；Semantica 的 FileIngestor 全程同步，目录扫描用 rglob 一次性收集再逐个 ingest_file。同步版本写起来简单，但大目录时首个文件要等全部文件扫完才开始处理；异步版本省了这份等待，代价是调用方要活在事件循环里。两者没有高下，取决于下游是批处理还是流式管线。
互动演示设计
形态是模拟器。一句话结论：多源摄取是把「加新源」从改代码变成「登记一条记录」，把「连内网」从约定变成「机器拦在连接之前」。
舞台是一块接线台加三个闸口：中央一个 ingest() 调度闸机，左侧一排源插头（文件、URL、连接串、仓库地址），右侧一条 SSRF 安检带。比喻：调度闸机是写字楼前台，SSRF 安检带是金属探测门，MethodRegistry 是前台手里的工牌登记簿。
分步动画与字幕：
- 一个源插头插进来，字幕：「一个字符串进来了，先看它长什么样」。
ingest()读source_str_lower，逐条比对前缀后缀。 - 调度闸机亮灯指向某个 ingestor，字幕：「http 开头先问是不是 feed，连接串开头归 db，扩展名兜底归 file」。
- 安检带启动，字幕：「web 源过 SSRF 门，scheme 不是 http/https 直接拒」。
validate_url_for_request依次过 scheme、netloc、hostname。 - 域名被解析成 IP，字幕：「解析出的每一个 IP 都要安全，有一个内网就拦」。
- 连接被钉死在已验 IP 上，字幕：「校验和连接用同一次解析，DNS 换答案也来不及」。
- 服务器回 301，字幕：「重定向不停车，每一跳重新安检」。手写循环
allow_redirects=False，对Location再跑_resolve_pinned_ips。 - 结果落到对应对象，字幕：「文件是 FileObject，网页是 WebContent，各回各家」。
读者可操作项：在 repo 根跑 python3 -c "from semantica.ingest import FileIngestor; f = FileIngestor().ingest_file('README.md'); print(f.file_type, f.size, f.text[:40])"，观察扩展名检测与 UTF-8 解码；再跑 python3 -c "from semantica.ingest.ingest_provenance import PDFIngestorWithProvenance; PDFIngestorWithProvenance()"，观察它抛 ModuleNotFoundError，坐实溯源挂载点是断的。
逻辑轨迹面板伪代码（右侧标真实行号）：
route(source) # semantica/ingest/methods.py:1283
if source 以 http 开头 and 含 feed 特征: source_type = feed # methods.py:1341-1345
elif 连接串前缀: source_type = db # methods.py:1349-1352
elif 仓库地址: source_type = repo # methods.py:1354-1357
else: source_type = file # methods.py:1372
return registry.get(source_type, method) # registry.py:102
guard(url) # semantica/ingest/ssrf.py:205
if scheme not in {http, https}: 拒 # ssrf.py:221-225
if host 是 localhost: 拒 # ssrf.py:242-244
if host 解析出任何内网 IP: 拒 # ssrf.py:198-203
pin(IP 列表) -> 连接钉死 # ssrf.py:341-403
每个 301/302/307/308 重定向重新校验 # ssrf.py:611-661可迁移结论
值得抄的第一件是「每源一个文件 + 模块级 __getattr__ 懒加载」。最小成本形态：一个 registry 字典加一个顶层 __getattr__，把重依赖 import 从模块顶部移进访问时才触发的函数里。哪怕项目只有三个源，这一步也能让「只装 A 源的部署」不被 B 源的重依赖拖垮。这条不依赖 Python，任何有「惰性模块加载」机制的语言都能照搬，比如 Node 的 require 惰性调用或 Go 的 init 拆分。
第二件是 SSRF 的 fail closed 三件套：scheme 白名单、DNS 结果逐条查、重定向逐跳重验。最小成本形态只抄前两件：ALLOWED_URL_SCHEMES 一个 frozenset 加一段「解析后逐 IP 查 ipaddress.is_private」的函数，几十行就能挡住最常见的云元数据事故。第三件 IP pinning 是给「必须防 DNS rebinding」的高压场景，普通内部工具属于过度设计。
第三件是「返回字典 key 因源而异」。它不是 bug，是把「下游按源类型分派」的复杂度显式化。抄的时候要认清代价：调用方必须 if "files" in result 式地分派，拿不到一个统一可遍历的文档流；新增一个源还要记得它对应的 key，否则调用方会静默拿到 None。如果你的下游能接受统一文本行，学 GraphRAG 归一成 TextDocument 反而更省事，因为你换来的是整个下游只认一种文档形状。
过度设计的部分也要说：MethodRegistry 的 list_all/unregister/clear 三个方法、ingest_config 的 env 变量前缀自动扫描（INGEST_ 和 MCP_ 两套循环），在小项目里是没人会去动的表面面积。真正会被高频使用的是 register、get、ingest() 三个入口，其余可以在有真实第三方插件需求时再补。
思考题
-
ingest()对列表输入只做了 parquet、arrow、xml 三种「全元素同扩展名」的判断，其余都落到file。传一个["a.csv", "b.json"]会怎么路由？这条路径在methods.py哪个分支？它是否会产生你预期之外的结果？ -
FileObject.text对content先试 UTF-8 再试 latin-1，最后返回空串。一个 GBK 编码的中文文本会得到什么？这种「解码失败不抛异常而是静默返回」在摄取层是可接受的吗，还是应该把原始字节保留给解析层处理？结合file_ingestor.py:56-76回答。 -
SSRF 防线里
allow_private_ips=True只对「原始 host」生效，重定向到不同 host 时用allow_private_ips_on_redirect兜底。如果调用方把两者都设成True，一个被信任的内网端点能不能通过重定向把请求带到169.254.169.254？ssrf.py:654-661的hop_allow_private_ips分支在什么条件下会放行，什么条件下不会？ -
动手验证：在 semantica 仓库根目录依次运行
python3 -c "from semantica.ingest import FileIngestor; f = FileIngestor().ingest_file('README.md'); print(f.file_type, f.size, f.text[:40])"和python3 -c "from semantica.ingest.ingest_provenance import PDFIngestorWithProvenance; PDFIngestorWithProvenance()"。记录第一个命令打出的file_type是什么、第二个命令抛出的异常类型与消息，并解释第二个异常对应源码里的哪一行。