Haystack 的 Microsoft SharePoint 集成指南基于 Microsoft Graph 的检索与全文抓取管线实战【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本篇技术指南围绕 Haystack 的 Microsoft SharePoint 集成展开核心讲解MSSharePointRetriever通过 Microsoft Search/Graph API 检索 SharePoint 与 OneDrive 内容与MSSharePointFetcher通过 Graphshares端点下载命中条目的完整内容两个组件。读完本文你将掌握如何配置委托令牌认证、用 KQL 精化检索范围、按实体类型分发抓取结果并搭建一条完整的「OAuth 取令牌 → 检索 → 抓取 → 类型路由 → 文档转换」生产级查询管线。集成概览检索Retrieve→ 抓取Fetch→ 转换ConvertMicrosoft SharePoint 集成在 Haystack 生态中遵循检索与抓取分离的设计哲学由两个组件协同完成企业内容接入MSSharePointRetriever只负责找到什么它调用 Microsoft SearchGraphAPI 的POST /search/query把每个命中映射成一个 HaystackDocument其中content是搜索引擎返回的摘要snippetmeta携带资源元数据。它不下载、不转换底层文件。MSSharePointFetcher负责取到什么它消费检索器产出的Document或一组web_url把每个条目解析为ByteStream供下游转换器Converter继续处理。两者通过同一个web_url衔接——抓取器完全依赖检索器已暴露的web_url经由 Microsoft Graph 的shares端点解析目标因此不需要额外的 ID 或路径拼接逻辑。由于抓取器输出的是一批类型混合的ByteStream典型的下一步是接一个FileTypeRouter参见 FileTypeRouter 组件文档把不同 MIME 类型的流分别路由到PyPDFToDocument、DOCXToDocument、HTMLToDocument或 JSON 转换器对应组件文档见 PyPDFToDocument、DOCXToDocument、HTMLToDocument。抓取器按命中条目的实体类型分发抓取行为文件driveItem下载原始字节PDF、DOCX 等以二进制ByteStream返回列表项listItem以 JSONByteStream返回该项各列的值fieldsSharePoint 页面sitePage以 HTMLByteStream返回内容由页面的 web parts 构建。每个ByteStream的meta都会携带url、file_name、content_type以及归一化的entity_type取值为driveItem、listItem或sitePage。ByteStream是 Haystack 核心数据类其meta字段为可选的字典元数据参见 ByteStream 数据类定义。安装与认证准备安装集成包Microsoft SharePoint 集成以独立包发布包名为microsoft-sharepoint-haystackpip install microsoft-sharepoint-haystack如果使用OAuthTokenResolver获取令牌还需安装 OAuth 集成包pip install oauth-haystack委托权限delegated permissions与令牌输入两个组件都以每个用户per-user的access_token作为运行时输入而不是在初始化时静态配置。该令牌必须是携带 Microsoft Graph委托权限delegated permissions的 Bearer Token例如Files.Read.All读取文件driveItem所需Sites.Read.All读取列表项listItem与 SharePoint 页面sitePage所需同时用于站点/列表级范围限定。需要特别注意的是Microsoft Search API仅支持委托权限不支持应用程序权限因此令牌必须代表发起检索的用户身份。参考文档 Microsoft SharePoint API 参考 与 MSSharePointRetriever 组件文档、MSSharePointFetcher 组件文档 均强调了这一点。access_token参数同时接受普通字符串str与Secret两种类型Secret会在组件内部解析。官方推荐的取令牌方式是使用上游的OAuthTokenResolver组件详见 OAuthTokenResolver 组件文档它在管线运行时解析 OAuth 访问令牌并通过access_token输出槽发射下游组件通过普通连接消费令牌而无需关心令牌从何而来。OAuthTokenResolver是一个薄封装真正的工作委托给可插拔的token source均可从haystack_integrations.utils.oauth导入Token Source适用场景逐请求输入OAuthRefreshTokenSource单一固定身份持有刷新令牌由 source 兑换短时访问令牌并缓存无OAuthTokenExchangeSource多用户/多副本后端将入站用户断言兑换为下游令牌实现 RFC 8693 token exchange 与 Microsoft on-behalf-of 流程subject_tokenOAuthStaticTokenSource供应商签发长期有效、不过期令牌的场景如 Slack、Notion无对 Microsoft Graph 而言请求的 OAuth scope 形如https://graph.microsoft.com/Files.Read.All、https://graph.microsoft.com/Sites.Read.All并需配合offline_access才能获得刷新令牌。MSSharePointRetriever检索 SharePoint 与 OneDrive 内容MSSharePointRetriever位于查询管线的起点紧随提供令牌的OAuthTokenResolver之后。初始化时无需任何必填参数构造函数签名如下__init__( *, entity_types: list[str] | None None, top_k: int 10, fields: list[str] | None None, query_template: str | None None, graph_url: str DEFAULT_GRAPH_URL, timeout: float 30.0, max_retries: int 3 ) - None初始化参数详解参数类型默认值说明entity_typeslist[str] \| None[driveItem, listItem]要查询的 Microsoft Search 实体类型。默认值覆盖文件、文件夹、SharePoint 页面与新闻、列表项。其他合法值还有list与site具体支持的值与组合见 Microsoft 的 SearchRequest 资源文档top_kint10最多返回的文档数。映射到 Search API 的size参数超过单页上限时分页拉取fieldslist[str] \| NoneNone通过 Search API 的fields选择机制请求的额外资源属性列表仅对listItem和driveItem实体类型生效query_templatestr \| NoneNone用于限定搜索范围的可复用查询模板例如{searchTerms} path:https://contoso.sharepoint.com/sites/Team。字面量{searchTerms}占位符会在运行时被替换为实际查询词模板采用 KQL 语法graph_urlstrhttps://graph.microsoft.com/v1.0Microsoft Graph 基础 URL主权云sovereign clouds场景下需要覆盖timeoutfloat30.0每次请求 Microsoft Graph 的 HTTP 超时时间秒max_retriesint3对限流HTTP 429或瞬时服务器错误的最大重试次数校验规则若entity_types为空、top_k非正数或max_retries为负数初始化会抛出SharePointConfigError。run 与 run_asyncrun( query: str, access_token: str | Secret, top_k: int | None None ) - dict[str, list[Document]] run_async( query: str, access_token: str | Secret, top_k: int | None None ) - dict[str, list[Document]]两个方法参数一致querystr搜索查询串。可以直接在查询中内嵌KQLKeyword Query Language操作符来过滤结果例如filetype:docx、author:Jane Doe、path:https://contoso.sharepoint.com/sites/Teamaccess_tokenstr | Secret委托的 Microsoft Graph Bearer Token通常由上游OAuthResolver供应的普通字符串也接受Secret并在内部解析top_kint | None覆盖初始化时配置的top_k仅对本次运行生效。返回值为包含documents键的字典键值是对应的Document对象列表。异常行为若access_token是解析不出字符串的Secret抛出SharePointConfigError若 Microsoft Graph 返回错误响应抛出SharePointRequestError。输出 Document 的 meta 字段检索器把每个命中映射为Documentcontent为搜索摘要snippetmeta携带完整的资源元数据file_name、web_url、entity_type、mime_type、file_extensioncreated_date_time、last_modified_date_timecreated_by、last_modified_by以及下游抓取器按 ID 读取列表项和页面所需的 SharePoint 标识site_id、list_id、list_item_id、list_item_unique_id这些 SharePoint 标识是MSSharePointFetcher读取列表项与页面的关键输入即使仅凭web_url也能完成抓取保留这些 ID 字段能让抓取路径更健壮。检索范围精化KQL 与 query_template控制搜什么有三种互补手段entity_types限定实体类型集合查询内嵌 KQL如filetype:docx、author:Jane Doe、path:https://contoso.sharepoint.com/sites/Teamquery_template把常用范围限定固化成模板运行时仅替换{searchTerms}占位符例如retriever MSSharePointRetriever( query_template{searchTerms} path:https://contoso.sharepoint.com/sites/Team )独立使用示例from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) retriever MSSharePointRetriever(top_k5) # access_token 是每个用户独立的委托 Microsoft Graph Bearer Token。 result retriever.run( queryquarterly roadmap, access_tokenmy-delegated-graph-token, ) for doc in result[documents]: print(doc.meta[file_name], -, doc.meta[web_url])MSSharePointFetcher下载命中条目的完整内容MSSharePointFetcher通常位于检索器之后、路由/转换器之前。它同样没有必填的初始化参数__init__( *, graph_url: str DEFAULT_GRAPH_URL, timeout: float 30.0, max_retries: int 3, max_concurrent_requests: int 5, raise_on_failure: bool True ) - None初始化参数详解参数类型默认值说明graph_urlstrhttps://graph.microsoft.com/v1.0Microsoft Graph 基础 URL主权云场景覆盖timeoutfloat30.0每次请求 Microsoft Graph 的 HTTP 超时时间秒max_retriesint3对限流HTTP 429或瞬时服务器错误的最大重试次数max_concurrent_requestsint5run_async并发抓取的条目数量上限用于约束对 Microsoft Graph 的在途请求数、避免触发其速率限制。对同步run无效——同步版本逐条抓取raise_on_failureboolTrue为True时抓取失败抛出异常为False时记录日志并跳过该条目其余条目照常返回校验规则若max_retries为负数或max_concurrent_requests非正数抛出SharePointConfigError。run 与 run_asyncrun( access_token: str | Secret, targets: list[Document | str] ) - dict[str, list[ByteStream]] run_async( access_token: str | Secret, targets: list[Document | str] ) - dict[str, list[ByteStream]]access_tokenstr | Secret委托的 Microsoft Graph Bearer Token通常由上游OAuthTokenResolver供应发射普通字符串Secret也接受并在内部解析targetslist[Document \| str]要抓取的条目可以是MSSharePointRetriever产出的Document也可以是原始的 SharePoint/OneDriveweb_url字符串两种类型可以混在一个列表中。对Document抓取其 meta 中的web_url若 meta 中存在file_name、mime_type、entity_type及 SharePoint ID 则复用容器类命中如site、list因无可抽取内容会被跳过对原始 URL按文件探测失败则回退为列表项。返回值为包含streams键的字典键值是对应的ByteStream对象列表每个流的meta携带url、file_name、content_type、entity_type。异常行为若条目既不是Document也不是str或Secret无法解析为字符串抛出SharePointConfigError若抓取失败且raise_on_failureTrue抛出SharePointRequestError。独立使用示例from haystack_integrations.components.fetchers.microsoft_sharepoint import ( MSSharePointFetcher, ) fetcher MSSharePointFetcher() # access_token 是每个用户独立的委托 Microsoft Graph Bearer Token。 result fetcher.run( access_tokenmy-delegated-graph-token, targets[ https://contoso.sharepoint.com/sites/contoso-team/contoso-designs.docx, ], ) for stream in result[streams]: print(stream.meta[file_name], stream.meta[content_type])端到端管线检索 → 抓取 → 路由 → 转换下面这条查询管线把整个集成串在一起OAuthTokenResolver提供令牌MSSharePointRetriever搜索 SharePointMSSharePointFetcher下载命中条目FileTypeRouter把每个ByteStream发送给正确的转换器。注意resolver 唯一的access_token输出同时喂给检索器和抓取器同一输出槽可连接多个下游输入。from haystack import Pipeline from haystack.utils import Secret from haystack.components.routers import FileTypeRouter from haystack.components.converters import PyPDFToDocument, DOCXToDocument from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) from haystack_integrations.components.fetchers.microsoft_sharepoint import ( MSSharePointFetcher, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, https://graph.microsoft.com/Sites.Read.All, offline_access, ], ), ), ) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.add_component(fetcher, MSSharePointFetcher()) pipeline.add_component( router, FileTypeRouter( mime_types[ application/pdf, application/vnd.openxmlformats-officedocument.wordprocessingml.document, ], ), ) pipeline.add_component(pdf_converter, PyPDFToDocument()) pipeline.add_component(docx_converter, DOCXToDocument()) # 同一个令牌同时喂给检索器和抓取器。 pipeline.connect(resolver.access_token, retriever.access_token) pipeline.connect(resolver.access_token, fetcher.access_token) # 检索出的文档成为抓取器的 targets。 pipeline.connect(retriever.documents, fetcher.targets) # 把每个抓取到的 ByteStream 路由到匹配的转换器。 pipeline.connect(fetcher.streams, router.sources) pipeline.connect(router.application/pdf, pdf_converter.sources) pipeline.connect( router.application/vnd.openxmlformats-officedocument.wordprocessingml.document, docx_converter.sources, ) result pipeline.run({retriever: {query: quarterly roadmap}})运行这条管线时只需提供query——刷新令牌从环境变量MS_REFRESH_TOKEN读取通过 Haystack 的SecretAPI参见 Secret 管理概念访问令牌由 resolver 在运行时兑换并广播给两个消费者。如果只做检索不做全文抓取最小管线可以精简为 resolver retriever 两个组件pipeline Pipeline() pipeline.add_component(resolver, OAuthTokenResolver(...)) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.connect(resolver.access_token, retriever.access_token) result pipeline.run({retriever: {query: quarterly roadmap}}) documents result[retriever][documents]错误处理与并发控制重试与限流max_retries默认3在两个组件中都作用于两类错误被限流HTTP 429与瞬时服务器错误。对于高频抓取场景这可以有效吸收 Graph 侧的节流抖动。失败容忍抓取器独有的raise_on_failure参数决定了失败语义True默认任何一个条目抓取失败即抛出SharePointRequestError整次运行中断False失败被记录日志、该条目被跳过其余条目仍然返回。在批量同步大量文件的场景中把raise_on_failure设为False并辅以上游重试/补偿逻辑是更稳妥的生产配置。异步并发上限max_concurrent_requests默认5只约束run_async的在途请求数避免触发 Microsoft Graph 的速率限制同步run逐条抓取不受此参数影响。这是实现层面的明确设计从抓取器构造参数注释可见参见 MSSharePointFetcher API 参考。序列化支持to_dict / from_dict两个组件都实现了标准的 Haystack 序列化协议便于把组件状态保存到 YAML/JSON 并在不同进程中重建Haystack 的管线序列化机制参见 YAML 编组文档 与核心序列化实现 serialization.py# MSSharePointFetcher to_dict() - dict[str, Any] from_dict(data: dict[str, Any]) - MSSharePointFetcher # MSSharePointRetriever to_dict() - dict[str, Any] from_dict(data: dict[str, Any]) - MSSharePointRetrieverfrom_dict从字典表示反序列化出组件实例to_dict负责反向序列化。这意味着包含 SharePoint 组件的管线可以整体序列化、持久化、版本化管理并在部署时原样还原。小结Microsoft SharePoint 集成以检索器 抓取器的职责分离为骨架让 Haystack 管线既能低成本地做企业内容搜索只取摘要与元数据也能按需拉取完整内容进入 RAG 索引。实践要点可归纳为认证先行使用OAuthTokenResolver配OAuthRefreshTokenSource以委托权限换取 per-user 令牌scope 至少包含Files.Read.All与Sites.Read.All并始终把令牌作为运行输入而非初始化配置检索精化用entity_types、查询内嵌 KQL 与query_template三种手段收窄搜索范围抓取分发MSSharePointFetcher按driveItem/listItem/sitePage自动分发为原始字节、JSON 与 HTML统一以携带url/file_name/content_type/entity_type元数据的ByteStream输出稳健性善用max_retries429 重试、raise_on_failure失败跳过与max_concurrent_requests异步并发上限避免被 Graph 限流拖垮整条管线。进一步可参考仓库中的 MSSharePointFetcher 组件文档、MSSharePointRetriever 组件文档 与 OAuthTokenResolver 组件文档以及本参考文档 Microsoft SharePoint API 参考 获取更完整的签名与参数说明。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Haystack 的 Microsoft SharePoint 集成指南：基于 Microsoft Graph 的检索与全文抓取管线实战
Haystack 的 Microsoft SharePoint 集成指南基于 Microsoft Graph 的检索与全文抓取管线实战【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本篇技术指南围绕 Haystack 的 Microsoft SharePoint 集成展开核心讲解MSSharePointRetriever通过 Microsoft Search/Graph API 检索 SharePoint 与 OneDrive 内容与MSSharePointFetcher通过 Graphshares端点下载命中条目的完整内容两个组件。读完本文你将掌握如何配置委托令牌认证、用 KQL 精化检索范围、按实体类型分发抓取结果并搭建一条完整的「OAuth 取令牌 → 检索 → 抓取 → 类型路由 → 文档转换」生产级查询管线。集成概览检索Retrieve→ 抓取Fetch→ 转换ConvertMicrosoft SharePoint 集成在 Haystack 生态中遵循检索与抓取分离的设计哲学由两个组件协同完成企业内容接入MSSharePointRetriever只负责找到什么它调用 Microsoft SearchGraphAPI 的POST /search/query把每个命中映射成一个 HaystackDocument其中content是搜索引擎返回的摘要snippetmeta携带资源元数据。它不下载、不转换底层文件。MSSharePointFetcher负责取到什么它消费检索器产出的Document或一组web_url把每个条目解析为ByteStream供下游转换器Converter继续处理。两者通过同一个web_url衔接——抓取器完全依赖检索器已暴露的web_url经由 Microsoft Graph 的shares端点解析目标因此不需要额外的 ID 或路径拼接逻辑。由于抓取器输出的是一批类型混合的ByteStream典型的下一步是接一个FileTypeRouter参见 FileTypeRouter 组件文档把不同 MIME 类型的流分别路由到PyPDFToDocument、DOCXToDocument、HTMLToDocument或 JSON 转换器对应组件文档见 PyPDFToDocument、DOCXToDocument、HTMLToDocument。抓取器按命中条目的实体类型分发抓取行为文件driveItem下载原始字节PDF、DOCX 等以二进制ByteStream返回列表项listItem以 JSONByteStream返回该项各列的值fieldsSharePoint 页面sitePage以 HTMLByteStream返回内容由页面的 web parts 构建。每个ByteStream的meta都会携带url、file_name、content_type以及归一化的entity_type取值为driveItem、listItem或sitePage。ByteStream是 Haystack 核心数据类其meta字段为可选的字典元数据参见 ByteStream 数据类定义。安装与认证准备安装集成包Microsoft SharePoint 集成以独立包发布包名为microsoft-sharepoint-haystackpip install microsoft-sharepoint-haystack如果使用OAuthTokenResolver获取令牌还需安装 OAuth 集成包pip install oauth-haystack委托权限delegated permissions与令牌输入两个组件都以每个用户per-user的access_token作为运行时输入而不是在初始化时静态配置。该令牌必须是携带 Microsoft Graph委托权限delegated permissions的 Bearer Token例如Files.Read.All读取文件driveItem所需Sites.Read.All读取列表项listItem与 SharePoint 页面sitePage所需同时用于站点/列表级范围限定。需要特别注意的是Microsoft Search API仅支持委托权限不支持应用程序权限因此令牌必须代表发起检索的用户身份。参考文档 Microsoft SharePoint API 参考 与 MSSharePointRetriever 组件文档、MSSharePointFetcher 组件文档 均强调了这一点。access_token参数同时接受普通字符串str与Secret两种类型Secret会在组件内部解析。官方推荐的取令牌方式是使用上游的OAuthTokenResolver组件详见 OAuthTokenResolver 组件文档它在管线运行时解析 OAuth 访问令牌并通过access_token输出槽发射下游组件通过普通连接消费令牌而无需关心令牌从何而来。OAuthTokenResolver是一个薄封装真正的工作委托给可插拔的token source均可从haystack_integrations.utils.oauth导入Token Source适用场景逐请求输入OAuthRefreshTokenSource单一固定身份持有刷新令牌由 source 兑换短时访问令牌并缓存无OAuthTokenExchangeSource多用户/多副本后端将入站用户断言兑换为下游令牌实现 RFC 8693 token exchange 与 Microsoft on-behalf-of 流程subject_tokenOAuthStaticTokenSource供应商签发长期有效、不过期令牌的场景如 Slack、Notion无对 Microsoft Graph 而言请求的 OAuth scope 形如https://graph.microsoft.com/Files.Read.All、https://graph.microsoft.com/Sites.Read.All并需配合offline_access才能获得刷新令牌。MSSharePointRetriever检索 SharePoint 与 OneDrive 内容MSSharePointRetriever位于查询管线的起点紧随提供令牌的OAuthTokenResolver之后。初始化时无需任何必填参数构造函数签名如下__init__( *, entity_types: list[str] | None None, top_k: int 10, fields: list[str] | None None, query_template: str | None None, graph_url: str DEFAULT_GRAPH_URL, timeout: float 30.0, max_retries: int 3 ) - None初始化参数详解参数类型默认值说明entity_typeslist[str] \| None[driveItem, listItem]要查询的 Microsoft Search 实体类型。默认值覆盖文件、文件夹、SharePoint 页面与新闻、列表项。其他合法值还有list与site具体支持的值与组合见 Microsoft 的 SearchRequest 资源文档top_kint10最多返回的文档数。映射到 Search API 的size参数超过单页上限时分页拉取fieldslist[str] \| NoneNone通过 Search API 的fields选择机制请求的额外资源属性列表仅对listItem和driveItem实体类型生效query_templatestr \| NoneNone用于限定搜索范围的可复用查询模板例如{searchTerms} path:https://contoso.sharepoint.com/sites/Team。字面量{searchTerms}占位符会在运行时被替换为实际查询词模板采用 KQL 语法graph_urlstrhttps://graph.microsoft.com/v1.0Microsoft Graph 基础 URL主权云sovereign clouds场景下需要覆盖timeoutfloat30.0每次请求 Microsoft Graph 的 HTTP 超时时间秒max_retriesint3对限流HTTP 429或瞬时服务器错误的最大重试次数校验规则若entity_types为空、top_k非正数或max_retries为负数初始化会抛出SharePointConfigError。run 与 run_asyncrun( query: str, access_token: str | Secret, top_k: int | None None ) - dict[str, list[Document]] run_async( query: str, access_token: str | Secret, top_k: int | None None ) - dict[str, list[Document]]两个方法参数一致querystr搜索查询串。可以直接在查询中内嵌KQLKeyword Query Language操作符来过滤结果例如filetype:docx、author:Jane Doe、path:https://contoso.sharepoint.com/sites/Teamaccess_tokenstr | Secret委托的 Microsoft Graph Bearer Token通常由上游OAuthResolver供应的普通字符串也接受Secret并在内部解析top_kint | None覆盖初始化时配置的top_k仅对本次运行生效。返回值为包含documents键的字典键值是对应的Document对象列表。异常行为若access_token是解析不出字符串的Secret抛出SharePointConfigError若 Microsoft Graph 返回错误响应抛出SharePointRequestError。输出 Document 的 meta 字段检索器把每个命中映射为Documentcontent为搜索摘要snippetmeta携带完整的资源元数据file_name、web_url、entity_type、mime_type、file_extensioncreated_date_time、last_modified_date_timecreated_by、last_modified_by以及下游抓取器按 ID 读取列表项和页面所需的 SharePoint 标识site_id、list_id、list_item_id、list_item_unique_id这些 SharePoint 标识是MSSharePointFetcher读取列表项与页面的关键输入即使仅凭web_url也能完成抓取保留这些 ID 字段能让抓取路径更健壮。检索范围精化KQL 与 query_template控制搜什么有三种互补手段entity_types限定实体类型集合查询内嵌 KQL如filetype:docx、author:Jane Doe、path:https://contoso.sharepoint.com/sites/Teamquery_template把常用范围限定固化成模板运行时仅替换{searchTerms}占位符例如retriever MSSharePointRetriever( query_template{searchTerms} path:https://contoso.sharepoint.com/sites/Team )独立使用示例from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) retriever MSSharePointRetriever(top_k5) # access_token 是每个用户独立的委托 Microsoft Graph Bearer Token。 result retriever.run( queryquarterly roadmap, access_tokenmy-delegated-graph-token, ) for doc in result[documents]: print(doc.meta[file_name], -, doc.meta[web_url])MSSharePointFetcher下载命中条目的完整内容MSSharePointFetcher通常位于检索器之后、路由/转换器之前。它同样没有必填的初始化参数__init__( *, graph_url: str DEFAULT_GRAPH_URL, timeout: float 30.0, max_retries: int 3, max_concurrent_requests: int 5, raise_on_failure: bool True ) - None初始化参数详解参数类型默认值说明graph_urlstrhttps://graph.microsoft.com/v1.0Microsoft Graph 基础 URL主权云场景覆盖timeoutfloat30.0每次请求 Microsoft Graph 的 HTTP 超时时间秒max_retriesint3对限流HTTP 429或瞬时服务器错误的最大重试次数max_concurrent_requestsint5run_async并发抓取的条目数量上限用于约束对 Microsoft Graph 的在途请求数、避免触发其速率限制。对同步run无效——同步版本逐条抓取raise_on_failureboolTrue为True时抓取失败抛出异常为False时记录日志并跳过该条目其余条目照常返回校验规则若max_retries为负数或max_concurrent_requests非正数抛出SharePointConfigError。run 与 run_asyncrun( access_token: str | Secret, targets: list[Document | str] ) - dict[str, list[ByteStream]] run_async( access_token: str | Secret, targets: list[Document | str] ) - dict[str, list[ByteStream]]access_tokenstr | Secret委托的 Microsoft Graph Bearer Token通常由上游OAuthTokenResolver供应发射普通字符串Secret也接受并在内部解析targetslist[Document \| str]要抓取的条目可以是MSSharePointRetriever产出的Document也可以是原始的 SharePoint/OneDriveweb_url字符串两种类型可以混在一个列表中。对Document抓取其 meta 中的web_url若 meta 中存在file_name、mime_type、entity_type及 SharePoint ID 则复用容器类命中如site、list因无可抽取内容会被跳过对原始 URL按文件探测失败则回退为列表项。返回值为包含streams键的字典键值是对应的ByteStream对象列表每个流的meta携带url、file_name、content_type、entity_type。异常行为若条目既不是Document也不是str或Secret无法解析为字符串抛出SharePointConfigError若抓取失败且raise_on_failureTrue抛出SharePointRequestError。独立使用示例from haystack_integrations.components.fetchers.microsoft_sharepoint import ( MSSharePointFetcher, ) fetcher MSSharePointFetcher() # access_token 是每个用户独立的委托 Microsoft Graph Bearer Token。 result fetcher.run( access_tokenmy-delegated-graph-token, targets[ https://contoso.sharepoint.com/sites/contoso-team/contoso-designs.docx, ], ) for stream in result[streams]: print(stream.meta[file_name], stream.meta[content_type])端到端管线检索 → 抓取 → 路由 → 转换下面这条查询管线把整个集成串在一起OAuthTokenResolver提供令牌MSSharePointRetriever搜索 SharePointMSSharePointFetcher下载命中条目FileTypeRouter把每个ByteStream发送给正确的转换器。注意resolver 唯一的access_token输出同时喂给检索器和抓取器同一输出槽可连接多个下游输入。from haystack import Pipeline from haystack.utils import Secret from haystack.components.routers import FileTypeRouter from haystack.components.converters import PyPDFToDocument, DOCXToDocument from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) from haystack_integrations.components.fetchers.microsoft_sharepoint import ( MSSharePointFetcher, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, https://graph.microsoft.com/Sites.Read.All, offline_access, ], ), ), ) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.add_component(fetcher, MSSharePointFetcher()) pipeline.add_component( router, FileTypeRouter( mime_types[ application/pdf, application/vnd.openxmlformats-officedocument.wordprocessingml.document, ], ), ) pipeline.add_component(pdf_converter, PyPDFToDocument()) pipeline.add_component(docx_converter, DOCXToDocument()) # 同一个令牌同时喂给检索器和抓取器。 pipeline.connect(resolver.access_token, retriever.access_token) pipeline.connect(resolver.access_token, fetcher.access_token) # 检索出的文档成为抓取器的 targets。 pipeline.connect(retriever.documents, fetcher.targets) # 把每个抓取到的 ByteStream 路由到匹配的转换器。 pipeline.connect(fetcher.streams, router.sources) pipeline.connect(router.application/pdf, pdf_converter.sources) pipeline.connect( router.application/vnd.openxmlformats-officedocument.wordprocessingml.document, docx_converter.sources, ) result pipeline.run({retriever: {query: quarterly roadmap}})运行这条管线时只需提供query——刷新令牌从环境变量MS_REFRESH_TOKEN读取通过 Haystack 的SecretAPI参见 Secret 管理概念访问令牌由 resolver 在运行时兑换并广播给两个消费者。如果只做检索不做全文抓取最小管线可以精简为 resolver retriever 两个组件pipeline Pipeline() pipeline.add_component(resolver, OAuthTokenResolver(...)) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.connect(resolver.access_token, retriever.access_token) result pipeline.run({retriever: {query: quarterly roadmap}}) documents result[retriever][documents]错误处理与并发控制重试与限流max_retries默认3在两个组件中都作用于两类错误被限流HTTP 429与瞬时服务器错误。对于高频抓取场景这可以有效吸收 Graph 侧的节流抖动。失败容忍抓取器独有的raise_on_failure参数决定了失败语义True默认任何一个条目抓取失败即抛出SharePointRequestError整次运行中断False失败被记录日志、该条目被跳过其余条目仍然返回。在批量同步大量文件的场景中把raise_on_failure设为False并辅以上游重试/补偿逻辑是更稳妥的生产配置。异步并发上限max_concurrent_requests默认5只约束run_async的在途请求数避免触发 Microsoft Graph 的速率限制同步run逐条抓取不受此参数影响。这是实现层面的明确设计从抓取器构造参数注释可见参见 MSSharePointFetcher API 参考。序列化支持to_dict / from_dict两个组件都实现了标准的 Haystack 序列化协议便于把组件状态保存到 YAML/JSON 并在不同进程中重建Haystack 的管线序列化机制参见 YAML 编组文档 与核心序列化实现 serialization.py# MSSharePointFetcher to_dict() - dict[str, Any] from_dict(data: dict[str, Any]) - MSSharePointFetcher # MSSharePointRetriever to_dict() - dict[str, Any] from_dict(data: dict[str, Any]) - MSSharePointRetrieverfrom_dict从字典表示反序列化出组件实例to_dict负责反向序列化。这意味着包含 SharePoint 组件的管线可以整体序列化、持久化、版本化管理并在部署时原样还原。小结Microsoft SharePoint 集成以检索器 抓取器的职责分离为骨架让 Haystack 管线既能低成本地做企业内容搜索只取摘要与元数据也能按需拉取完整内容进入 RAG 索引。实践要点可归纳为认证先行使用OAuthTokenResolver配OAuthRefreshTokenSource以委托权限换取 per-user 令牌scope 至少包含Files.Read.All与Sites.Read.All并始终把令牌作为运行输入而非初始化配置检索精化用entity_types、查询内嵌 KQL 与query_template三种手段收窄搜索范围抓取分发MSSharePointFetcher按driveItem/listItem/sitePage自动分发为原始字节、JSON 与 HTML统一以携带url/file_name/content_type/entity_type元数据的ByteStream输出稳健性善用max_retries429 重试、raise_on_failure失败跳过与max_concurrent_requests异步并发上限避免被 Graph 限流拖垮整条管线。进一步可参考仓库中的 MSSharePointFetcher 组件文档、MSSharePointRetriever 组件文档 与 OAuthTokenResolver 组件文档以及本参考文档 Microsoft SharePoint API 参考 获取更完整的签名与参数说明。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考