Platform Quickstart (API)

# Adding Nasa PDF's

datastore_name = 'NASA_Datastore'

# Check if datastore exists
datastores = client.datastores.list()
existing_datastore = next((ds for ds in datastores if ds.name == datastore_name), None)

if existing_datastore:
    datastore_id = existing_datastore.id
    print(f"Using existing datastore with ID: {datastore_id}")
else:
    result = client.datastores.create(name=datastore_name)
    datastore_id = result.id
    print(f"Created new datastore with ID: {datastore_id}")

print(client.datastores.list())

import os
import requests

# Create data directory if it doesn't exist
if not os.path.exists('data'):
    os.makedirs('data')

files_to_upload = [
    (
        "A_Fault_Recovery_Distributed_Electric_Propulsion.pdf",
        "https://ntrs.nasa.gov/api/citations/20240013567/downloads/TM-20240013567.pdf",
    ),
    (
        "B_Lunar_Surface_Autonomy_Certification_FDIR.pdf",
        "https://ntrs.nasa.gov/api/citations/20250010214/downloads/MDA%20Paper%20ISS%20CaseStudy%202025%20Author%20Name%20Added.docx",
    ),
]

# Download and ingest all files
document_ids = []
for filename, url in files_to_upload:
    file_path = f'data/{filename}'

    # Download file if it doesn't exist
    if not os.path.exists(file_path):
        print(f"Fetching {file_path}")
        try:
            response = requests.get(url)
            response.raise_for_status()  # Raise an exception for bad status codes
            with open(file_path, 'wb') as f:
                f.write(response.content)
        except Exception as e:
            print(f"Error downloading {filename}: {str(e)}")
            continue

    # Upload to datastore
    try:
        with open(file_path, 'rb') as f:
            ingestion_result = client.datastores.documents.ingest(datastore_id, file=f)
            document_id = ingestion_result.id
            document_ids.append(document_id)
            print(f"Successfully uploaded {filename} to datastore {datastore_id}")
    except Exception as e:
        print(f"Error uploading {filename}: {str(e)}")

print(f"Successfully uploaded {len(document_ids)} files to datastore")
print(f"Document IDs: {document_ids}")

metadata = client.datastores.documents.metadata(datastore_id = datastore_id, document_id = document_ids[0])
print("Document metadata:", metadata)

# Load the YAML (The workflow graph -> will add an explanation to this as well)
acl_yaml = """
version: 0.1
inputs:
  query: str

outputs:
  response: str

nodes:
  create_message_history:
    type: CreateMessageHistoryStep
    input_mapping:
      query: __inputs__#query

  research:
    type: AgenticResearchStep
    ui_stream_types:
      retrievals: true
    config:
      tools_config:
        - name: search_docs
          description: |
            Search the datastore containing user-uploaded documents. This datastore is a vector database of text chunks which uses hybrid semantic and lexical search to find the most relevant chunks.
            Use this tool to find information within the uploaded documents.
          step_config:
            type: SearchUnstructuredDataStep
            config:
              top_k: 50
              lexical_alpha: 0.1
              semantic_alpha: 0.9
              reranker: "ctxl-rerank-v2-instruct-multilingual-FP8"
              rerank_top_k: 12
              reranker_score_filter_threshold: 0.2

      agent_config:
        agent_loop:
          num_turns: 10
          parallel_tool_calls: false
          model_name_or_path: "vertex_ai/claude-opus-4-5@20251101"
          identity_guidelines_prompt: |
            You are a retrieval-augmented assistant created by Contextual AI. You provide factual, grounded answers to user's questions by retrieving information via tools and then synthesizing a response based only on what you retrieved.

          research_guidelines_prompt: |
            You MUST always explore the unstructured datastore before answering. Use breadth-then-depth retrieval, avoid redundant searches, and be comprehensive.

    input_mapping:
      message_history: create_message_history#message_history

  generate:
    type: GenerateFromResearchStep
    ui_stream_types:
      generation: true
    config:
      model_name_or_path: "vertex_ai/claude-opus-4-5@20251101"
      identity_guidelines_prompt: |
        You are a retrieval-augmented assistant created by Contextual AI. Provide factual, grounded answers based only on retrieved information.
      response_guidelines_prompt: |
        - Output concise Markdown with headings/bullets.
        - Start immediately with the answer (no preamble).
        - If info is missing, say what's missing and what you can infer safely.

    input_mapping:
      message_history: create_message_history#message_history
      research: research#research

  __outputs__:
    type: output
    input_mapping:
      response: generate#response
""".strip()

import requests
from google.colab import userdata

BASE_URL = userdata.get("CONTEXTUAL_BASE_URL")
API_KEY  = userdata.get("CONTEXTUAL_API_KEY")
HEADERS  = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

query = "What are the main topics or themes covered in the documents?"

print(f"Query: {query}\n")

payload = {
    "messages": [{"role": "user", "content": query}],
    "stream": False
}

resp = requests.post(f"{BASE_URL}/agents/{agent_id}/query/acl", headers=HEADERS, json=payload)
resp.raise_for_status()
out = resp.json()

print(out["message"]["content"])