# Auto generated by LangBot Plugin SDK.
# Please refer to https://docs.langbot.app/en/plugin/dev/tutor.html for more details.
...
class FastGPT(KnowledgeRetriever):
async def retrieve(self, context: RetrievalContext) -> list[RetrievalResultEntry]:
"""Retrieve knowledge from FastGPT knowledge base"""
# 1. Get configuration parameters
api_base_url = self.config.get('api_base_url', 'http://localhost:3000')
api_key = self.config.get('api_key')
dataset_id = self.config.get('dataset_id')
search_mode = self.config.get('search_mode', 'embedding')
using_rerank = self.config.get('using_rerank', False)
# 2. Parameter validation
if not api_key or not dataset_id:
logger.error("Missing required configuration: api_key or dataset_id")
return []
# 3. Build API request
url = f"{api_base_url.rstrip('/')}/api/core/dataset/searchTest"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"datasetId": dataset_id,
"text": context.query, # Get query text from RetrievalContext
"searchMode": search_mode,
"usingReRank": using_rerank,
}
try:
# 4. Call external API
async with httpx.AsyncClient() as client:
response = await client.post(url, json=payload, headers=headers, timeout=30.0)
response.raise_for_status()
result = response.json()
# 5. Parse response and convert to RetrievalResultEntry
results = []
for record in result.get('data', []):
# Combine main data and auxiliary data as content
content_text = '\n'.join([
record.get('q', ''),
record.get('a', '')
]).strip()
# Create retrieval result entry
entry = RetrievalResultEntry(
id=record.get('id', ''),
content=[ContentElement.from_text(content_text)],
metadata={
'dataset_id': record.get('datasetId', ''),
'source_name': record.get('sourceName', ''),
'score': record.get('score', 0.0),
},
# Convert similarity score to distance (higher score = smaller distance)
distance=1.0 - float(record.get('score', 0.0)),
)
results.append(entry)
logger.info(f"Retrieved {len(results)} chunks from FastGPT dataset {dataset_id}")
return results
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error from FastGPT API: {e.response.status_code} - {e.response.text}")
return []
except httpx.RequestError as e:
logger.error(f"Request error when calling FastGPT API: {str(e)}")
return []
except Exception as e:
traceback.print_exc()
logger.error(f"Unexpected error during retrieval: {str(e)}")
return []
