feat: add external knowledge retrieval and cross-project activity
Connect Dify, FastGPT and RAGFlow through read-only bindings, encrypted local credentials, shared retrieval and persistent citations. Configuration and removal affect GoodBuddy only. Show active conversations across projects, retain older live work, and improve project dialog scrolling. Queue concurrent questions and route managed SSH OpenCode questions through the Agent. Preserve recoverable read/search errors and partial search results. Document the Electron UI automation standard, real model and external knowledge validation, and remaining acceptance limits. Validation: earlier full regression passed 4081 tests with 66 skipped; subsequent focused remote-question checks and real Linux evidence are recorded in feature progress. Precommit typecheck and lint passed; latest project dialog and shared UI suites passed 58 tests. The full suite was not rerun for the final dialog changes. Release note: Use external knowledge in conversations, find active work across projects, and answer concurrent local or managed SSH OpenCode questions. Managed SSH questions require the updated Agent adapter.
This commit is contained in:
mesalogo
+220
-3
@@ -1,7 +1,9 @@
|
||||
import { createServer } from 'node:http'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
ExternalKnowledgeClient,
|
||||
ExternalKnowledgeError
|
||||
ExternalKnowledgeError,
|
||||
type ExternalKnowledgeRetrievalConfig
|
||||
} from './external-knowledge-client'
|
||||
|
||||
const jsonResponse = (body: unknown, status = 200): Response =>
|
||||
@@ -52,7 +54,7 @@ describe('ExternalKnowledgeClient', () => {
|
||||
expect(JSON.parse(String(fetcher.mock.calls[1]?.[1]?.body))).toEqual({ query: 'leave policy' })
|
||||
})
|
||||
|
||||
it('filters FastGPT folders and maps searchTest results', async () => {
|
||||
it('retains FastGPT folders for navigation and maps searchTest results', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>()
|
||||
.mockResolvedValueOnce(jsonResponse({
|
||||
code: 200,
|
||||
@@ -83,7 +85,7 @@ describe('ExternalKnowledgeClient', () => {
|
||||
})
|
||||
|
||||
await expect(client.listKnowledgeBases()).resolves.toMatchObject({
|
||||
items: [{ id: 'dataset-1', name: 'Products' }]
|
||||
items: [{ id: 'folder-1', kind: 'folder' }, { id: 'dataset-1', name: 'Products', kind: 'dataset' }]
|
||||
})
|
||||
await expect(client.retrieve('dataset-1', 'pricing', {
|
||||
provider: 'fastgpt',
|
||||
@@ -98,6 +100,221 @@ describe('ExternalKnowledgeClient', () => {
|
||||
})])
|
||||
|
||||
expect(String(fetcher.mock.calls[1]?.[0])).toBe('https://fastgpt.example/api/core/dataset/searchTest')
|
||||
expect(JSON.parse(String(fetcher.mock.calls[0]?.[1]?.body))).toEqual({ parentId: null })
|
||||
})
|
||||
|
||||
it.each(['current', 'legacy'] as const)('preserves typed FastGPT scores and result order in the %s envelope', async (envelope) => {
|
||||
const scores = [{ type: 'fullText', value: 12.4, index: 2 }, { type: 'embedding', value: 0.8, index: 0 }, { type: 'reRank', value: 0.3, index: 1 }]
|
||||
const list = [{ q: 'First', score: scores, id: 'first' }, { q: 'Second', score: 0.99 }]
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse({ code: 200, data: envelope === 'legacy' ? list : { list } }))
|
||||
const client = new ExternalKnowledgeClient({ provider: 'fastgpt', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const results = await client.retrieve('dataset', 'query', { provider: 'fastgpt', searchMode: 'mixedRecall', tokenLimit: 2000, similarity: 0.2, usingRerank: true })
|
||||
expect(results.map((item) => item.snippet)).toEqual(['First', 'Second'])
|
||||
expect(results[0]).toMatchObject({ providerScores: scores, remoteChunkId: 'first' })
|
||||
expect(results[0]?.providerScore).toBeUndefined()
|
||||
expect(results[0]?.remoteDocumentId).toBeUndefined()
|
||||
expect(results[1]?.providerScore).toBe(0.99)
|
||||
expect(JSON.parse(String(fetcher.mock.calls[0]?.[1]?.body))).toEqual({ datasetId: 'dataset', text: 'query', limit: 2000, similarity: 0.2, searchMode: 'mixedRecall', usingReRank: true })
|
||||
})
|
||||
|
||||
it.each([
|
||||
{ code: 200, data: { data: [] } },
|
||||
{ code: 500, data: [] },
|
||||
{ code: 200, data: [{ a: 'Answer without main text' }] },
|
||||
{ code: 200, data: [{ q: 'Text', score: [{ type: 'embedding', value: '0.8' }] }] }
|
||||
])('rejects unsupported FastGPT envelopes and malformed evidence', async (body) => {
|
||||
const client = new ExternalKnowledgeClient({ provider: 'fastgpt', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher: vi.fn<typeof fetch>().mockResolvedValue(jsonResponse(body)) })
|
||||
await expect(client.retrieve('dataset', 'query', { provider: 'fastgpt', searchMode: 'embedding', tokenLimit: 2000, similarity: 0, usingRerank: false })).rejects.toMatchObject({ code: 'EXTERNAL_KB_INVALID_RESPONSE' })
|
||||
})
|
||||
|
||||
it('paginates FastGPT directory results locally and traverses parent IDs', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>().mockImplementation(async () => jsonResponse({ code: 200, data: [
|
||||
{ _id: 'folder', name: 'Folder', type: 'folder' },
|
||||
{ _id: 'one', name: 'One', type: 'dataset' },
|
||||
{ _id: 'two', name: 'Two', type: 'dataset' }
|
||||
] }))
|
||||
const client = new ExternalKnowledgeClient({ provider: 'fastgpt', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
await expect(client.listKnowledgeBases(undefined, { parentId: 'parent', page: 2, pageSize: 2 })).resolves.toMatchObject({ items: [{ id: 'two' }], total: 3, hasMore: false })
|
||||
expect(JSON.parse(String(fetcher.mock.calls[0]?.[1]?.body))).toEqual({ parentId: 'parent' })
|
||||
await expect(client.listKnowledgeBases(undefined, { search: 'one', pageSize: 1 })).resolves.toMatchObject({ items: [{ id: 'one' }], total: 1, hasMore: false })
|
||||
})
|
||||
|
||||
it.each(['dify', 'ragflow'] as const)('requests subsequent %s pages and keeps continuation when local search has no matches', async (provider) => {
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse(provider === 'dify'
|
||||
? { data: [{ id: 'one', name: 'One' }], total: 5, has_more: true }
|
||||
: { code: 0, data: [{ id: 'one', name: 'One' }], total_datasets: 5 }))
|
||||
const client = new ExternalKnowledgeClient({ provider, baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
await expect(client.listKnowledgeBases(undefined, { page: 2, pageSize: 1, search: 'missing' })).resolves.toMatchObject({ items: [], hasMore: true })
|
||||
const url = new URL(String(fetcher.mock.calls[0]?.[0]))
|
||||
expect(url.searchParams.get('page')).toBe('2')
|
||||
expect(url.searchParams.get(provider === 'dify' ? 'limit' : 'page_size')).toBe('1')
|
||||
})
|
||||
|
||||
it.each(['dify', 'fastgpt', 'ragflow'] as const)('reads %s details through its fixed read endpoint', async (provider) => {
|
||||
const record = { id: 'dataset', _id: 'dataset', name: 'Dataset', type: 'dataset', graphrag_task_finish_at: '2026-09-10T10:00:00Z', compilation_template_group_id: 'template' }
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse(provider === 'dify' ? record : { code: provider === 'fastgpt' ? 200 : 0, data: provider === 'fastgpt' ? record : [record] }))
|
||||
const client = new ExternalKnowledgeClient({ provider, baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const detail = await client.getKnowledgeBase('dataset')
|
||||
expect(detail).toMatchObject({ id: 'dataset', name: 'Dataset', kind: 'dataset' })
|
||||
if (provider === 'ragflow') expect(detail).toMatchObject({ graphEnabled: true, knowledgeCompilationEnabled: true })
|
||||
expect(String(fetcher.mock.calls[0]?.[0])).toBe(`https://kb.example${provider === 'dify' ? '/v1/datasets/dataset' : provider === 'fastgpt' ? '/api/core/dataset/detail?id=dataset' : '/api/v1/datasets?id=dataset'}`)
|
||||
expect(fetcher.mock.calls[0]?.[1]?.method).toBe('GET')
|
||||
})
|
||||
|
||||
it('distinguishes missing RAGFlow details and unsupported old Dify details', async () => {
|
||||
const ragflow = new ExternalKnowledgeClient({ provider: 'ragflow', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher: vi.fn<typeof fetch>().mockResolvedValue(jsonResponse({ code: 0, data: [] })) })
|
||||
await expect(ragflow.getKnowledgeBase('missing')).rejects.toMatchObject({ code: 'EXTERNAL_KB_NOT_FOUND' })
|
||||
const dify = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher: vi.fn<typeof fetch>().mockResolvedValue(new Response(null, { status: 405 })) })
|
||||
await expect(dify.getKnowledgeBase('old')).rejects.toMatchObject({ code: 'EXTERNAL_KB_INCOMPATIBLE' })
|
||||
})
|
||||
|
||||
it('sends complete explicit Dify configuration and rejects missing reranking identifiers before fetch', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse({ records: [] }))
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const retrievalModel = { search_method: 'hybrid_search', reranking_enable: true, top_k: 4, score_threshold_enabled: true, score_threshold: 0.2, reranking_mode: 'reranking_model', reranking_model: { reranking_provider_name: 'provider', reranking_model_name: 'model' } } as const
|
||||
await client.retrieve('dataset', 'query', { provider: 'dify', useDatasetDefaults: false, retrievalModel })
|
||||
expect(JSON.parse(String(fetcher.mock.calls[0]?.[1]?.body))).toEqual({ query: 'query', retrieval_model: retrievalModel })
|
||||
await expect(client.retrieve('dataset', 'query', { provider: 'dify', useDatasetDefaults: false, retrievalModel: { ...retrievalModel, reranking_model: undefined } } as ExternalKnowledgeRetrievalConfig)).rejects.toThrow()
|
||||
expect(fetcher).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('does not fetch when already cancelled', async () => {
|
||||
const controller = new AbortController()
|
||||
controller.abort(new Error('private reason'))
|
||||
const fetcher = vi.fn<typeof fetch>()
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const operation = client.listKnowledgeBases(controller.signal)
|
||||
await expect(operation).rejects.toMatchObject({ code: 'EXTERNAL_KB_CANCELLED' })
|
||||
await expect(operation).rejects.not.toHaveProperty('cause')
|
||||
expect(fetcher).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it.each(['headers', 'body'] as const)('distinguishes cancellation and timeout while reading %s', async (phase) => {
|
||||
for (const cancellation of [true, false]) {
|
||||
const controller = new AbortController()
|
||||
const cancelBody = vi.fn()
|
||||
let started!: () => void
|
||||
const ready = new Promise<void>((resolve) => { started = resolve })
|
||||
const fetcher = vi.fn<typeof fetch>().mockImplementation(async (_url, init) => {
|
||||
if (phase === 'headers') return new Promise<Response>((_resolve, reject) => {
|
||||
init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true })
|
||||
started()
|
||||
})
|
||||
return new Response(new ReadableStream({ pull() { started() }, cancel: cancelBody }))
|
||||
})
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher, timeoutMs: cancellation ? 1000 : 20 })
|
||||
const operation = client.listKnowledgeBases(controller.signal)
|
||||
const assertion = expect(operation).rejects.toMatchObject({ code: cancellation ? 'EXTERNAL_KB_CANCELLED' : 'EXTERNAL_KB_TIMEOUT' })
|
||||
await ready
|
||||
if (cancellation) controller.abort()
|
||||
await assertion
|
||||
if (phase === 'body') expect(cancelBody).toHaveBeenCalledTimes(1)
|
||||
}
|
||||
})
|
||||
|
||||
it('maps body transport failures to a sanitized network error', async () => {
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher: vi.fn<typeof fetch>().mockResolvedValue(new Response(new ReadableStream({ pull(controller) { controller.error(new Error('private transport detail')) } }))) })
|
||||
const operation = client.listKnowledgeBases()
|
||||
await expect(operation).rejects.toMatchObject({ code: 'EXTERNAL_KB_NETWORK' })
|
||||
await expect(operation).rejects.not.toHaveProperty('cause')
|
||||
})
|
||||
|
||||
it.each([200, 401])('cancels a response body when cancellation races with HTTP %s headers', async (status) => {
|
||||
const controller = new AbortController()
|
||||
const cancel = vi.fn()
|
||||
const response = new Response(new ReadableStream({ cancel }), { status })
|
||||
const fetcher = vi.fn<typeof fetch>().mockImplementation(async () => {
|
||||
controller.abort()
|
||||
return response
|
||||
})
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
await expect(client.listKnowledgeBases(controller.signal)).rejects.toMatchObject({ code: 'EXTERNAL_KB_CANCELLED' })
|
||||
expect(cancel).toHaveBeenCalledTimes(1)
|
||||
expect(response.body?.locked).toBe(false)
|
||||
})
|
||||
|
||||
it.each(['headers', 'body'] as const)('aborts real HTTP requests stalled at %s on cancellation and timeout', async (phase) => {
|
||||
for (const cancellation of [true, false]) {
|
||||
let started!: () => void
|
||||
const ready = new Promise<void>((resolve) => { started = resolve })
|
||||
const server = createServer((_request, response) => {
|
||||
if (phase === 'body') {
|
||||
response.writeHead(200, { 'Content-Type': 'application/json' })
|
||||
response.write('{"data":[')
|
||||
}
|
||||
started()
|
||||
})
|
||||
try {
|
||||
await new Promise<void>((resolve) => { server.listen(0, '127.0.0.1', resolve) })
|
||||
const address = server.address()
|
||||
if (!address || typeof address === 'string') throw new Error('Missing test server address')
|
||||
const controller = new AbortController()
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: `http://127.0.0.1:${address.port}`, apiKey: 'test-key', timeoutMs: cancellation ? 2000 : 200 })
|
||||
const assertion = expect(client.listKnowledgeBases(controller.signal)).rejects.toMatchObject({ code: cancellation ? 'EXTERNAL_KB_CANCELLED' : 'EXTERNAL_KB_TIMEOUT' })
|
||||
await ready
|
||||
if (cancellation) controller.abort()
|
||||
await assertion
|
||||
} finally {
|
||||
server.closeAllConnections()
|
||||
await new Promise<void>((resolve, reject) => { server.close((error) => error ? reject(error) : resolve()) })
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
it.each(['dify', 'fastgpt', 'ragflow'] as const)('validates %s scalar scores without clamping or treating zero as missing', async (provider) => {
|
||||
for (const score of [0, 12.4, -0.5, undefined, null, '0.8', {}, true]) {
|
||||
const body = provider === 'dify' ? { records: [{ segment: { content: 'Text' }, score }] }
|
||||
: provider === 'fastgpt' ? { code: 200, data: { list: [{ q: 'Text', score }] } }
|
||||
: { code: 0, data: { chunks: [{ content: 'Text', similarity: score }] } }
|
||||
const config: ExternalKnowledgeRetrievalConfig = provider === 'dify' ? { provider, useDatasetDefaults: true }
|
||||
: provider === 'fastgpt' ? { provider, searchMode: 'embedding', tokenLimit: 2000, similarity: 0, usingRerank: false }
|
||||
: { provider, similarityThreshold: 0, vectorSimilarityWeight: 0.3, knnTopK: 128, useKg: false, includeKnowledgeCompilation: false }
|
||||
const client = new ExternalKnowledgeClient({ provider, baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher: vi.fn<typeof fetch>().mockResolvedValue(jsonResponse(body)) })
|
||||
const operation = client.retrieve('dataset', 'query', config)
|
||||
if (score === undefined || score === null || typeof score === 'number') {
|
||||
await expect(operation).resolves.toEqual([expect.objectContaining({ providerScore: score ?? undefined })])
|
||||
} else {
|
||||
await expect(operation).rejects.toMatchObject({ code: 'EXTERNAL_KB_INVALID_RESPONSE' })
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
it('counts Unicode code points for the Dify API query limit', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse({ records: [] }))
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const query = '\u{20000}'.repeat(250)
|
||||
await expect(client.retrieve('dataset', query, { provider: 'dify', useDatasetDefaults: true })).resolves.toEqual([])
|
||||
expect(JSON.parse(String(fetcher.mock.calls[0]?.[1]?.body))).toEqual({ query })
|
||||
await expect(client.retrieve('dataset', `${query}a`, { provider: 'dify', useDatasetDefaults: true })).rejects.toThrow('250 characters')
|
||||
expect(fetcher).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('does not infer RAGFlow capabilities from configuration alone or invalid completion times', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse({ code: 0, data: ['0', '0.0', '', 'invalid'].map((time, index) => ({ id: String(index), name: 'Dataset', graphrag_task_finish_at: time, parser_config: { graphrag: { use_graphrag: true } } })) }))
|
||||
const client = new ExternalKnowledgeClient({ provider: 'ragflow', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const page = await client.listKnowledgeBases()
|
||||
expect(page.items).toHaveLength(4)
|
||||
for (const item of page.items) expect(item).toMatchObject({ graphEnabled: false, knowledgeCompilationEnabled: false })
|
||||
})
|
||||
|
||||
it('sends Dify weighted configuration without merging and rejects inconsistent weights', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>().mockResolvedValue(jsonResponse({ records: [] }))
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', fetcher })
|
||||
const retrievalModel = { search_method: 'hybrid_search', reranking_enable: true, top_k: 4, score_threshold_enabled: false, reranking_mode: 'weighted_score', weights: { weight_type: 'customized', vector_setting: { vector_weight: 0.7, embedding_provider_name: 'provider', embedding_model_name: 'model' }, keyword_setting: { keyword_weight: 0.3 } } } as const
|
||||
await client.retrieve('dataset', 'query', { provider: 'dify', useDatasetDefaults: false, retrievalModel })
|
||||
expect(JSON.parse(String(fetcher.mock.calls[0]?.[1]?.body))).toEqual({ query: 'query', retrieval_model: retrievalModel })
|
||||
await expect(client.retrieve('dataset', 'query', { provider: 'dify', useDatasetDefaults: false, retrievalModel: { ...retrievalModel, weights: { ...retrievalModel.weights, keyword_setting: { keyword_weight: 0.7 } } } })).rejects.toThrow('weights must sum to one')
|
||||
await expect(client.retrieve('dataset', 'q'.repeat(251), { provider: 'dify', useDatasetDefaults: true })).rejects.toThrow('250 characters')
|
||||
expect(fetcher).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it.each([true, false])('cancels oversized response streams (declared length: %s)', async (declared) => {
|
||||
const cancel = vi.fn()
|
||||
const response = new Response(new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode('123456')) }, cancel }), { headers: declared ? { 'content-length': '6' } : {} })
|
||||
const client = new ExternalKnowledgeClient({ provider: 'dify', baseUrl: 'https://kb.example', apiKey: 'test-key', maximumResponseBytes: 5, fetcher: vi.fn<typeof fetch>().mockResolvedValue(response) })
|
||||
await expect(client.listKnowledgeBases()).rejects.toMatchObject({ code: 'EXTERNAL_KB_INVALID_RESPONSE' })
|
||||
expect(cancel).toHaveBeenCalledTimes(1)
|
||||
expect(response.body?.locked).toBe(false)
|
||||
})
|
||||
|
||||
it('maps RAGFlow graph capability and retrieval fields', async () => {
|
||||
|
||||
+144
-92
@@ -1,44 +1,21 @@
|
||||
export type ExternalKnowledgeProvider = 'dify' | 'fastgpt' | 'ragflow'
|
||||
import {
|
||||
externalKnowledgeProviderConfigSchema,
|
||||
type ExternalKnowledgeProvider,
|
||||
type ExternalKnowledgeCatalogItem,
|
||||
type ExternalKnowledgeCatalogPage,
|
||||
type ExternalKnowledgeProviderConfig,
|
||||
type ExternalKnowledgeRemoteResult
|
||||
} from '../../../shared/external-knowledge-contracts'
|
||||
|
||||
export type ExternalKnowledgeCatalogItem = {
|
||||
id: string
|
||||
name: string
|
||||
description?: string
|
||||
graphEnabled?: boolean
|
||||
}
|
||||
export type { ExternalKnowledgeProvider, ExternalKnowledgeCatalogItem, ExternalKnowledgeCatalogPage }
|
||||
export type ExternalKnowledgeRetrievalConfig = ExternalKnowledgeProviderConfig
|
||||
export type ExternalKnowledgeRetrievalResult = ExternalKnowledgeRemoteResult
|
||||
|
||||
export type ExternalKnowledgeCatalogPage = {
|
||||
items: ExternalKnowledgeCatalogItem[]
|
||||
total?: number
|
||||
hasMore: boolean
|
||||
}
|
||||
|
||||
export type ExternalKnowledgeRetrievalConfig =
|
||||
| { provider: 'dify'; useDatasetDefaults: true }
|
||||
| {
|
||||
provider: 'fastgpt'
|
||||
searchMode: 'embedding' | 'fullTextRecall' | 'mixedRecall'
|
||||
tokenLimit: number
|
||||
similarity: number
|
||||
usingRerank: boolean
|
||||
}
|
||||
| {
|
||||
provider: 'ragflow'
|
||||
similarityThreshold: number
|
||||
vectorSimilarityWeight: number
|
||||
knnTopK: number
|
||||
useKg: boolean
|
||||
includeKnowledgeCompilation: boolean
|
||||
}
|
||||
|
||||
export type ExternalKnowledgeRetrievalResult = {
|
||||
documentTitle: string
|
||||
sourceDisplayName: string
|
||||
snippet: string
|
||||
providerScore?: number
|
||||
remoteDocumentId?: string
|
||||
remoteChunkId?: string
|
||||
location?: string
|
||||
export type ExternalKnowledgeCatalogInput = {
|
||||
page?: number
|
||||
pageSize?: number
|
||||
parentId?: string | null
|
||||
search?: string
|
||||
}
|
||||
|
||||
export class ExternalKnowledgeError extends Error {
|
||||
@@ -46,6 +23,8 @@ export class ExternalKnowledgeError extends Error {
|
||||
readonly code:
|
||||
| 'EXTERNAL_KB_NETWORK'
|
||||
| 'EXTERNAL_KB_TIMEOUT'
|
||||
| 'EXTERNAL_KB_CANCELLED'
|
||||
| 'EXTERNAL_KB_INCOMPATIBLE'
|
||||
| 'EXTERNAL_KB_AUTH'
|
||||
| 'EXTERNAL_KB_FORBIDDEN'
|
||||
| 'EXTERNAL_KB_NOT_FOUND'
|
||||
@@ -60,7 +39,7 @@ export class ExternalKnowledgeError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
type ExternalKnowledgeClientOptions = {
|
||||
export type ExternalKnowledgeClientOptions = {
|
||||
provider: ExternalKnowledgeProvider
|
||||
baseUrl: string
|
||||
apiKey: string
|
||||
@@ -109,6 +88,7 @@ function endpoint(baseUrl: URL, path: string): URL {
|
||||
}
|
||||
|
||||
function errorForStatus(status: number): ExternalKnowledgeError {
|
||||
if (status === 405) return new ExternalKnowledgeError('EXTERNAL_KB_INCOMPATIBLE', 'External knowledge operation is not supported by this deployment')
|
||||
if (status === 401) return new ExternalKnowledgeError('EXTERNAL_KB_AUTH', 'External knowledge authentication failed')
|
||||
if (status === 403) return new ExternalKnowledgeError('EXTERNAL_KB_FORBIDDEN', 'External knowledge access is forbidden')
|
||||
if (status === 404) return new ExternalKnowledgeError('EXTERNAL_KB_NOT_FOUND', 'External knowledge target was not found')
|
||||
@@ -116,36 +96,42 @@ function errorForStatus(status: number): ExternalKnowledgeError {
|
||||
return new ExternalKnowledgeError('EXTERNAL_KB_SERVER', `External knowledge service returned HTTP ${status}`)
|
||||
}
|
||||
|
||||
async function readBoundedJson(response: Response, maximumBytes: number): Promise<unknown> {
|
||||
async function readBoundedJson(response: Response, maximumBytes: number, signal: AbortSignal): Promise<unknown> {
|
||||
const declaredLength = Number(response.headers.get('content-length'))
|
||||
if (Number.isFinite(declaredLength) && declaredLength > maximumBytes) {
|
||||
void response.body?.cancel().catch(() => {})
|
||||
throw new ExternalKnowledgeError('EXTERNAL_KB_INVALID_RESPONSE', 'External knowledge response is too large')
|
||||
}
|
||||
if (!response.body) {
|
||||
throw new ExternalKnowledgeError('EXTERNAL_KB_INVALID_RESPONSE', 'External knowledge response is empty')
|
||||
}
|
||||
const reader = response.body.getReader()
|
||||
const cancel = (): void => { void reader.cancel().catch(() => {}) }
|
||||
signal.addEventListener('abort', cancel, { once: true })
|
||||
const chunks: Uint8Array[] = []
|
||||
let total = 0
|
||||
try {
|
||||
signal.throwIfAborted()
|
||||
while (true) {
|
||||
const part = await reader.read()
|
||||
signal.throwIfAborted()
|
||||
if (part.done) break
|
||||
total += part.value.byteLength
|
||||
if (total > maximumBytes) {
|
||||
await reader.cancel()
|
||||
cancel()
|
||||
throw new ExternalKnowledgeError('EXTERNAL_KB_INVALID_RESPONSE', 'External knowledge response is too large')
|
||||
}
|
||||
chunks.push(part.value)
|
||||
}
|
||||
} finally {
|
||||
signal.removeEventListener('abort', cancel)
|
||||
reader.releaseLock()
|
||||
}
|
||||
const body = Buffer.concat(chunks.map((chunk) => Buffer.from(chunk))).toString('utf8')
|
||||
try {
|
||||
return JSON.parse(body) as unknown
|
||||
} catch (error) {
|
||||
throw new ExternalKnowledgeError('EXTERNAL_KB_INVALID_RESPONSE', 'External knowledge response is not valid JSON', { cause: error })
|
||||
} catch {
|
||||
throw new ExternalKnowledgeError('EXTERNAL_KB_INVALID_RESPONSE', 'External knowledge response is not valid JSON')
|
||||
}
|
||||
}
|
||||
|
||||
@@ -169,16 +155,23 @@ export class ExternalKnowledgeClient {
|
||||
this.maximumResponseBytes = options.maximumResponseBytes ?? 5_000_000
|
||||
}
|
||||
|
||||
async listKnowledgeBases(signal?: AbortSignal): Promise<ExternalKnowledgeCatalogPage> {
|
||||
async listKnowledgeBases(signal?: AbortSignal, input: ExternalKnowledgeCatalogInput = {}): Promise<ExternalKnowledgeCatalogPage> {
|
||||
const { page = 1, pageSize = 100, parentId = null } = input
|
||||
if (!Number.isInteger(page) || page < 1 || page > 500 || !Number.isInteger(pageSize) || pageSize < 1 || pageSize > 100) {
|
||||
throw new RangeError('Invalid external knowledge catalog pagination')
|
||||
}
|
||||
const search = input.search?.trim().toLowerCase()
|
||||
const filter = (items: ExternalKnowledgeCatalogItem[]): ExternalKnowledgeCatalogItem[] =>
|
||||
search ? items.filter((item) => `${item.name}\n${item.description ?? ''}`.toLowerCase().includes(search)) : items
|
||||
if (this.provider === 'dify') {
|
||||
const url = endpoint(this.normalizedBaseUrl, '/v1/datasets')
|
||||
url.searchParams.set('page', '1')
|
||||
url.searchParams.set('limit', '100')
|
||||
url.searchParams.set('page', String(page))
|
||||
url.searchParams.set('limit', String(pageSize))
|
||||
const body = asRecord(await this.request(url, { method: 'GET' }, signal))
|
||||
const items = Array.isArray(body?.data) ? body.data : undefined
|
||||
if (!items) throw this.invalidResponse()
|
||||
return {
|
||||
items: items.slice(0, 100).map((item) => this.catalogItem(item)),
|
||||
items: filter(items.slice(0, pageSize).map((item) => this.catalogItem(item))),
|
||||
total: finiteNumber(body?.total),
|
||||
hasMore: body?.has_more === true
|
||||
}
|
||||
@@ -187,63 +180,81 @@ export class ExternalKnowledgeClient {
|
||||
if (this.provider === 'fastgpt') {
|
||||
const body = asRecord(await this.request(
|
||||
endpoint(this.normalizedBaseUrl, '/core/dataset/list'),
|
||||
{ method: 'POST', body: JSON.stringify({ pageNum: 1, pageSize: 100 }) },
|
||||
{ method: 'POST', body: JSON.stringify({ parentId }) },
|
||||
signal
|
||||
))
|
||||
const data = asRecord(body?.data)
|
||||
const items = Array.isArray(body?.data)
|
||||
? body.data
|
||||
: Array.isArray(data?.data)
|
||||
? data.data
|
||||
: Array.isArray(data?.list)
|
||||
? data.list
|
||||
: undefined
|
||||
const items = Array.isArray(body?.data) ? body.data : undefined
|
||||
if (body?.code !== 200 || !items) throw this.invalidResponse()
|
||||
const datasets = items.filter((item) => asRecord(item)?.type === 'dataset')
|
||||
const datasets = filter(items.filter((item) => ['dataset', 'folder'].includes(String(asRecord(item)?.type)))
|
||||
.map((item) => this.catalogItem(item, '_id', 'intro')))
|
||||
const offset = (page - 1) * pageSize
|
||||
return {
|
||||
items: datasets.slice(0, 100).map((item) => this.catalogItem(item, '_id', 'intro')),
|
||||
total: finiteNumber(data?.total) ?? datasets.length,
|
||||
hasMore: datasets.length > 100
|
||||
items: datasets.slice(offset, offset + pageSize),
|
||||
total: datasets.length,
|
||||
hasMore: datasets.length > offset + pageSize
|
||||
}
|
||||
}
|
||||
|
||||
const url = endpoint(this.normalizedBaseUrl, '/api/v1/datasets')
|
||||
url.searchParams.set('page', '1')
|
||||
url.searchParams.set('page_size', '100')
|
||||
url.searchParams.set('page', String(page))
|
||||
url.searchParams.set('page_size', String(pageSize))
|
||||
const body = asRecord(await this.request(url, { method: 'GET' }, signal))
|
||||
if (body?.code !== 0 || !Array.isArray(body.data)) throw this.invalidResponse()
|
||||
return {
|
||||
items: body.data.slice(0, 100).map((item) => {
|
||||
const record = asRecord(item)
|
||||
const graphFinishedAt = record?.graphrag_task_finish_at
|
||||
return {
|
||||
...this.catalogItem(item),
|
||||
graphEnabled:
|
||||
(typeof graphFinishedAt === 'number' && graphFinishedAt > 0) ||
|
||||
(typeof graphFinishedAt === 'string' && graphFinishedAt.length > 0)
|
||||
}
|
||||
}),
|
||||
items: filter(body.data.slice(0, pageSize).map((item) => this.catalogItem(item))),
|
||||
total: finiteNumber(body.total_datasets),
|
||||
hasMore: (finiteNumber(body.total_datasets) ?? 0) > 100
|
||||
hasMore: finiteNumber(body.total_datasets) === undefined
|
||||
? body.data.length >= pageSize
|
||||
: Number(body.total_datasets) > page * pageSize
|
||||
}
|
||||
}
|
||||
|
||||
async getKnowledgeBase(remoteKnowledgeBaseId: string, signal?: AbortSignal): Promise<ExternalKnowledgeCatalogItem> {
|
||||
const id = remoteKnowledgeBaseId.trim()
|
||||
if (!id) throw new TypeError('Remote knowledge base ID is required')
|
||||
const url = endpoint(this.normalizedBaseUrl, this.provider === 'dify'
|
||||
? `/v1/datasets/${encodeURIComponent(id)}`
|
||||
: this.provider === 'fastgpt' ? '/core/dataset/detail' : '/api/v1/datasets')
|
||||
if (this.provider !== 'dify') url.searchParams.set('id', id)
|
||||
const body = asRecord(await this.request(url, { method: 'GET' }, signal))
|
||||
let item: ExternalKnowledgeCatalogItem
|
||||
if (this.provider === 'dify') item = this.catalogItem(body)
|
||||
else if (this.provider === 'fastgpt') {
|
||||
if (body?.code !== 200) throw this.invalidResponse()
|
||||
item = this.catalogItem(body.data, '_id', 'intro')
|
||||
} else {
|
||||
if (body?.code !== 0 || !Array.isArray(body.data)) throw this.invalidResponse()
|
||||
const match = body.data.find((value) => asRecord(value)?.id === id)
|
||||
if (!match) throw errorForStatus(404)
|
||||
item = this.catalogItem(match)
|
||||
}
|
||||
if (item.id !== id) throw this.invalidResponse()
|
||||
return item
|
||||
}
|
||||
|
||||
async retrieve(
|
||||
remoteKnowledgeBaseId: string,
|
||||
query: string,
|
||||
config: ExternalKnowledgeRetrievalConfig,
|
||||
signal?: AbortSignal
|
||||
): Promise<ExternalKnowledgeRetrievalResult[]> {
|
||||
config = externalKnowledgeProviderConfigSchema.parse(config)
|
||||
if (config.provider !== this.provider) throw new TypeError('Provider configuration does not match client')
|
||||
const remoteId = remoteKnowledgeBaseId.trim()
|
||||
const question = query.trim()
|
||||
if (!remoteId || !question) throw new TypeError('Remote knowledge base ID and query are required')
|
||||
|
||||
if (this.provider === 'dify') {
|
||||
if (question.length > 250) throw new RangeError('Dify queries cannot exceed 250 characters')
|
||||
if (this.provider === 'dify' && config.provider === 'dify') {
|
||||
if (!config.useDatasetDefaults && config.retrievalModel.reranking_mode === 'weighted_score' && config.retrievalModel.weights) {
|
||||
const weights = config.retrievalModel.weights
|
||||
if (Math.abs(weights.vector_setting.vector_weight + weights.keyword_setting.keyword_weight - 1) > 1e-9) {
|
||||
throw new TypeError('Dify retrieval weights must sum to one')
|
||||
}
|
||||
}
|
||||
if (Array.from(question).length > 250) throw new RangeError('Dify queries cannot exceed 250 characters')
|
||||
const body = asRecord(await this.request(
|
||||
endpoint(this.normalizedBaseUrl, `/v1/datasets/${encodeURIComponent(remoteId)}/retrieve`),
|
||||
{ method: 'POST', body: JSON.stringify({ query: question }) },
|
||||
{ method: 'POST', body: JSON.stringify({ query: question, ...(!config.useDatasetDefaults ? { retrieval_model: config.retrievalModel } : {}) }) },
|
||||
signal
|
||||
))
|
||||
if (!Array.isArray(body?.records)) throw this.invalidResponse()
|
||||
@@ -267,8 +278,10 @@ export class ExternalKnowledgeClient {
|
||||
signal
|
||||
))
|
||||
const data = asRecord(body?.data)
|
||||
if (body?.code !== 200 || !Array.isArray(data?.list)) throw this.invalidResponse()
|
||||
return data.list.slice(0, 20).map((item) => this.fastGptResult(item))
|
||||
// The official legacy manual documents data[]; current OpenAPI uses data.list[].
|
||||
const items = Array.isArray(body?.data) ? body.data : data?.list
|
||||
if (body?.code !== 200 || !Array.isArray(items)) throw this.invalidResponse()
|
||||
return items.slice(0, 20).map((item) => this.fastGptResult(item))
|
||||
}
|
||||
|
||||
if (this.provider === 'ragflow' && config.provider === 'ragflow') {
|
||||
@@ -299,12 +312,13 @@ export class ExternalKnowledgeClient {
|
||||
}
|
||||
|
||||
private async request(url: URL, init: RequestInit, signal?: AbortSignal): Promise<unknown> {
|
||||
const timeoutSignal = AbortSignal.timeout(this.timeoutMs)
|
||||
const effectiveSignal = signal
|
||||
? AbortSignal.any([signal, AbortSignal.timeout(this.timeoutMs)])
|
||||
: AbortSignal.timeout(this.timeoutMs)
|
||||
let response: Response
|
||||
? AbortSignal.any([signal, timeoutSignal])
|
||||
: timeoutSignal
|
||||
try {
|
||||
response = await this.fetcher(url, {
|
||||
effectiveSignal.throwIfAborted()
|
||||
const response = await this.fetcher(url, {
|
||||
...init,
|
||||
redirect: 'error',
|
||||
signal: effectiveSignal,
|
||||
@@ -314,20 +328,40 @@ export class ExternalKnowledgeClient {
|
||||
...(init.body ? { 'Content-Type': 'application/json' } : {})
|
||||
}
|
||||
})
|
||||
if (effectiveSignal.aborted || !response.ok) {
|
||||
void response.body?.cancel().catch(() => {})
|
||||
effectiveSignal.throwIfAborted()
|
||||
throw errorForStatus(response.status)
|
||||
}
|
||||
return await readBoundedJson(response, this.maximumResponseBytes, effectiveSignal)
|
||||
} catch (error) {
|
||||
const code = effectiveSignal.aborted ? 'EXTERNAL_KB_TIMEOUT' : 'EXTERNAL_KB_NETWORK'
|
||||
throw new ExternalKnowledgeError(code, code === 'EXTERNAL_KB_TIMEOUT' ? 'External knowledge request timed out' : 'External knowledge service is unreachable', { cause: error })
|
||||
if (effectiveSignal.aborted) {
|
||||
const timedOut = effectiveSignal.reason === timeoutSignal.reason && timeoutSignal.aborted
|
||||
throw new ExternalKnowledgeError(timedOut ? 'EXTERNAL_KB_TIMEOUT' : 'EXTERNAL_KB_CANCELLED',
|
||||
timedOut ? 'External knowledge request timed out' : 'External knowledge request was cancelled')
|
||||
}
|
||||
if (error instanceof ExternalKnowledgeError) throw error
|
||||
throw new ExternalKnowledgeError('EXTERNAL_KB_NETWORK', 'External knowledge service is unreachable')
|
||||
}
|
||||
if (!response.ok) throw errorForStatus(response.status)
|
||||
return readBoundedJson(response, this.maximumResponseBytes)
|
||||
}
|
||||
|
||||
private catalogItem(value: unknown, idKey = 'id', descriptionKey = 'description'): ExternalKnowledgeCatalogItem {
|
||||
const record = asRecord(value)
|
||||
const id = boundedString(record?.[idKey], 128)
|
||||
const id = boundedString(record?.[idKey], 512)
|
||||
const name = boundedString(record?.name, 512)
|
||||
if (!id || !name) throw this.invalidResponse()
|
||||
return { id, name, description: boundedString(record?.[descriptionKey], 2_000) }
|
||||
const item: ExternalKnowledgeCatalogItem = { id, name, description: boundedString(record?.[descriptionKey], 2_000), kind: 'dataset' }
|
||||
if (this.provider === 'fastgpt') {
|
||||
if (record?.type !== 'dataset' && record?.type !== 'folder') throw this.invalidResponse()
|
||||
item.kind = record.type
|
||||
}
|
||||
if (this.provider === 'ragflow') {
|
||||
const finished = record?.graphrag_task_finish_at
|
||||
item.graphEnabled = typeof finished === 'number' ? Number.isFinite(finished) && finished > 0
|
||||
: typeof finished === 'string' && (Number.isFinite(Number(finished)) ? Number(finished) > 0 : Date.parse(finished) > 0)
|
||||
item.knowledgeCompilationEnabled = !!boundedString(record?.compilation_template_group_id, 512)
|
||||
}
|
||||
return item
|
||||
}
|
||||
|
||||
private difyResult(value: unknown): ExternalKnowledgeRetrievalResult {
|
||||
@@ -341,7 +375,7 @@ export class ExternalKnowledgeClient {
|
||||
documentTitle: title,
|
||||
sourceDisplayName: title,
|
||||
snippet,
|
||||
providerScore: finiteNumber(record?.score),
|
||||
providerScore: this.resultScore(record?.score),
|
||||
remoteDocumentId: boundedString(document?.id, 128),
|
||||
remoteChunkId: boundedString(segment?.id, 128),
|
||||
location: finiteNumber(segment?.position) === undefined ? undefined : `segment ${segment?.position}`
|
||||
@@ -351,15 +385,26 @@ export class ExternalKnowledgeClient {
|
||||
private fastGptResult(value: unknown): ExternalKnowledgeRetrievalResult {
|
||||
const record = asRecord(value)
|
||||
const question = boundedString(record?.q, 8_000)
|
||||
if (!question) throw this.invalidResponse()
|
||||
const answer = boundedString(record?.a, 8_000)
|
||||
const snippet = [question, answer].filter(Boolean).join('\n')
|
||||
if (!snippet) throw this.invalidResponse()
|
||||
const title = boundedString(record?.sourceName, 512) ?? 'FastGPT document'
|
||||
// Preserve channel semantics and provider order; these are not one comparable total score.
|
||||
const providerScores = Array.isArray(record?.score) ? record.score.slice(0, 20).map((value) => {
|
||||
const score = asRecord(value)
|
||||
const type = boundedString(score?.type, 128)
|
||||
const number = finiteNumber(score?.value)
|
||||
const index = finiteNumber(score?.index)
|
||||
if (!type || number === undefined || (score?.index !== undefined && (index === undefined || !Number.isInteger(index)))) throw this.invalidResponse()
|
||||
return { type, value: number, ...(index === undefined ? {} : { index }) }
|
||||
}) : undefined
|
||||
return {
|
||||
documentTitle: title,
|
||||
sourceDisplayName: title,
|
||||
snippet: snippet.slice(0, 8_000),
|
||||
providerScore: finiteNumber(record?.score),
|
||||
providerScore: providerScores ? undefined : this.resultScore(record?.score),
|
||||
...(providerScores ? { providerScores } : {}),
|
||||
remoteDocumentId: boundedString(record?.sourceId, 128),
|
||||
remoteChunkId: boundedString(record?.id, 128),
|
||||
location: finiteNumber(record?.chunkIndex) === undefined ? undefined : `chunk ${record?.chunkIndex}`
|
||||
@@ -376,13 +421,20 @@ export class ExternalKnowledgeClient {
|
||||
documentTitle: title,
|
||||
sourceDisplayName: title,
|
||||
snippet,
|
||||
providerScore: finiteNumber(record?.similarity),
|
||||
providerScore: this.resultScore(record?.similarity),
|
||||
remoteDocumentId: boundedString(record?.document_id, 128),
|
||||
remoteChunkId: boundedString(record?.id, 128),
|
||||
location: positions ? JSON.stringify(positions).slice(0, 8_192) : undefined
|
||||
}
|
||||
}
|
||||
|
||||
private resultScore(value: unknown): number | undefined {
|
||||
if (value === undefined || value === null) return undefined
|
||||
const score = finiteNumber(value)
|
||||
if (score === undefined) throw this.invalidResponse()
|
||||
return score
|
||||
}
|
||||
|
||||
private invalidResponse(): ExternalKnowledgeError {
|
||||
return new ExternalKnowledgeError('EXTERNAL_KB_INVALID_RESPONSE', 'External knowledge response does not match the supported API')
|
||||
}
|
||||
|
||||
@@ -0,0 +1,230 @@
|
||||
import { createServer, type Server } from 'node:http'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { DatabaseSync } from 'node:sqlite'
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { KnowledgeService } from '../knowledge-service'
|
||||
import { KnowledgeMcpGateway } from '../../agent/knowledge-mcp-gateway'
|
||||
import { knowledgeReferenceKey, toKnowledgeReference } from '../../../shared/knowledge-reference'
|
||||
import { conversationMessageSchema } from '../../../shared/assistant-contracts'
|
||||
|
||||
const services: KnowledgeService[] = []
|
||||
const directories: string[] = []
|
||||
const servers: Server[] = []
|
||||
afterEach(async () => {
|
||||
await Promise.all(services.splice(0).map(service => service.dispose()))
|
||||
await Promise.all(servers.splice(0).map(server => new Promise<void>(resolve => server.close(() => resolve()))))
|
||||
await Promise.all(directories.splice(0).map(directory => rm(directory, { recursive: true, force: true })))
|
||||
})
|
||||
|
||||
async function setup(fetcher?: typeof fetch) {
|
||||
const directory = await mkdtemp(join(tmpdir(), 'external-kb-'))
|
||||
directories.push(directory)
|
||||
const path = join(directory, 'knowledge.db')
|
||||
const cipher = { isAvailable: () => true, encrypt: (value: string) => Buffer.from(value), decrypt: (value: Buffer) => value.toString() }
|
||||
const options = { databasePath: path, managedRoot: join(directory, 'managed'), externalFetcher: fetcher, credentialCipher: cipher }
|
||||
const service = new KnowledgeService(options)
|
||||
await service.initialize()
|
||||
services.push(service)
|
||||
return { service, path, cipher, options }
|
||||
}
|
||||
|
||||
const commonConfig = { resultLimit: 20, requestTimeoutMs: 1000, maxSnippetCharacters: 8000 }
|
||||
const providerConfig = { provider: 'dify' as const, useDatasetDefaults: true as const }
|
||||
const records = (count = 1, size = 16) => ({ records: Array.from({ length: count }, (_, index) => ({
|
||||
score: 0.8, segment: { id: `chunk-${index}`, content: 'e'.repeat(size), document: { id: 'remote-doc', name: 'Handbook' } }
|
||||
})) })
|
||||
|
||||
function instance(service: KnowledgeService, baseUrl = 'https://kb.example/v1') {
|
||||
return service.external.saveInstance({ name: 'Dify', provider: 'dify', baseUrl, enabled: true, credential: { action: 'replace', value: 'test-key' } })
|
||||
}
|
||||
|
||||
function bindingInput(instanceId: string, remoteKnowledgeBaseId = 'dataset') {
|
||||
return { instanceId, remoteKnowledgeBaseId, name: 'Handbook', remoteName: 'Remote handbook', testQuery: 'policy', commonConfig, providerConfig }
|
||||
}
|
||||
|
||||
function retrievalInput(instanceId: string) {
|
||||
return { instanceId, remoteKnowledgeBaseId: 'dataset', testQuery: 'policy', commonConfig, providerConfig }
|
||||
}
|
||||
|
||||
it('routes the real HTTP client through directory, detail, binding and scoped MCP without remote writes', async () => {
|
||||
const requests: string[] = []
|
||||
const server = createServer((request, response) => {
|
||||
requests.push(`${request.method} ${request.url}`)
|
||||
response.setHeader('content-type', 'application/json')
|
||||
response.end(JSON.stringify(request.url?.includes('/retrieve') ? records() : request.url?.includes('?')
|
||||
? { data: [{ id: 'dataset', name: 'Handbook' }], has_more: false }
|
||||
: { id: 'dataset', name: 'Handbook' }))
|
||||
})
|
||||
servers.push(server)
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address() as { port: number }
|
||||
const { service } = await setup()
|
||||
const saved = instance(service, `http://127.0.0.1:${address.port}/v1`)
|
||||
expect(saved).not.toHaveProperty('credential')
|
||||
expect((await service.external.listCatalog({ instanceId: saved.id })).items).toHaveLength(1)
|
||||
expect(await service.external.getCatalog({ instanceId: saved.id, remoteKnowledgeBaseId: 'dataset' })).toMatchObject({ id: 'dataset' })
|
||||
const binding = await service.external.saveBinding(bindingInput(saved.id))
|
||||
const gateway = new KnowledgeMcpGateway(service)
|
||||
const token = gateway.grant('request', [binding.knowledgeBaseId], new AbortController().signal)!
|
||||
try {
|
||||
const references = await gateway.search(token, { query: 'policy' })
|
||||
expect(references).toHaveLength(1)
|
||||
expect(references[0]).toMatchObject({ external: { remoteDocumentId: 'remote-doc', remoteChunkId: 'chunk-0', providerScore: 0.8 } })
|
||||
expect(references[0]?.documentId).toBeUndefined()
|
||||
expect(service.database.listDocuments(binding.knowledgeBaseId)).toEqual([])
|
||||
expect(requests).toEqual(['GET /v1/datasets?page=1&limit=100', 'GET /v1/datasets/dataset', 'POST /v1/datasets/dataset/retrieve', 'POST /v1/datasets/dataset/retrieve'])
|
||||
} finally { await gateway.dispose() }
|
||||
})
|
||||
|
||||
it('keeps actual failures, bounds mixed retrieval to 48k, and rejects local mutations', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>(async () => Response.json(records(20, 8000)))
|
||||
const { service } = await setup(fetcher)
|
||||
const saved = instance(service)
|
||||
const first = await service.external.saveBinding(bindingInput(saved.id))
|
||||
const second = await service.external.saveBinding(bindingInput(saved.id, 'second'))
|
||||
const results = await service.retrieveMany([first.knowledgeBaseId, second.knowledgeBaseId], 'policy')
|
||||
expect(results.flatMap(item => item.response.results).reduce((sum, result) => sum + result.snippet.length, 0)).toBe(48000)
|
||||
fetcher.mockResolvedValue(new Response('', { status: 401 }))
|
||||
const failed = await service.retrieveMany([first.knowledgeBaseId], 'policy')
|
||||
expect(failed[0]?.response.diagnostics.failure).toBe('EXTERNAL_KB_AUTH')
|
||||
await expect(service.importPaths(first.knowledgeBaseId, [])).rejects.toThrow('EXTERNAL_KB_READ_ONLY')
|
||||
await expect(service.rebuildLibrary({ knowledgeBaseId: first.knowledgeBaseId })).rejects.toThrow('EXTERNAL_KB_READ_ONLY')
|
||||
expect(() => service.updateSettings({ knowledgeBaseId: first.knowledgeBaseId, retrieval: service.database.getKnowledgeBase(first.knowledgeBaseId)!.retrievalSettings })).toThrow('EXTERNAL_KB_READ_ONLY')
|
||||
expect(() => service.database.updateKnowledgeBase(first.knowledgeBaseId, { graphEnabled: true })).toThrow('EXTERNAL_KB_READ_ONLY')
|
||||
})
|
||||
|
||||
it('rolls metadata back with binding failures and rejects concurrent duplicate creates', async () => {
|
||||
const { service, path } = await setup(async () => Response.json(records()))
|
||||
const saved = instance(service)
|
||||
const first = await service.external.saveBinding(bindingInput(saved.id))
|
||||
const inspection = new DatabaseSync(path)
|
||||
inspection.exec("CREATE TRIGGER reject_binding BEFORE UPDATE ON external_knowledge_bindings BEGIN SELECT RAISE(ABORT, 'binding failed'); END")
|
||||
try {
|
||||
await expect(service.external.saveBinding({ ...bindingInput(saved.id), knowledgeBaseId: first.knowledgeBaseId, name: 'Changed' })).rejects.toThrow('binding failed')
|
||||
expect(service.database.getKnowledgeBase(first.knowledgeBaseId)?.name).toBe('Handbook')
|
||||
expect(inspection.prepare('PRAGMA user_version').get()).toEqual({ user_version: 12 })
|
||||
} finally { inspection.close() }
|
||||
const outcomes = await Promise.allSettled([service.external.saveBinding(bindingInput(saved.id, 'new')), service.external.saveBinding(bindingInput(saved.id, 'new'))])
|
||||
expect(outcomes.filter(item => item.status === 'fulfilled')).toHaveLength(1)
|
||||
expect(service.database.listKnowledgeBases()).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('cancels pending probes on disable and prevents shutdown writeback', async () => {
|
||||
let release!: (value: Response) => void
|
||||
const fetcher = vi.fn<typeof fetch>(() => new Promise<Response>(resolve => { release = resolve }))
|
||||
const { service } = await setup(fetcher)
|
||||
const saved = instance(service)
|
||||
const probe = service.external.testInstance(saved.id)
|
||||
service.external.setEnabled(saved.id, false)
|
||||
release(Response.json({ data: [], has_more: false }))
|
||||
await expect(probe).rejects.toThrow('EXTERNAL_KB_CONFIG_CHANGED')
|
||||
expect(service.external.listInstances()[0]?.enabled).toBe(false)
|
||||
service.external.setEnabled(saved.id, true)
|
||||
const pending = service.external.saveBinding(bindingInput(saved.id))
|
||||
const rejected = expect(pending).rejects.toThrow()
|
||||
await service.dispose()
|
||||
services.splice(services.indexOf(service), 1)
|
||||
release(Response.json(records()))
|
||||
await rejected
|
||||
expect(() => service.external.listInstances()).toThrow('EXTERNAL_KB_CANCELLED')
|
||||
})
|
||||
|
||||
it('persists external locators without local IDs and distinguishes remote chunks', async () => {
|
||||
const { service } = await setup(async () => Response.json(records(2)))
|
||||
const binding = await service.external.saveBinding(bindingInput(instance(service).id))
|
||||
const response = await service.retrieve({ knowledgeBaseId: binding.knowledgeBaseId, query: 'policy' })
|
||||
const references = response.results.map(result => toKnowledgeReference(result, 'Handbook'))
|
||||
expect(knowledgeReferenceKey(references[0]!)).not.toBe(knowledgeReferenceKey(references[1]!))
|
||||
const message = conversationMessageSchema.parse({ id: crypto.randomUUID(), role: 'assistant', content: '', createdAt: 1, state: 'complete', sourceReferences: references })
|
||||
expect(message.sourceReferences?.[0]?.external).toEqual(references[0]?.external)
|
||||
})
|
||||
|
||||
it('reports unusable credentials and preserves the stored key when encryption fails', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>(async () => Response.json(records()))
|
||||
const { service, cipher } = await setup(fetcher)
|
||||
const saved = instance(service)
|
||||
const before = service.database.externalStore.getInstance(saved.id)
|
||||
const decrypt = vi.spyOn(cipher, 'decrypt').mockImplementation(() => { throw new Error('OS key unavailable') })
|
||||
expect(service.external.listInstances()[0]?.credentialStatus).toBe('unavailable')
|
||||
await expect(service.external.testRetrieval(retrievalInput(saved.id))).rejects.toThrow('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE')
|
||||
decrypt.mockReturnValue(JSON.stringify(' '))
|
||||
expect(service.external.listInstances()[0]?.credentialStatus).toBe('unavailable')
|
||||
await expect(service.external.testRetrieval(retrievalInput(saved.id))).rejects.toThrow('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE')
|
||||
expect(fetcher).not.toHaveBeenCalled()
|
||||
decrypt.mockRestore()
|
||||
vi.spyOn(cipher, 'encrypt').mockImplementation(() => { throw new Error('OS encryption failed') })
|
||||
expect(() => service.external.saveInstance({ id: saved.id, name: 'Changed', provider: saved.provider, baseUrl: saved.baseUrl, enabled: true,
|
||||
credential: { action: 'replace', value: 'new-key' } })).toThrow('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE')
|
||||
expect(service.database.externalStore.getInstance(saved.id)).toEqual(before)
|
||||
expect(service.external.listInstances()[0]?.credentialStatus).toBe('configured')
|
||||
})
|
||||
|
||||
it('persists keep and clear credential mutations and normalized binding updates across restart', async () => {
|
||||
const { service, options } = await setup(async () => Response.json(records()))
|
||||
const saved = instance(service)
|
||||
const binding = await service.external.saveBinding(bindingInput(saved.id))
|
||||
const credential = service.database.externalStore.getInstance(saved.id)?.credential
|
||||
service.external.saveInstance({ id: saved.id, name: 'Renamed', provider: saved.provider, baseUrl: saved.baseUrl, enabled: true, credential: { action: 'keep' } })
|
||||
const updated = await service.external.saveBinding({ ...bindingInput(saved.id), knowledgeBaseId: ` ${binding.knowledgeBaseId} `, name: 'Updated' })
|
||||
expect(updated.knowledgeBaseId).toBe(binding.knowledgeBaseId)
|
||||
expect(service.database.externalStore.getInstance(saved.id)?.credential).toEqual(credential)
|
||||
await service.dispose()
|
||||
services.splice(services.indexOf(service), 1)
|
||||
const reopened = new KnowledgeService(options)
|
||||
services.push(reopened)
|
||||
await reopened.initialize()
|
||||
expect(reopened.external.listInstances()[0]).toMatchObject({ name: 'Renamed', credentialStatus: 'configured', bindingCount: 1 })
|
||||
expect(reopened.database.getKnowledgeBase(binding.knowledgeBaseId)?.name).toBe('Updated')
|
||||
expect(reopened.database.externalStore.listBindings()).toEqual([updated])
|
||||
await expect(reopened.external.testRetrieval(retrievalInput(saved.id))).resolves.toMatchObject({ results: [expect.anything()] })
|
||||
reopened.external.saveInstance({ id: saved.id, name: 'Renamed', provider: saved.provider, baseUrl: saved.baseUrl, enabled: true, credential: { action: 'clear' } })
|
||||
expect(reopened.external.listInstances()[0]).toMatchObject({ credentialStatus: 'missing', bindingCount: 1 })
|
||||
const inspection = new DatabaseSync(options.databasePath)
|
||||
try {
|
||||
const row = inspection.prepare('SELECT value_json FROM external_knowledge_instances WHERE id=?').get(saved.id)!
|
||||
expect(JSON.parse(String(row.value_json))).not.toHaveProperty('credential')
|
||||
} finally { inspection.close() }
|
||||
await expect(reopened.external.testRetrieval(retrievalInput(saved.id))).rejects.toThrow('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE')
|
||||
})
|
||||
|
||||
it('does not invalidate retrieval or binding verification when a concurrent probe persists its result', async () => {
|
||||
const releases: ((response: Response) => void)[] = []
|
||||
const { service } = await setup(() => new Promise<Response>(resolve => releases.push(resolve)))
|
||||
const saved = instance(service)
|
||||
const binding = service.external.saveBinding(bindingInput(saved.id))
|
||||
const retrieval = service.external.testRetrieval(retrievalInput(saved.id))
|
||||
const probe = service.external.testInstance(saved.id)
|
||||
releases[2]!(Response.json({ data: [], has_more: false }))
|
||||
await expect(probe).resolves.toMatchObject({ probeStatus: 'catalog-ready' })
|
||||
releases[0]!(Response.json(records()))
|
||||
releases[1]!(Response.json(records()))
|
||||
await expect(binding).resolves.toMatchObject({ instanceId: saved.id })
|
||||
await expect(retrieval).resolves.toMatchObject({ results: [expect.anything()] })
|
||||
expect(service.external.listInstances()[0]?.probeStatus).toBe('catalog-ready')
|
||||
})
|
||||
|
||||
it('normalizes pre-request cancellation and persists network probe failures as unreachable', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>(async () => { throw new TypeError('fetch failed') })
|
||||
const { service } = await setup(fetcher)
|
||||
const saved = instance(service)
|
||||
await expect(service.external.testRetrieval(retrievalInput(saved.id), AbortSignal.abort())).rejects.toMatchObject({ code: 'EXTERNAL_KB_CANCELLED' })
|
||||
expect(fetcher).not.toHaveBeenCalled()
|
||||
await expect(service.external.testInstance(saved.id)).resolves.toMatchObject({ probeStatus: 'unreachable', lastErrorCode: 'EXTERNAL_KB_NETWORK' })
|
||||
})
|
||||
|
||||
it('includes the RAGFlow capability check in the retrieval timeout budget', async () => {
|
||||
const fetcher = vi.fn<typeof fetch>(async url => {
|
||||
await new Promise(resolve => setTimeout(resolve, 600))
|
||||
return Response.json(String(url).includes('/datasets')
|
||||
? { code: 0, data: [{ id: 'dataset', name: 'Graph', graphrag_task_finish_at: 1 }] }
|
||||
: { code: 0, data: { chunks: [] } })
|
||||
})
|
||||
const { service } = await setup(fetcher)
|
||||
const saved = service.external.saveInstance({ name: 'RAGFlow', provider: 'ragflow', baseUrl: 'https://kb.example', enabled: true, credential: { action: 'replace', value: 'test-key' } })
|
||||
await expect(service.external.testRetrieval({ ...retrievalInput(saved.id), providerConfig: {
|
||||
provider: 'ragflow', similarityThreshold: 0.2, vectorSimilarityWeight: 0.3, knnTopK: 10, useKg: true, includeKnowledgeCompilation: false
|
||||
} })).rejects.toMatchObject({ code: 'EXTERNAL_KB_TIMEOUT' })
|
||||
expect(fetcher).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
@@ -0,0 +1,154 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { externalKnowledgeBindingSaveInputSchema, externalKnowledgeBindingTestInputSchema, externalKnowledgeBindingUpdateInputSchema, externalKnowledgeInstanceSaveInputSchema, type ExternalKnowledgeBinding, type ExternalKnowledgeBindingSaveInput, type ExternalKnowledgeBindingTestInput, type ExternalKnowledgeInstanceSaveInput, type ExternalKnowledgeInstanceSummary, type ExternalKnowledgeTestResult } from '../../../shared/external-knowledge-contracts'
|
||||
import { decryptSettingsCredential, encryptSettingsCredential, type SettingsCredentialCipher } from '../../settings-credential-cipher'
|
||||
import type { KnowledgeDatabase } from '../knowledge-database'
|
||||
import { ExternalKnowledgeClient, ExternalKnowledgeError } from './external-knowledge-client'
|
||||
import type { StoredExternalInstance } from './external-knowledge-store'
|
||||
|
||||
// Probe results are persisted metadata, not a change to an in-flight request's configuration.
|
||||
function instanceConfiguration(value: StoredExternalInstance): string {
|
||||
return JSON.stringify([value.id, value.name, value.provider, value.baseUrl, value.enabled, value.credential])
|
||||
}
|
||||
|
||||
export class ExternalKnowledgeService {
|
||||
private disposed = false
|
||||
private readonly controllers = new Map<AbortController,string>()
|
||||
constructor(private readonly database: KnowledgeDatabase, private readonly cipher?: SettingsCredentialCipher, private readonly fetcher?: typeof fetch) {}
|
||||
dispose(): void { this.disposed = true; for (const controller of this.controllers.keys()) controller.abort(); this.controllers.clear() }
|
||||
private checkActive(): void { if (this.disposed) throw new Error('EXTERNAL_KB_CANCELLED') }
|
||||
private cancelInstance(id: string): void {
|
||||
for (const [controller, instanceId] of this.controllers) if (instanceId === id) controller.abort()
|
||||
}
|
||||
private instance(id:string): StoredExternalInstance {
|
||||
this.checkActive()
|
||||
const value=this.database.externalStore.getInstance(id)
|
||||
if (!value) throw new Error('EXTERNAL_KB_NOT_FOUND')
|
||||
return value
|
||||
}
|
||||
listInstances(): ExternalKnowledgeInstanceSummary[] {
|
||||
this.checkActive()
|
||||
const bindings=this.database.externalStore.listBindings()
|
||||
return this.database.externalStore.listInstances().map(({credential,...value})=>{
|
||||
let credentialStatus: ExternalKnowledgeInstanceSummary['credentialStatus'] = 'missing'
|
||||
if (credential) {
|
||||
try { this.credential({ ...value, credential }); credentialStatus = 'configured' }
|
||||
catch { credentialStatus = 'unavailable' }
|
||||
}
|
||||
return {...value,credentialStatus,bindingCount:bindings.filter(binding=>binding.instanceId===value.id).length}
|
||||
})
|
||||
}
|
||||
private credential(instance: StoredExternalInstance): string {
|
||||
try {
|
||||
if (!instance.credential || !this.cipher?.isAvailable()) throw new Error()
|
||||
const key = decryptSettingsCredential(this.cipher, instance.credential)
|
||||
if (typeof key !== 'string' || !key.trim()) throw new Error()
|
||||
return key
|
||||
} catch { throw new Error('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE') }
|
||||
}
|
||||
saveInstance(raw:ExternalKnowledgeInstanceSaveInput):ExternalKnowledgeInstanceSummary {
|
||||
this.checkActive()
|
||||
const input=externalKnowledgeInstanceSaveInputSchema.parse(raw)
|
||||
const previous=input.id ? this.instance(input.id):undefined
|
||||
if (previous && previous.provider!==input.provider && this.database.externalStore.listBindings().some(binding=>binding.instanceId===previous.id)) throw new Error('EXTERNAL_KB_INSTANCE_IN_USE')
|
||||
let credential=previous?.credential
|
||||
if (input.credential.action==='clear') credential=undefined
|
||||
if (input.credential.action==='replace') {
|
||||
if (!this.cipher?.isAvailable()) throw new Error('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE')
|
||||
try { credential=encryptSettingsCredential(this.cipher,input.credential.value) }
|
||||
catch { throw new Error('EXTERNAL_KB_CREDENTIAL_UNAVAILABLE') }
|
||||
}
|
||||
const value:StoredExternalInstance={id:previous?.id??randomUUID(),name:input.name,provider:input.provider,baseUrl:input.baseUrl.replace(/\/+$/,''),enabled:input.enabled,credential,probeStatus:'untested'}
|
||||
this.database.externalStore.saveInstance(value)
|
||||
this.cancelInstance(value.id)
|
||||
return this.listInstances().find(item=>item.id===value.id)!
|
||||
}
|
||||
setEnabled(id:string,enabled:boolean):ExternalKnowledgeInstanceSummary {
|
||||
const instance=this.instance(id)
|
||||
this.database.externalStore.saveInstance({...instance,enabled})
|
||||
if (!enabled) for(const [controller,instanceId] of this.controllers) if(instanceId===id) controller.abort()
|
||||
return this.listInstances().find(item=>item.id===id)!
|
||||
}
|
||||
deleteInstance(id:string):void {
|
||||
this.instance(id)
|
||||
if(this.database.externalStore.listBindings().some(binding=>binding.instanceId===id)) throw new Error('EXTERNAL_KB_INSTANCE_IN_USE')
|
||||
for(const [controller,instanceId] of this.controllers) if(instanceId===id) controller.abort()
|
||||
this.database.externalStore.deleteInstance(id)
|
||||
}
|
||||
private async request<T>(id:string,timeout:number,signal:AbortSignal|undefined,operation:(client:ExternalKnowledgeClient,signal:AbortSignal)=>Promise<T>):Promise<T> {
|
||||
const instance=this.instance(id)
|
||||
if(!instance.enabled) throw new Error('EXTERNAL_KB_DISABLED')
|
||||
const key = this.credential(instance)
|
||||
const controller=new AbortController()
|
||||
this.controllers.set(controller,id)
|
||||
const timeoutSignal = AbortSignal.timeout(timeout)
|
||||
const combined = AbortSignal.any([controller.signal, timeoutSignal, ...(signal ? [signal] : [])])
|
||||
try {
|
||||
combined.throwIfAborted()
|
||||
const result = await operation(new ExternalKnowledgeClient({provider:instance.provider,baseUrl:instance.baseUrl,apiKey:key,timeoutMs:timeout,fetcher:this.fetcher}),combined)
|
||||
combined.throwIfAborted()
|
||||
return result
|
||||
} catch (error) {
|
||||
if (combined.aborted) {
|
||||
const timedOut = timeoutSignal.aborted && combined.reason === timeoutSignal.reason
|
||||
throw new ExternalKnowledgeError(timedOut ? 'EXTERNAL_KB_TIMEOUT' : 'EXTERNAL_KB_CANCELLED',
|
||||
timedOut ? 'External knowledge request timed out' : 'External knowledge request was cancelled')
|
||||
}
|
||||
throw error
|
||||
} finally {this.controllers.delete(controller)}
|
||||
}
|
||||
async testInstance(id:string):Promise<ExternalKnowledgeInstanceSummary> {
|
||||
const instance=this.instance(id)
|
||||
let probeStatus:StoredExternalInstance['probeStatus']='catalog-ready'
|
||||
let lastErrorCode:string|undefined
|
||||
try {await this.listCatalog({instanceId:id})} catch(error) {
|
||||
lastErrorCode=error instanceof ExternalKnowledgeError ? error.code : error instanceof Error && /^EXTERNAL_KB_/.test(error.message) ? error.message : 'EXTERNAL_KB_NETWORK'
|
||||
probeStatus=lastErrorCode==='EXTERNAL_KB_FORBIDDEN'?'list-restricted':lastErrorCode==='EXTERNAL_KB_AUTH'?'auth-failed':lastErrorCode==='EXTERNAL_KB_NETWORK'||lastErrorCode==='EXTERNAL_KB_TIMEOUT'?'unreachable':'failed'
|
||||
}
|
||||
const current = this.instance(id)
|
||||
if (instanceConfiguration(current) !== instanceConfiguration(instance)) throw new Error('EXTERNAL_KB_CONFIG_CHANGED')
|
||||
this.database.externalStore.saveInstance({...current,probeStatus,lastErrorCode,lastTestedAt:new Date().toISOString()})
|
||||
return this.listInstances().find(item=>item.id===id)!
|
||||
}
|
||||
listCatalog(input:{instanceId:string;page?:number;pageSize?:number;parentId?:string|null;search?:string},signal?:AbortSignal) {
|
||||
return this.request(input.instanceId,15000,signal,(client,current)=>client.listKnowledgeBases(current,input))
|
||||
}
|
||||
getCatalog(input:{instanceId:string;remoteKnowledgeBaseId:string},signal?:AbortSignal) {
|
||||
return this.request(input.instanceId,15000,signal,(client,current)=>client.getKnowledgeBase(input.remoteKnowledgeBaseId,current))
|
||||
}
|
||||
async testRetrieval(raw:ExternalKnowledgeBindingTestInput,signal?:AbortSignal):Promise<ExternalKnowledgeTestResult> {
|
||||
const input=externalKnowledgeBindingTestInputSchema.parse(raw)
|
||||
const instance=this.instance(input.instanceId)
|
||||
if(instance.provider!==input.providerConfig.provider) throw new Error('EXTERNAL_KB_CONFIG_INVALID')
|
||||
const started=Date.now()
|
||||
const results=await this.request(input.instanceId,input.commonConfig.requestTimeoutMs,signal,async (client,current)=>{
|
||||
if(input.providerConfig.provider==='ragflow' && (input.providerConfig.useKg||input.providerConfig.includeKnowledgeCompilation)) {
|
||||
const detail=await client.getKnowledgeBase(input.remoteKnowledgeBaseId,current)
|
||||
if((input.providerConfig.useKg&&!detail.graphEnabled)||(input.providerConfig.includeKnowledgeCompilation&&!detail.knowledgeCompilationEnabled)) throw new Error('EXTERNAL_KB_CAPABILITY_UNAVAILABLE')
|
||||
}
|
||||
return client.retrieve(input.remoteKnowledgeBaseId,input.testQuery,input.providerConfig,current)
|
||||
})
|
||||
if (instanceConfiguration(instance) !== instanceConfiguration(this.instance(input.instanceId))) throw new Error('EXTERNAL_KB_CONFIG_CHANGED')
|
||||
let remaining = 48_000
|
||||
const bounded = results.slice(0, input.commonConfig.resultLimit).map(item => {
|
||||
const snippet = item.snippet.slice(0, Math.min(remaining, input.commonConfig.maxSnippetCharacters))
|
||||
remaining -= snippet.length
|
||||
return { ...item, snippet }
|
||||
}).filter(item => item.snippet.length > 0)
|
||||
return {results:bounded,durationMs:Date.now()-started}
|
||||
}
|
||||
async saveBinding(raw:ExternalKnowledgeBindingSaveInput & {knowledgeBaseId?:string}):Promise<ExternalKnowledgeBinding> {
|
||||
this.checkActive()
|
||||
const input=raw.knowledgeBaseId !== undefined ? externalKnowledgeBindingUpdateInputSchema.parse(raw):externalKnowledgeBindingSaveInputSchema.parse(raw)
|
||||
const knowledgeBaseId = 'knowledgeBaseId' in input ? input.knowledgeBaseId : undefined
|
||||
const existing=knowledgeBaseId ? this.database.externalStore.listBindings().find(item=>item.knowledgeBaseId===knowledgeBaseId):undefined
|
||||
if(knowledgeBaseId&&!existing) throw new Error('EXTERNAL_KB_NOT_FOUND')
|
||||
if(this.database.externalStore.listBindings().some(item=>item.instanceId===input.instanceId&&item.remoteKnowledgeBaseId===input.remoteKnowledgeBaseId&&item.knowledgeBaseId!==existing?.knowledgeBaseId)) throw new Error('EXTERNAL_KB_DUPLICATE_BINDING')
|
||||
const instance = this.instance(input.instanceId)
|
||||
await this.testRetrieval({instanceId:input.instanceId,remoteKnowledgeBaseId:input.remoteKnowledgeBaseId,commonConfig:input.commonConfig,providerConfig:input.providerConfig,testQuery:input.testQuery})
|
||||
if (instanceConfiguration(instance) !== instanceConfiguration(this.instance(input.instanceId))) throw new Error('EXTERNAL_KB_CONFIG_CHANGED')
|
||||
const bindings = this.database.externalStore.listBindings()
|
||||
if (existing && JSON.stringify(bindings.find(item => item.knowledgeBaseId === existing.knowledgeBaseId)) !== JSON.stringify(existing)) throw new Error('EXTERNAL_KB_CONFIG_CHANGED')
|
||||
if (bindings.some(item => item.instanceId === input.instanceId && item.remoteKnowledgeBaseId === input.remoteKnowledgeBaseId && item.knowledgeBaseId !== existing?.knowledgeBaseId)) throw new Error('EXTERNAL_KB_DUPLICATE_BINDING')
|
||||
return this.database.saveExternalBinding({instanceId:input.instanceId,provider:input.providerConfig.provider,remoteKnowledgeBaseId:input.remoteKnowledgeBaseId,remoteName:input.remoteName,commonConfig:input.commonConfig,providerConfig:input.providerConfig,lastVerifiedAt:new Date().toISOString()}, {name:input.name,description:input.description,knowledgeBaseId:existing?.knowledgeBaseId})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
import type { DatabaseSync } from 'node:sqlite'
|
||||
import type { ExternalKnowledgeBinding, ExternalKnowledgeInstanceSummary } from '../../../shared/external-knowledge-contracts'
|
||||
import type { EncryptedSettingsCredential } from '../../settings-credential-cipher'
|
||||
|
||||
export type StoredExternalInstance = Omit<ExternalKnowledgeInstanceSummary, 'credentialStatus' | 'bindingCount'> & {credential?:EncryptedSettingsCredential}
|
||||
|
||||
export class ExternalKnowledgeStore {
|
||||
constructor(private readonly database: DatabaseSync) {}
|
||||
getInstance(id: string): StoredExternalInstance | undefined {
|
||||
const row = this.database.prepare('SELECT value_json FROM external_knowledge_instances WHERE id=?').get(id)
|
||||
return row ? JSON.parse(String(row.value_json)) as StoredExternalInstance : undefined
|
||||
}
|
||||
listInstances(): StoredExternalInstance[] {
|
||||
return this.database.prepare('SELECT value_json FROM external_knowledge_instances ORDER BY rowid').all().map(row => JSON.parse(String(row.value_json)) as StoredExternalInstance)
|
||||
}
|
||||
saveInstance(value: StoredExternalInstance): void {
|
||||
this.database.prepare('INSERT INTO external_knowledge_instances(id,value_json) VALUES (?,?) ON CONFLICT(id) DO UPDATE SET value_json=excluded.value_json').run(value.id, JSON.stringify(value))
|
||||
}
|
||||
deleteInstance(id: string): void {
|
||||
this.database.prepare('DELETE FROM external_knowledge_instances WHERE id=?').run(id)
|
||||
}
|
||||
listBindings(): ExternalKnowledgeBinding[] {
|
||||
return this.database.prepare('SELECT value_json FROM external_knowledge_bindings ORDER BY rowid').all().map(row => JSON.parse(String(row.value_json)) as ExternalKnowledgeBinding)
|
||||
}
|
||||
saveBinding(value: ExternalKnowledgeBinding): void {
|
||||
this.database.prepare('INSERT INTO external_knowledge_bindings(knowledge_base_id,instance_id,remote_id,value_json) VALUES (?,?,?,?) ON CONFLICT(knowledge_base_id) DO UPDATE SET instance_id=excluded.instance_id,remote_id=excluded.remote_id,value_json=excluded.value_json').run(value.knowledgeBaseId,value.instanceId,value.remoteKnowledgeBaseId,JSON.stringify(value))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user