Installation
Basic Usage
Creating an Ingest Job
Here’s how you can use the Agentset Python SDK to create an ingest job:
import os
from agentset import Agentset
a = Agentset(
namespace_id="ns_123",
token=os.environ['AGENTSET_API_KEY'],
)
res = a.ingest_jobs.create(
payload={
"type": "FILE",
"fileUrl": "https://example.com/annual-report-2024.pdf",
"name": "Annual Report 2024",
},
config={
"metadata": {
"documentType": "financial",
"year": 2024,
},
},
)
print(f"Job created with ID: {res.data.id}")
Ingest jobs are processed asynchronously. After creating a job, you’ll need to check its
status to know when processing is complete.
Checking Job Status
Since ingestion is asynchronous, you can check the status of your job:
import time
# Get the job status
job = a.ingest_jobs.get(res.data.id)
print(f"Job status: {job.data.status}")
# Poll for completion
while job.data.status not in ["COMPLETED", "FAILED"]:
time.sleep(10) # Wait 10 seconds before checking again
job = a.ingest_jobs.get(res.data.id)
print(f"Job status: {job.data.status}")
if job.data.status == "COMPLETED":
print("Job completed successfully!")
elif job.data.status == "FAILED":
print(f"Job failed: {job.error}")
Searching Documents
Once your documents are ingested, you can search through them:
# Perform a search
search_results = a.search.execute(
query="financial performance Q4 2024",
top_k=20,
rerank=True,
rerank_limit=10,
include_metadata=True,
filters={
"$or": [
{"documentType": "financial"},
{"year": {"$gte": 2024}}
]
}
)
# Process results
for result in search_results.data:
print(f"Score: {result.score}")
print(f"Content: {result.text}")
print(f"Metadata: {result.metadata}")
print("---")
Additional Resources
