Integration Guides

def create_ai_result(
    user: core_user.CoreUser,
    data: dict,
    cfg: dict = None,
):
    """
    create_ai_result

    provides an api for remote llm(s)
    to stream results over https
    without being co-located with the
    llm(s)

    e.g. run synthetic dataset generative
    jobs on the cloud (aws, gcp, azure, runpod.io
    and stream the results to your self-hosted rest api)

    :param CoreUser user: authenticated user
        that is making this request
    :param data: dictionary from
        ``data=CoreResultAI.get_dict()``
    :param cfg: optional **CoreConfig**
        dictionary

    :returns: **CoreResultAI** on success
        **None** on non-success
    :rtype: CoreResultAI or None
    """
    if not cfg:
        cfg = get_cfg.get_cfg()
    url = f'https://{cfg["endpoint"]}/ai/result'
    (cert_file, key_file) = tls_utils.get_certs(cfg)
    verify = tls_utils.get_verify(cfg)
    debug = cfg.get("debug", False)
    log.debug(f"create ai result: {url}")
    s = requests.Session()
    s.headers.update({"Bearer": f"{user.token}"})
    data.pop("id", None)
    r = s.post(
        url,
        json=data,
        verify=verify,
        cert=(cert_file, key_file),
        timeout=5,
    )
    if r.status_code != 201:
        log.error(
            "\n\n"
            "non-201 response:\n"
            f"  url: {url}\n"
            f"  job_result.id: {id}\n"
            f"  ca={verify}\n"
            f"  response:\n"
            f"  code: {r.status_code}\n"
            f"  text:\n"
            f"  {r.text}\n"
        )
        return None
    else:
        if debug:
            log.debug(
                f"create ai result success - {r.text}"
            )
        try:
            rec_dict = json.loads(r.text)
            cur_o = core_result_ai.CoreResultAI()
            cur_o.load_response_dict(rec_dict=rec_dict)
            return cur_o
        except Exception as e:
            log.error(
                f'failed to create ai result with ex="{e}"'
            )
    return None

def get_ai_result(
    id: int,
    user: core_user.CoreUser,
    cfg: dict = None,
):
    """
    get_ai_result

    after the llm finishes processing the question,
    you can get the ai results table using this
    method

    get the ai result by the **CoreJob.id** value
    using the rest api

    :param id: look up this **CoreJob.id**'s
        ai results
    :param CoreUser user: authenticated user
        that is making this request
    :param cfg: optional **CoreConfig** dictionary

    :returns: **CoreResultAI** on success
        **None** on non-success
    :rtype: CoreResultAI or None
    """
    if not cfg:
        cfg = get_cfg.get_cfg()
    url = f'https://{cfg["endpoint"]}/ai/result/{id}'
    (cert_file, key_file) = tls_utils.get_certs(cfg)
    verify = tls_utils.get_verify(cfg)
    debug = cfg.get("debug", False)
    log.debug(f"get ai result: {url}")
    s = requests.Session()
    s.headers.update({"Bearer": f"{user.token}"})
    data = {"user_id": user.id, "job_id": id}
    r = s.get(
        url,
        json=data,
        verify=verify,
        cert=(cert_file, key_file),
        timeout=5,
    )
    if r.status_code != 200:
        log.error(
            "\n\n"
            "non-200 response:\n"
            f"  url: {url}\n"
            f"  job_result.id: {id}\n"
            f"  ca={verify}\n"
            f"  response:\n"
            f"  code: {r.status_code}\n"
            f"  text:\n"
            f"  {r.text}\n"
        )
        return None
    else:
        if debug:
            log.info(f"get ai result success - {r.text}")
        try:
            cur_json = json.loads(r.text)
            cur_o = core_result_ai.CoreResultAI(
                id=cur_json.get("id", None),
                user_id=cur_json.get("user_id", None),
                job_id=cur_json.get("job_id", None),
                worker_id=cur_json.get("worker_id", None),
                state=cur_json.get("state", None),
                question=cur_json.get("question", None),
                answer=cur_json.get("answer", None),
                model_name=cur_json.get("model_name", None),
                score=cur_json.get("score", None),
                question_score=cur_json.get(
                    "question_score", None
                ),
                answer_score=cur_json.get(
                    "answer_score", None
                ),
                match_source=cur_json.get(
                    "match_source", None
                ),
                match_page=cur_json.get("match_page", None),
                match_content=cur_json.get(
                    "match_content", None
                ),
                summarized_question=cur_json.get(
                    "summarized_question", None
                ),
                summarized_answer=cur_json.get(
                    "summarized_answer", None
                ),
                summarized_score=cur_json.get(
                    "summarized_score", None
                ),
                reviewed_answer=cur_json.get(
                    "reviewed_answer", None
                ),
                reviewed_score=cur_json.get(
                    "reviewed_score", None
                ),
                reviewed_computed_score=cur_json.get(
                    "reviewed_computed_score",
                    None,
                ),
                reviewed_notes=cur_json.get(
                    "reviewed_notes", None
                ),
                collection=cur_json.get("collection", None),
                collection_notes=cur_json.get(
                    "collection_notes", None
                ),
                category=cur_json.get("category", None),
                tags=cur_json.get("tags", None),
                latency=cur_json.get("latency", None),
                data=cur_json.get("data", None),
                created_at=cur_json.get("created_at", None),
                updated_at=cur_json.get("updated_at", None),
            )
            return cur_o
        except Exception as e:
            log.error(
                f'failed to get ai_result.id={id} with ex="{e}"'
            )
    return None

def search_ai_results(
    user: core_user.CoreUser,
    data: dict,
    cfg: dict = None,
):
    """
    search_ai_results

    Search for ai results within the database

    search for specific ai results using the
    supported parameters

    :param CoreUser user: authenticated user
        that is making this request
    :param data: request values dictionary
    :param cfg: optional **CoreConfig** dictionary

    :returns: **CoreSearchResultAI** on success
        **None** on non-success
    :rtype: CoreSearchResultAI or None
    """
    if not cfg:
        cfg = get_cfg.get_cfg()
    url = f'https://{cfg["endpoint"]}/ai/result/search'
    (cert_file, key_file) = tls_utils.get_certs(cfg)
    debug = cfg.get("debug", False)
    verify = tls_utils.get_verify(cfg)
    log.debug(f"search ai result: {url}")
    s = requests.Session()
    s.headers.update({"Bearer": f"{user.token}"})
    r = s.post(
        url,
        json=data,
        verify=verify,
        cert=(cert_file, key_file),
        timeout=5,
    )
    if r.status_code != 200:
        log.error(
            "\n\n"
            "non-200 response:\n"
            f"  url: {url}\n"
            f"  job_result.id: {id}\n"
            f"  ca={verify}\n"
            f"  response:\n"
            f"  code: {r.status_code}\n"
            f"  text:\n"
            f"  {r.text}\n"
        )
        return None
    else:
        if debug:
            log.info(f"search ai result success - {r.text}")
        try:
            rec_dict = json.loads(r.text)
            cur_o = (
                core_search_result_ai.CoreSearchResultAI()
            )
            cur_o.load_response_dict(rec_dict=rec_dict)
            return cur_o
        except Exception as e:
            log.error(
                f"failed to search ai_result.id={id} "
                f'with ex="{e}"'
            )
    return None

def update_ai_result(
    user: core_user.CoreUser,
    ai_result: core_result_ai.CoreResultAI,
    cfg=None,
):
    """
    update_ai_result

    update an existing ai result

    supports reinforcement learning
    using human feedback with
    the **reviewed_answer** and **reviewed_score** fields

    :param CoreUser user: authenticated user
        that is making this request
    :param ai_result: in-memory object with
        values to send to the database
    :param cfg: optional **CoreConfig** dictionary

    :returns: **CoreResultAI** on success
        **None** on non-success
    :rtype: CoreResultAI or None
    """
    if not cfg:
        cfg = get_cfg.get_cfg()
    url = f'https://{cfg["endpoint"]}/ai/result'
    (cert_file, key_file) = tls_utils.get_certs(cfg)
    verify = tls_utils.get_verify(cfg)
    debug = cfg.get("debug", False)
    log.debug(f"update ai result: {url}")
    s = requests.Session()
    s.headers.update({"Bearer": f"{user.token}"})
    data = ai_result.get_dict()
    r = s.put(
        url,
        json=data,
        verify=verify,
        cert=(cert_file, key_file),
        timeout=5,
    )
    if r.status_code != 200:
        log.error(
            "\n\n"
            "non-200 response:\n"
            f"  url: {url}\n"
            f"  job_result.id: {id}\n"
            f"  ca={verify}\n"
            f"  response:\n"
            f"  code: {r.status_code}\n"
            f"  text:\n"
            f"  {r.text}\n"
        )
        return None
    else:
        if debug:
            log.info(f"update ai result success - {r.text}")
        try:
            cur_json = json.loads(r.text)
            cur_o = core_result_ai.CoreResultAI()
            cur_o.load_response_dict(rec_dict=cur_json)
            return cur_o
        except Exception as e:
            log.error(
                f'failed to update ai_result with ex="{e}"'
            )
    return None