JaydenAI头像
[LLM Wiki理论与实践-06]详细解析OpenWiki如何摄取外部数据源封面图

[LLM Wiki理论与实践-06]详细解析OpenWiki如何摄取外部数据源

前面两篇文章分别介绍了如何采用OpenWiki的Personal模式构建属于个人的知识库，以及这个由Wiki页面构成的知识库具有怎样的结构，这篇文章聚焦知识库原始数据的摄取，看看源自如下这些来源的数据是如何被对应的Connector摄取。

  • custom-mcp: 摄取任意已配置的HTTP或stdio MCP服务器数据（仅限只读安全工具），提供高度定制化的外部异构系统接入。
  • git-repo: 摄取本地版本库数据，读取仓库路径并生成紧凑的清单文件，提供本地代码与资产上下文。
  • x: 摄取主页时间线、用户帖子、提及、书签和列表推文，提供个人社交网络与外部关注动态。
  • notion: 摄取Notion工作区内的页面与文档，提供个人知识库与结构化笔记上下文。
  • google: 摄取最近的邮件通信数据，提供个人日常沟通与事务往来信息。
  • web-search: 摄取实时的公开网页搜索结果，提供广泛的外部互联网信息追踪。
  • hackernews: 摄取公开的科技新闻与讨论数据，提供科技、创业领域的行业趋势与社区信号。
  • slack: 摄取团队工作区内的对话与消息，提供企业内部协作与即时通讯上下文。

1. ConnectorRuntime

Connector在OpenWiki中被表示成如下这个ConnectorRuntime类型，ConnectorDefinition提供了Connector的基本定义，具体的数据摄取操作则体现在ingest方法上。

export type ConnectorRuntime = ConnectorDefinition & {
  ingest: (options?: ConnectorIngestOptions) => Promise<ConnectorIngestResult>;
};

1.1 ConnectorDefinition

ConnectorDefinition是OpenWiki连接器子系统的静态元数据契约。它只负责声明一个连接器是什么以及需要什么配置，不包含实际的运行逻辑（运行逻辑由扩展了它的ConnectorRuntime通过ingest方法提供）。

export type ConnectorDefinition = {
  backend: ConnectorBackend;
  description: string;
  displayName: string;
  id: ConnectorId;
  mode: "code" | "personal";
  requiredEnv: string[];
  supportsAgenticDiscovery: boolean;
};

export type ConnectorBackend = "direct-api" | "local-git" | "mcp-http" | "mcp-stdio";

各字段说明如下：

  • backend：连接器与外部数据源通信的传输方式，对应类型ConnectorBackend提供了分别代表四种传输方式的字面量；
  • description：连接器功能的人类可读说明，描述其数据来源与用途；
  • displayName：面向用户展示的可读名称，在界面中代替id呈现；
  • id：连接器的唯一标识符，用于注册表查找和状态持久化；
  • mode：运行模式，code为仓库文档生成，personal为个人知识库；
  • requiredEnv：正常工作所需的环境变量名称列表，用于判定配置是否就绪；
  • supportsAgenticDiscovery：是否支持Agent自主发现模式，否则采用确定性摄取逻辑。

由于Connector在进行数据摄取的时候都需要提供用户凭证（比如API-KEY），所以只有在独立运行时才能正常工作。由于OpenWiki在code模式是以MCP Server的方式运行在Codeing Agent的进程中（采用stdio协议），无法获取凭证，所以OpenWiki提供的9种Connector都只能应用在personal模型下。实际上code模式下根本就不会有Connector被创建。

1.2 ConnectorIngestOptions

ConnectorRuntime用来完成数据摄取的ingest方法是一个将ConnectorIngestOptions和ConnectorIngestResult作为输入和输出的异步方法。ConnectorIngestOptions是Connector进行数据摄取时的运行时配置选项，用于在单次摄取时传入可选的配置、实例标识、数量/时间范围限制、本地仓库根路径以及要处理的数据流列表，从而让同一连接器按不同参数灵活执行数据拉取。

export type ConnectorIngestOptions = {
  connectorConfig?: Record<string, unknown>;
  instanceId?: string;
  limit?: number;
  repoRoot?: string;
  streams?: string[];
  windowHours?: number;
};

各字段说明如下：

  • connectorConfig：连接器的自定义配置项，以键值对形式传入，供具体连接器读取个性化参数；
  • instanceId：连接器实例的唯一标识，用于区分同一连接器的多个实例；
  • limit：本次摄取的最大条目数量限制；
  • repoRoot：仓库根目录路径，供需要访问本地仓库的连接器使用；
  • streams：要摄取的数据流名称列表，用于筛选特定来源或频道；
  • windowHours：摄取的时间窗口（小时），限定只拉取该时间范围内的数据。

1.3 ConnectorIngestResult

ConnectorRuntime的ingest方法在完成了数据摄取后由ConnectorIngestResult对象返回摄取的结果。ConnectorIngestResult记录一次摄取产生的连接器标识、摘要消息、原始文件列表、运行 ID、状态持久化路径、执行状态（成功/跳过/出错）以及非致命警告，用于追踪、展示和后续处理单次摄取的结果。

export type ConnectorIngestResult = {
  connectorId: ConnectorId;
  message: string;
  rawFiles: string[];
  runId: string;
  statePath: string;
  status: "error" | "skipped" | "success";
  warnings: string[];
};

export type ConnectorId =
  | "custom-mcp"
  | "git-repo"
  | "google"
  | "hackernews"
  | "langsmith"
  | "notion"
  | "slack"
  | "web-search"
  | "x";

各字段说明如下：

  • connectorId：产生本次结果的连接器标识，取自ConnectorId联合类型，后者体现了目前默认支持的9种数据来源；
  • message：本次摄取结果的摘要信息，用于向用户或日志说明执行情况；
  • rawFiles：本次摄取生成的原始文件路径列表；
  • runId：本次摄取运行的唯一标识，用于追踪和关联状态；
  • statePath：本次运行状态持久化的路径；
  • status：运行结果状态，success成功、skipped跳过、error出错；
  • warnings：本次运行产生的非致命警告信息列表。

2. 配置的读取

Connector通常都具有属于自己的专属配置，并且会根据配置结构定义对应的配置类型。按照约定，配置文件以JSON的形式存放在connectors/{connector-id}/config.json中。OpenWiki定义了如下这个readConnectorConfig函数根据指定的connector-id加载对应的配置文件，并将其反序列化成指定的配置类型T。参数defaultConfig提供默认的配置选项。

export async function readConnectorConfig<T extends object>(
  connectorId: ConnectorId,
  defaultConfig: T,
): Promise<T> {
  await ensureConnectorHome(connectorId);

  try {
    return {
      ...defaultConfig,
      ...(JSON.parse(
        await readFile(getConnectorConfigPath(connectorId), "utf8"),
      ) as T),
    };
  } catch (error) {
    if (isFileNotFoundError(error)) {
      return defaultConfig;
    }

    throw error;
  }
}

3. 原始内容的存储

OpenWiki在执行某个Connector的摄取操作前会调用如下这个createRunId函数创建一个标识当前摄取操作的ID，我们将其称为run_id。在完成数据摄取任务后会将得到的内容写入connectors/{connector-id}/raw/{run_id}中目录下。具体的写入工作由如下这个writeRawJson函数完成，该函数返回写入文件的完整路径。

export function createRunId(): string {
  return new Date().toISOString().replace(/[:.]/gu, "-");
}

export async function writeRawJson(
  connectorId: ConnectorId,
  runId: string,
  filename: string,
  value: unknown,
): Promise<string> {
  await ensureConnectorHome(connectorId);
  const filePath = path.join(getConnectorRawDir(connectorId), runId, filename);
  await writePrivateJson(filePath, value);
  return filePath;
}

4. 状态的维护

Connector针对每次数据摄取都会记录在connectors/{connector-id}/state.json这个状态文件中，该文件还会单独记录最后一次执行的时间。OpenWiki定义了如下这个ConnectorState类型专门定义某个Collector的当前状态。

export type ConnectorState = {
  lastRunAt?: string;
  latestIds?: Record<string, string>;
  runs?: ConnectorRunSummary[];
  version: 1;
};

各状态字段说明如下：

  • lastRunAt： 最后执行数据摄取的时间戳；
  • latestIds：标识最终一次数据摄取操作的ID；
  • runs：历次数据摄取操作的摘要描述；
  • version：当前版本

在完成摄取后会将内容存储一个或者多个源文件中。用来承载每次摄取摘要信息的ConnectorRunSummary分别利用runId和rawFiles字段记录了当前的run_id和写入的文件路径。at、status和warnings字段分别记录了摄取的时间戳、状态（"error" | "skipped" | "success"）和警告信息。

export type ConnectorRunSummary = {
  at: string;
  rawFiles: string[];
  runId: string;
  status: ConnectorIngestResult["status"];
  warnings: string[];
};

OpenWiki定义了如下所示的readConnectorState和updateStateWithRun函数实现了针对Collector状态的读写

export async function readConnectorState(
  connectorId: ConnectorId,
): Promise<ConnectorState> {
  await ensureConnectorHome(connectorId);
  try {
    return JSON.parse(
      await readFile(getConnectorStatePath(connectorId), "utf8"),
    ) as ConnectorState;
  } catch (error) {
    if (isFileNotFoundError(error)) {
      return { version: 1 };
    }
    throw error;
  }
}

export function updateStateWithRun(
  state: ConnectorState,
  run: NonNullable<ConnectorState["runs"]>[number],
): ConnectorState {
  return {
    ...state,
    lastRunAt: run.at,
    runs: [run, ...(state.runs ?? [])].slice(0, 20),
    version: 1,
  };
}

5. 基于web-search的Connector

OpenWiki默认提供的9种Connector的设计和实现大同小异，它们都提供了用来创建对应ConnectorRuntime的工厂方法，接下来就以web-search这种Connector为例，看看它如何利用Tavily进行网络搜索摄取对应元数据。Tavily是一个专为LLM和AI Agent 设计的搜索API，核心目标是让Agent以更少的token消耗获得结构化、可直接使用的高质量搜索结果。这种Connector对应的工厂函数定义如下：

export function createWebSearchConnector(): ConnectorRuntime {
  return {
    ...definition,
    ingest,
  };
}

5.1 基础定义

对于createWebSearchConnector工厂函数创建的ConnectorRuntime对象来说，其基础定义体现在definition常量上，该常量定义如下：

const definition: ConnectorDefinition = {
  backend: "direct-api",
  description:"Fetches web search results with Tavily through the LangChain Tavily integration.",
  displayName: "Web Search",
  id: "web-search",
  mode: "personal",
  requiredEnv: [OPENWIKI_TAVILY_API_KEY_ENV_KEY],
  supportsAgenticDiscovery: false,
};
export const OPENWIKI_TAVILY_API_KEY_ENV_KEY = "TAVILY_API_KEY";

由于Connector直接调用Tavily API来实施网络搜索，所以其backend字段体现的后端类型为direct-api，并要求利用环境变量TAVILY_API_KEY来提供对应的API-KEY。

5.2 WebSearchConfig

WebSearchConfig是这个Connector的配置类型，用于控制是否启用、搜索范围（包含/排除域名）、结果形态（答案、图片、原始内容、数量）、查询语句、搜索深度、时间范围与主题类别。

type WebSearchConfig = {
  enabled?: boolean;
  excludeDomains?: string[];
  includeAnswer?: boolean;
  includeDomains?: string[];
  includeImages?: boolean;
  includeRawContent?: boolean;
  maxResults?: number;
  queries?: string[];
  searchDepth?: "advanced" | "basic";
  timeRange?: "day" | "month" | "week" | "year";
  topic?: "general" | "news";
};

各配置字段说明如下：

  • enabled：是否启用该web-search配置；
  • excludeDomains：搜索时排除的域名列表；
  • includeAnswer：是否让搜索服务返回一段汇总答案；
  • includeDomains：限定只在这些域名内搜索；
  • includeImages：是否在结果中包含图片；
  • includeRawContent：是否返回原始页面内容而非摘要；
  • maxResults：单次搜索返回的最大结果数量；
  • queries：预定义的搜索查询语句列表；
  • searchDepth：搜索深度，basic快速、advanced更深入；
  • timeRange：结果时间范围，限定为day、week、month或year内；
  • topic：搜索主题类别，general通用或news新闻。

如有没有对配置选项进行显式设置，其默认的兜底配置体现在如下的代码中。

async function ingest(
  options: ConnectorIngestOptions = {},
): Promise<ConnectorIngestResult> {
  ...
  const config = {
    ...(await readConnectorConfig<WebSearchConfig>("web-search", {
      enabled: true,
      includeAnswer: true,
      includeImages: false,
      includeRawContent: false,
      maxResults: 5,
      queries: [],
      searchDepth: "basic",
      topic: "general",
    })),
    ...((options.connectorConfig ?? {}) as WebSearchConfig),
  };
}

LLM Wiki理论与实践-04：使用OpenWiki的Personal模式构建知识库提供的配置文件就是针对这个结构定义的：

{
    "queries": [
      "美国中期选举",
      "US midterm elections",
      "2026 midterm elections",
      "congressional race 2026"
    ]
}

5.3 调用Tavily API

OpenWiki将通过调用Tavily API实施网络搜索实现在一个LangChain工具中，工具类型的声明体现为如下这个派生自StructuredTool的TavilySearch类型。

declare class TavilySearch extends StructuredTool<typeof inputSchema> {
  static lc_name(): string;
  description: string;
  name: string;
  schema: z.ZodObject<{
    query: z.ZodString;
    includeDomains: z.ZodOptional<z.ZodArray<z.ZodString, "many">>;
    excludeDomains: z.ZodOptional<z.ZodArray<z.ZodString, "many">>;
    searchDepth: z.ZodOptional<z.ZodEnum<["basic", "advanced"]>>;
    includeImages: z.ZodOptional<z.ZodBoolean>;
    timeRange: z.ZodOptional<z.ZodEnum<["day", "week", "month", "year"]>>;
    topic: z.ZodOptional<z.ZodEnum<["general", "news", "finance"]>>;
  }, "strip", z.ZodTypeAny, {
    query: string;
    includeDomains?: string[] | undefined;
    excludeDomains?: string[] | undefined;
    searchDepth?: "advanced" | "basic" | undefined;
    includeImages?: boolean | undefined;
    timeRange?: "day" | "month" | "week" | "year" | undefined;
    topic?: "finance" | "general" | "news" | undefined;
  }, {
    query: string;
    includeDomains?: string[] | undefined;
    excludeDomains?: string[] | undefined;
    searchDepth?: "advanced" | "basic" | undefined;
    includeImages?: boolean | undefined;
    timeRange?: "day" | "month" | "week" | "year" | undefined;
    topic?: "finance" | "general" | "news" | undefined;
  }>;
  apiBaseUrl?: string;
  maxResults?: number;
  includeImages?: boolean;
  includeImageDescriptions?: boolean;
  includeAnswer?: boolean;
  includeRawContent?: boolean | "markdown" | "text";
  includeDomains?: string[];
  excludeDomains?: string[];
  searchDepth?: SearchDepth;
  topic?: TopicType;
  days?: number;
  timeRange?: TimeRange;
  chunksPerSource?: number;
  country?: string;
  autoParameters?: boolean;
  includeFavicon?: boolean;
  includeUsage?: boolean;
  handleToolError: boolean;
  apiWrapper: TavilySearchAPIWrapper;
  constructor(params?: TavilySearchAPIRetrieverFields);
  _call(input: InferInteropZodOutput<typeof inputSchema>, _runManager?: CallbackManagerForToolRun): Promise<TavilySearchResponse | {
    error: string;
  }>;
}

5.4 完整的摄取流程

Connector的摄取操作实现在如下这个ingest函数中，具体的流程如下：

  • 调用readConnectorConfig函数读取配置文件并将其反序列化成WebSearchConfig对象；
  • 调用readConnectorState函数读取状态文件并返回ConnectorState对象；
  • 如果配置将Connector设置成禁用状态，返回一个状态为skipped的ConnectorIngestResult对象；
  • 提取环境变量TAVILY_API_KEY，如果失败则返回一个状态为error的ConnectorIngestResult对象；
  • 如果WebSearchConfig对象没有利用queries字段提供查询文本，返回一个状态为skipped的ConnectorIngestResult对象；
  • 创建工具对象TavilySearch，并利用它调用Tavily API来实施网络搜索，并调用writeRawJson函数将内容写入原始文件；
  • 调用writeConnectorState函数记录最新的状态；
  • 返回一个状态为success的ConnectorIngestResult对象；
async function ingest(
  options: ConnectorIngestOptions = {},
): Promise<ConnectorIngestResult> {
  const runId = createRunId();
  const config = {
    ...(await readConnectorConfig<WebSearchConfig>("web-search", {
      enabled: true,
      includeAnswer: true,
      includeImages: false,
      includeRawContent: false,
      maxResults: 5,
      queries: [],
      searchDepth: "basic",
      topic: "general",
    })),
    ...((options.connectorConfig ?? {}) as WebSearchConfig),
  };
  const state = await readConnectorState("web-search");
  const warnings: string[] = [];
  const rawFiles: string[] = [];

  if (!config.enabled) {
    return {
      connectorId: "web-search",
      message: `Web Search connector is not enabled. Set enabled=true in ${openWikiConnectorsDisplayPath}/web-search/config.json.`,
      rawFiles,
      runId,
      statePath: `${openWikiConnectorsDisplayPath}/web-search/state.json`,
      status: "skipped",
      warnings,
    };
  }

  const tavilyApiKey = process.env[OPENWIKI_TAVILY_API_KEY_ENV_KEY];
  if (!tavilyApiKey) {
    return {
      connectorId: "web-search",
      message: `${OPENWIKI_TAVILY_API_KEY_ENV_KEY} is required for Web Search ingestion.`,
      rawFiles,
      runId,
      statePath: `${openWikiConnectorsDisplayPath}/web-search/state.json`,
      status: "error",
      warnings,
    };
  }

  const queries = normalizeStringArray(config.queries);
  if (queries.length === 0) {
    return {
      connectorId: "web-search",
      message: `No web search queries configured. Add queries to ${openWikiConnectorsDisplayPath}/web-search/config.json.`,
      rawFiles,
      runId,
      statePath: `${openWikiConnectorsDisplayPath}/web-search/state.json`,
      status: "skipped",
      warnings,
    };
  }

  const limit = getOptionLimit(options.limit, config.maxResults);
  const timeRange = getWindowedTimeRange(config.timeRange, options.windowHours);
  const tool = new TavilySearch({
    excludeDomains: normalizeStringArray(config.excludeDomains),
    includeAnswer: config.includeAnswer ?? true,
    includeDomains: normalizeStringArray(config.includeDomains),
    includeImages: config.includeImages ?? false,
    includeRawContent: config.includeRawContent ?? false,
    maxResults: limit,
    searchDepth: normalizeSearchDepth(config.searchDepth),
    tavilyApiKey,
    timeRange,
    topic: normalizeTopic(config.topic),
  });

  const results = [];
  for (const query of queries) {
    results.push({
      query,
      response: (await tool.invoke({ query })) as TavilySearchResult,
    });
  }

  rawFiles.push(
    await writeRawJson("web-search", runId, "web-search-results.json", {
      fetchedAt: new Date().toISOString(),
      instanceId: options.instanceId,
      maxResults: limit,
      queryCount: queries.length,
      results,
      searchDepth: normalizeSearchDepth(config.searchDepth),
      timeRange,
      topic: normalizeTopic(config.topic),
      windowHours: normalizeWindowHours(options.windowHours),
    }),
  );

  await writeConnectorState(
    "web-search",
    updateStateWithRun(state, {
      at: new Date().toISOString(),
      rawFiles,
      runId,
      status: "success",
      warnings,
    }),
  );

  return {
    connectorId: "web-search",
    message: `Fetched Tavily results for ${queries.length} web search quer${
      queries.length === 1 ? "y" : "ies"
    }.`,
    rawFiles,
    runId,
    statePath: `${openWikiConnectorsDisplayPath}/web-search/state.json`,
    status: "success",
    warnings,
  };
}

6. Connector的注册

OpenWiki目前提供的8个Connector都提供了类似于createWebSearchConnector的工厂函数，它们分别是：

  • custom-mcp：createMcpConnector
  • git-repo：createGitRepoConnector
  • google：createGmailConnector
  • hackernews：createHackerNewsConnector
  • langsmith：createLangSmithConnector
  • slack：createSlackConnector
  • x: createXConnector
  • web-search：createWebSearchConnector

至于Notion，由于对应Connector直接连的是Notion发布的MCP服务器，所以直接使用createMcpConnector作为工厂函数。9个Connector通过如下这个createConnectorRegistry函数进行统一注册：

export function createConnectorRegistry(): Record<
  ConnectorId,
  ConnectorRuntime
> {
  return {
    "custom-mcp": createMcpConnector({
      description: `Generic read-only MCP knowledge source. Point OpenWiki at any MCP server via ${openWikiConnectorsDisplayPath}/custom-mcp/config.json (HTTP or stdio transport). Prefer allowedTools and/or MCP readOnlyHint; do not guess mutating tools.`,
      displayName: "Custom MCP",
      id: "custom-mcp",
      requiredEnv: [],
    }),
    "git-repo": createGitRepoConnector(),
    google: createGmailConnector(),
    hackernews: createHackerNewsConnector(),
    langsmith: createLangSmithConnector(),
    notion: createMcpConnector({
      description:
        "Notion connector backed by the hosted Notion MCP server or another configured read-only MCP server.",
      displayName: "Notion",
      id: "notion",
      requiredEnv: ["OPENWIKI_NOTION_MCP_ACCESS_TOKEN"],
    }),
    slack: createSlackConnector(),
    "web-search": createWebSearchConnector(),
    x: createXConnector(),
  };
}

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/JaydenAI/article/details/170411322

文章来源转载

评论

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--