# Introduction ## Overview Canso is a Managed Data and Feature Platform for operationalizing Machine Learning initiatives. The goal of Canso is to enable ML Teams (Data Engineers, Data Scientists, ML Engineers) to define their requirements in a declarative and standardized manner via a concise [DSL](https://en.wikipedia.org/wiki/Domain-specific_language) without having to focus on writing custom code for Features, DAGs etc and managing infrastructure. This enables ML teams to * Iterate fast i.e. move from development to production in hours/days as opposed to weeks * Promote Reliability i.e build standardized ML pipelines Canso's core focus is on user experience and speed of iteration, without compromising on reliability - * Define data sources where features can be created and computed. * Specify data sinks where processed data is stored after a successful ML pipeline run. * Define Machine Learning features in a standardized manner on top of existing Datasources and deploy them. These features can be used while Model training as well as for Model inference. Canso supports Raw, Derived and Streaming features currently. * Register and deploy features to execute the ML pipeline. ## User Experience ## Getting Started #### 1. Install Gru Package For installing gru package will need to username and PAT as password. * A Personal Access Token (PAT) is a kind of key that authenticates a user across all applications they have access to. ```python pip3 install git+https://github.com/Yugen-ai/gru.git ``` #### 2. Create Yugen client ```python yugen_client = YugenClient(access_token=access_token, config_path="./gru/config.yaml") ``` #### 3. Define a s3 Data Source ```python s3_data_source_obj = S3DataSource( name="survey_telemetry_data", bucket="internal-ml-demos", base_key="recsys/survey-data/phase3_3/survey-telemetry/", varying_key_suffix_format="%Y-%m-%d/%H%M", varying_key_suffix_freq="30min", time_offset=0, description="random desc of data source", owners=["xyz"], created_at=datetime.now(), file_type=CSVType(header=True), schema=schema_obj, event_timestamp_field="time", event_timestamp_format="yyyy-MM-dd HH:mm:ssXXX", ) ``` #### 4. Register Data Source ```python yugen_client.register(s3_data_source_obj) ``` #### 5. Define a Raw Feature ```python raw_feature = RawFeature( name="avg_order_val_3_days", description="Avg order per cusotmer for last 3 days", data_type=DataType.FLOAT, source_name=["survey_telemetry_data"], staging_sink=["s3_sink_ml_yugen_internal"], online_sink=["elasticache-redis-yugen"], owners=["vanshika@yugen.ai"], entity=["test"], feature_logic=FeatureLogic( field=["ad_id"], filter="""ad_id is NOT NULL""", transform=SlidingWindowAggregation( function="avg", partition_by="provider", order_by="cpi", # rangeBetween= {"frame_start": 1, "frame_end": 6}, rowsBetween={"frame_start": 1, "frame_end": 2}, ), time_window="3d", groupby_keys=["project_id"], timestamp_field="time", timestamp_format="yyyy-MM-dd HH:mm:ssXXX", ), online=True, offline=True, schedule="0 0 * * *", active=True, start_time=datetime(2023, 4, 1, 0, 0, 0), ) ``` #### 6. Register Raw Feature ```python yugen_client.register(raw_feature) ``` #### 7. Dry run Raw Feature ```python yugen_client.dry_run("avg_order_val_3_days", entity_type=EntityType.RAW_FEATURE, start_date=datetime(2023, 4, 1, 0, 0, 0), end_date=datetime(2023, 4, 2, 0, 0, 0)) ``` #### 8. Deploy Raw Feature ```python yugen_client.deploy("avg_order_val_3_days", EntityType.RAW_FEATURE) ``` #### 9. Define a Derived Feature ```python derived_feature = DerivedFeature( name="total_purchases", description="Total purchase amount for the store", staging_sink=["s3_sink_ml_yugen_internal"], online_sink=["elasticache-redis-yugen"], data_type=DataType.FLOAT, owners=["all-ds@company.com"], schedule="0 0 * * *", entity=["CASE WHEN cpi> 0.5 THEN 10 ELSE 0 END"], online=False, offline=True, transform=multiply("avg_orders_last_3_days", "number_users"), start_time=datetime(2022, 8, 26, 0, 0, 0), ) ``` #### 10. Register Derived Feature ```python yugen_client.register(derived_feature) ``` #### 11. Dry run Derived Feature ```python yugen_client.dry_run("total_purchases", entity_type=EntityType.DERIVED_FEATURE, start_date=datetime(2022, 8, 26, 0, 0, 0), end_date=datetime(2022, 8, 27, 0, 0, 0)) ``` #### 12. Deploy Derived Feature ```python yugen_client.deploy("total_purchases", EntityType.DERIVED_FEATURE) ``` #### 13. Define Pre-Processing Transform ```python ppt = PreProcessingTransform( transform_name="user_avg_spend_transform_final_testing_for_dry_run", description="test preprocess transform", data_source_names=["marketing_survey_data_info", "data_telemetry_info"], data_source_lookback_config={ "marketing_survey_data_info": "1d", "data_telemetry_info": "1d", }, staging_sink=["s3_sink_ml_yugen_internal"], logic=sql_logic, schedule="0 0 * * *", output_schema=schema_obj, owners=["john.doe@company.ai"], active=True, transform_start_time=datetime(2022, 8, 27, 0, 0, 0), ) ``` #### 14. Register Pre-Processing Transform ```python yugen_client.register(ppt) ``` #### 15. Dry run Pre-Processing Transform ```python yugen_client.dry_run( "user_avg_spend_transform_final_testing_for_dry_run", entity_type=EntityType.PRE_PROCESSING_TRANSFORM, start_date=datetime(2022, 8, 27, 0, 0, 0), end_date=datetime(2022, 8, 28, 0, 0, 0), ) ``` #### 16. Deploy Pre-Processing Transform ```python yugen_client.deploy( "user_avg_spend_transform_final_testing_for_dry_run", EntityType.PRE_PROCESSING_TRANSFORM ) ``` #### 17. Define Training Data ```python training_run = TrainingData( name="example_training_run", description="A sample run to generate training data", historical_data_source="user_events", entities=["project_id","cpi"], features=["raw_for_testing_gtd","raw_for_testing_gtd_for_telemetry","raw_for_testing_gtd_for_survey","raw_for_testing_gtd_for_user_events","raw_for_testing_gtd_for_weather_forecats"], ttl=5, owners=["john.doe@company.ai"], ) ``` #### 18. Register Training Data ```python yugen_client.register(training_run) ``` #### 19. Deploy Training Data ```python yugen_client.deploy("example_training_run", EntityType.TRAINING_DATA) ``` #### 20. Define Infrastructure Data ```python register_infrastructure = RegisterInfrastructure( client_id="platform-release-1", cloud_provider="aws", cluster_name="yugen-platform-v2", region="ap-south-1", subnet_ids=["subnet-4b250507", "subnet-fac53691", "subnet-86cea5fd"], resource_node_group_instance_types={ "instance_types": {"node_group_1": "t3.medium", "node_group_2": "t3.large",} }, resource_node_group_scaling_config={ "scaling_config": { "node_group_1": {"max_size": 6, "min_size": 2, "desired_size": 3}, "node_group_2": {"max_size": 6, "min_size": 2, "desired_size": 3,}, } }, admins=[ "arn:aws:iam::832344679060:user/ashish.prajapati@yugen.ai", "arn:aws:iam::832344679060:user/john.doe@company.ai", "arn:aws:iam::832344679060:user/sandeep.mishra@yugen.ai", "arn:aws:iam::832344679060:user/shaktimaan@yugen.ai", "arn:aws:iam::832344679060:user/shashank.mishra@yugen.ai", "arn:aws:iam::832344679060:user/soumanta@yugen.ai", "arn:aws:iam::832344679060:user/vanshika.agrawal@yugen.ai", ], airflow_users={ "admin": { "username": "admin", "password": "yugen@123", "email": "admin@example.com", "firstName": "admin", "lastName": "admin", } }, slack_details={ "failure_alerts": { "host_url": "https://hooks.slack.com/services", "host_password": "/TSRAELEL9/B04Q09X9W75/PhfxMaFBE81ZBXjeAktTTIyN", }, "notifications": { "host_url": "https://hooks.slack.com/services", "host_password": "/TSRAELEL9/B04Q09X9W75/PhfxMaFBE81ZBXjeAktTTIyN", }, }, created_at =datetime(2023, 4, 28, 0, 0, 0), ) ``` #### 21. Register Infrastructure Data ```python yugen_client.register(register_infrastructure) ``` #### 22. Deploy Infrastructure Data ```python yugen_client.deploy("platform-release-1_yugen-platform-v2_2023-04-28 00:00:00", EntityType.INFRASTRUCTURE) ``` ## Roadmap ### DataSources #### Batch * [x] S3 * [ ] GCS * [ ] RedShift * [ ] BigQuery * [ ] Snowflake #### Streaming * [x] Kafka ### DataSinks #### Online DataSinks Online data sinks offers real-time data storage for fast write operations. It ensures low-latency access to data, making it suitable for applications requiring immediate data retrieval and updates, such as retrieval for ML predictions. Currently, Canso supports Redis cache for storing data online. #### Offline DataSinks Offline data sinks provides durable and scalable storage for batch-processed and historical data. It supports large volumes of data with high reliability, making it ideal for data warehousing and archival storage. Currently, Canso supports S3 storing data offline. #### Batch * [x] S3 * [x] Redis * [ ] RedShift * [ ] DynamoDB #### Streaming * [ ] Kafka ### Online Feature Store * [x] Elasticache for Redis (AWS) * [ ] Memorystore for Redis (GCP) * [ ] DynamoDB * [ ] Bigtable # Canso Architecture ![Components Overview](/files/FVdBIn7fVEHVmvYFqxIP) ### Communication - Control Plane & Data Plane Canso Control Plane communicates with Tenant Data Plane cluster using message queues. The diagram below shows how canso communicates with multiple tenants, also allowing each tenant to own multiple Data Plane clusters. ![Communication - Control Plane & Data Plane](/files/Obsv3N1tH56s4eWY46iS) #### Queues Each data plane cluster has 2 dedicated queues: 1. A **Canso Outgoing Queue** which is used to send the instructions sent by user (using Canso Web app or Python Client) from control plane to data plane. 2. A **Canso Incoming Queue** which is used to send monitoring information (which can be viewed by the user on the Canso Web App) from data plane to control plane. #### Canso Agent The Canso Agent runs in each Data Plane cluster, reads messages from its instruction queue and performs instructed operations in the data plane. #### Canso Notification Subscriber A dedicated Canso Notification Subscriber for each Data Plane cluster runs in the Control Plane. It reads monitoring information from Canso Incoming Queue and makes it available to be viewed by end users. A dedicated Notification Subscriber ensures that each tenant gets equal priority on the control plane and is unaffected by outages/issues related to other tenants. # Overview The Canso Platform is designed to support production grade ML use cases (features, pre-processing transforms etc) from Day 1. There are 2 important pre-requisites before Data Scientists/ML Engineers can use the core functionalities of the Platform - 1. Setting up an Kubernetes Cluster & other infra components that are Canso compatible. This includes core infra (but is not limited to) VPCs, subnets, the K8s cluster and other cloud services such as EFS (to store Airflow DAG files), Redis (to act as the online feature store), RDS (production grade DB for Airflow) etc. 2. Installing Canso Helm charts to bootstrap & deploy Canso services & applications. We expose a superchart for each cloud provider, which automatically installs all applications & services needed to set up the Platform. The Helm chart installation is dependant on the successful provisioning of Step 1 and on ArgoCD being installed in the Canso compatible cluster. The Canso Superchart by default installs Prometheus, Grafana, Loki etc so that you do not have to spend additional effort monitoring your applications. *** * Go back to [Table of Contents](https://github.com/Yugen-ai/gru/blob/main/gru_docs/summary.md) * Reach out to us [Canso Community](https://github.com/Yugen-ai/gru/blob/main/gru_docs/gru_docs/community.md) # Provison K8s Clusters This guide illustrates how to provision Kubernetes clusters on different cloud providers using Terraform. Jobs, Pipelines, Services, Applications managed by Canso, will run on these provisoned clusters. > These clusters are also referred to as a the Data Plane Cluster. The Data Plane resides in the Customer's/Tenant's Cloud, therefore your data never leaves your cloud environment. ## AWS - EKS The Data Plane cluster (which is compatible with Canso Platform) has the following components * VPC and Subnets * EKS Cluster * IAM/IRSA Roles * Drivers such as EFS, EBS etc * A production grade DB such as RDS * S3 Buckets (These are buckets where Canso jobs will persist outputs and artifacts) * AWS Secrets Manager to Store Secrets * A key-value store such as Redis All necessary modules, scripts to provision the above are in this open sourced [`canso-data-plane-k8s-cluster-tf`](https://github.com/Yugen-ai/canso-data-plane-k8s-cluster-tf) repository. Make sure you follow the steps noted in the [Usage section ](https://github.com/Yugen-ai/canso-data-plane-k8s-cluster-tf?tab=readme-ov-file#usage)and please be mindful of the disclaimers and special comments highlighted in there. If you do not have a dedicated DevOps/Infra team, the Canso team can help provision the infra for you. Please reach out to us if you need any assistance and see the [Canso Community page](https://github.com/Yugen-ai/gru/blob/main/gru_docs/community.md) for ways to reach out to us. ## GCP - GKE * [ ] Not yet available. Part of the roadmap. ## Azure - AKS * [ ] Not yet available. Part of the roadmap. ## Oracle - OKE * [ ] Not yet available. Part of the roadmap. ## Red Hat Openshift * [ ] Not yet available. Part of the roadmap. *** * Go back to [Table of Contents](https://github.com/Yugen-ai/gru/blob/main/gru_docs/summary.md) * Reach out to us [Canso Community](https://github.com/Yugen-ai/gru/blob/main/gru_docs/gru_docs/community.md) # Install Canso Helm Charts This guide illustrates how to deploy the Canso Superchart. This is a [Helm chart](https://helm.sh/), which deploys Canso services and the Canso Agent as ArgoCD applications. Currently, ArgoCD must be installed in the cluster for our helm chart to work. You can [install ArgoCD using kubectl commands](https://argo-cd.readthedocs.io/en/stable/#quick-start) or even via Terraform. #### Option 1 - ArgoCD ```console kubectl create namespace argocd kubectl apply -n argocd -f https://raw.githubusercontent.com/argoproj/argo-cd/stable/manifests/install.yaml ``` > \[!IMPORTANT] We recommend checking out the latest stable documentation in ArgoCD in case the commands above change. #### Option 2 - Terraform ```terraform # main.tf resource "kubernetes_namespace" "argocd" { metadata { name = var.namespace } } resource "helm_release" "argocd" { name = "argocd" repository = "https://argoproj.github.io/argo-helm" chart = "argo-cd" version = var.argo_version" namespace = var.namespace timeout = "900" values = [file(var.argo_values_file)] depends_on = [kubernetes_namespace.argocd] } ``` > \[!IMPORTANT] We recommend defining the variables in a `variables.tf` file and setting the values in `auto.tfvars` files. Also, for more details, check out the official `helm_release` resource in the [official Terraform Documentation](https://registry.terraform.io/providers/hashicorp/helm/latest/docs/resources/release). ## Installing the Canso Superchart Helm Helm charts including the superchart are available in [this open source repository](https://github.com/Yugen-ai/canso-helm-charts) and the charts are hosted on [gh-pages](https://yugen-ai.github.io/canso-helm-charts/). ### Prerequisites Before updating the chart, ensure that you have added the [argocd cluster context](https://argo-cd.readthedocs.io/en/stable/user-guide/commands/argocd_cluster_add/) locally. Follow these steps: 1. Add the cluster to ArgoCD using the AWS ARN: ```console argocd cluster add arn:aws:eks:::cluster/ ``` Replace the ARN with your specific cluster ARN. 2. When prompted with the following message, type 'y' and press Enter: ``` This will create a service account `argocd-manager` on the cluster referenced by context `arn:aws:eks:::` with full cluster level privileges. Do you want to continue [y/N]? ``` Note: This step is crucial as the chart depends on the argocd-manager serviceaccount being present in the cluster. Here are the steps to install the superchart - ### Add Helm Repo ```console helm repo add canso-helm-charts https://yugen-ai.github.io/canso-helm-charts/ ``` ### (Optional) Check all charts available in the repo ```console helm search repo canso-helm-chart ``` ### Install the chart ```console helm install -f my-values.yaml canso-superchart canso-helm-charts/canso-aws-eks-superchart --wait ``` #### Values for the Superchart Most of the values for the Superchart are automatically populated when you log into the Canso Web App and navigate the Install agent process. All you have to do is fill in certain inputs related to your Cluster and the Role ARN names that you used at the time of [provisioning IRSA roles](https://github.com/Yugen-ai/canso-data-plane-k8s-cluster-tf?tab=readme-ov-file#integration-with-canso-platform-helm-charts). > \[!CAUTION] Some of the values in the Superchart are references to resources in the Canso Control Plane for e.g. Canso API Server, Message Queues, Image Pull Secrets[^1]. We do not have a completely automated process to share these values at the time of our MVP release and therefore we request you to reach our to the Canso team for the same. We will have a more seamless process in future releases. *** ### Updating the chart #### Scenario 1: Upgrading to a Newer Chart Version If you want to install a newer version of the chart, follow these steps: 1. Update the Helm repository: ```console helm repo update ``` 2. Upgrade the chart, specifying the new version: ```console helm upgrade canso-superchart canso-helm-charts/canso-aws-eks-superchart --version -f my-values.yaml ``` Replace \ with the desired chart version. #### Scenario 2: Updating Configuration (Same Chart Version) If you only need to update your configuration without changing the chart version: 1. Modify your `my-values.yaml` file with the desired changes. 2. Apply the updates using: ```console helm upgrade canso-superchart canso-helm-charts/canso-aws-eks-superchart -f my-values.yaml ``` This command will use the same chart version but apply your updated configuration. ### Troubleshooting: Manually Installing Hooks If you encounter an error during the initial installation and the hooks fail to install, follow these steps to manually install the hooks: 1. Retrieve the hook manifests: ```console helm get hooks canso-superchart > hooks.yaml ``` This command will save the hook manifests to a file named hooks.yaml. 2. Apply the hooks manually using kubectl: ```console kubectl apply -f hooks.yaml ``` Note: Manually installing hooks should be done cautiously and only when necessary. If you're unsure, consult with your system administrator or the chart maintainer. * Go back to [Table of Contents](https://github.com/Yugen-ai/gru/blob/main/gru_docs/summary.md) * Reach out to us [Canso Community](https://github.com/Yugen-ai/gru/blob/main/gru_docs/gru_docs/community.md) [^1]: See [Canso Architecture Overview](/architecture) # πŸπŸ”— Canso Python Client & Web App Data Scientists, ML Engineers, Data Engineers can install the Canso Python client to interact with the Canso API Server to define and deploy ML pipelines and AI Agents. The Python client is a pip installable python library. All you need to do is a `pip install` or a `pip3 install` and you're all set to access the API. Please note that prior to using this, you must have a [Canso-compatible Kubernetes cluster](/getting-started/provision-k8s-cluster) and the [Canso Helm charts](/getting-started/canso-helm-charts) installed on the cluster. ```console pip3 install gru ``` *** * Go back to [Table of Contents](https://github.com/Yugen-ai/gru/blob/main/summary.md) * Reach out to us [Canso Community](https://github.com/Yugen-ai/gru/blob/main/gru_docs/community.md) # Health Metrics for Features in the Data Plane ## Table of Contents 1. [Introduction](#1-introduction) 2. [System Architecture Overview](#2-system-architecture-overview) 3. [New Features and Improvements](#3-new-features-and-improvements) 3.1. [Canso Agent Proxy](#31-canso-agent-proxy) 3.2. [Airflow Job Health Metrics](#32-airflow-job-health-metrics) 3.3. [Spark Streaming Health Metrics](#33-spark-streaming-health-metrics) 4. [Technical Details](#4-technical-details) 4.1. [Control Plane](#41-control-plane) 4.2. [Data Plane](#42-data-plane) 4.3. [Communication and Data Flow](#43-communication-and-data-flow) 5. [Benefits and Impact](#5-benefits-and-impact) 6. [Future Roadmap](#6-future-roadmap) ## 1. Introduction Welcome to the latest release of our Health Metrics Collection for features. This release marks a significant milestone in our efforts to provide robust, scalable, and efficient monitoring capabilities for complex distributed systems. Our focus has been on enhancing the observability of data plane architectures, with particular emphasis on Airflow jobs and Spark streaming applications. ## 2. System Architecture Overview Our system is built on a Control Plane and Data Plane architecture, designed to provide comprehensive monitoring while maintaining a clear separation of concerns. ![High-Level Design Diagram](/files/jeOerRKbjwDTw7XdcNYD) * **Control Plane**: Centralized management and monitoring hub * Houses the RabbitMQ message broker * Will include a metrics database in future releases * Responsible for processing and analyzing collected metrics * **Data Plane**: Client-side infrastructure * Hosts Airflow jobs and Spark streaming applications * New Canso Agent Proxy for efficient metrics collection ## 3. New Features and Improvements ### 3.1 Canso Agent Proxy We've introduced a new component called the Canso Agent Proxy, significantly enhancing our metrics collection capabilities without impacting core functionalities. **Key Features:** * Deployed as a separate pod within the Canso Agent Helm chart * Runs a Flask background scheduler for automated metric collection * Operates independently from the main Canso Agent pod **Technical Details:** * Implemented using Python Flask * Utilizes the APScheduler library for task scheduling * Communicates with Airflow and Prometheus for metric collection **Benefits:** * Separation of concerns: Metric collection doesn't interfere with deployment tasks * Improved reliability and scalability of the monitoring system * Flexible configuration options for collection intervals ### 3.2 Airflow Job Health Metrics We've implemented a robust system for collecting and reporting Airflow job health metrics. **Key Features:** * Collects metrics every 5 minutes * Utilizes Airflow's REST API for data retrieval * Captures comprehensive information about DAG runs and task states **Technical Details:** * Interacts with Airflow API endpoints such as `/api/v1/dags` and `/api/v1/dags/{dag_id}/dagRuns` * Processes API responses to extract relevant health information * Structures data into a standardized metric format before publishing **Metrics Collected:** * Number of active DAGs * Success/failure rates of DAG runs * Average duration of DAG runs * Task-level statistics (success rates, durations, etc.) ### 3.3 Spark Streaming Health Metrics Our new release includes advanced monitoring capabilities for Spark streaming jobs. **Key Features:** * Collects metrics every 1 minute * Leverages Prometheus for efficient metric gathering * Focuses on critical Spark driver health indicators **Technical Details:** * Uses HTTPS calls to the Prometheus query API * Employs specific PromQL queries to extract relevant Spark metrics * Processes and transforms Prometheus data into our standardized metric format **Metrics Collected:** * Streaming query progress (input rate, process rate, etc.) * Streaming state information (active queries, waiting batches, etc.) ## 4. Technical Details ### 4.1 Control Plane The Control Plane serves as the centralized hub for metric aggregation and analysis. **Components:** * RabbitMQ message broker * Configured for high availability and durability * Uses topic exchanges for flexible routing of metrics * Future: Metrics database (e.g., TimescaleDB or InfluxDB) * Will provide long-term storage and querying capabilities **Data Flow:** 1. Receives metrics from Data Plane components via RabbitMQ 2. Processes incoming messages for immediate alerting or visualization 3. Stores processed metrics in the database (future feature) ### 4.2 Data Plane The Data Plane represents the client-side infrastructure where actual workloads run. **Components:** * Airflow cluster * Runs batch processing jobs * Exposes REST API for metric collection * Spark cluster * Executes streaming jobs * Monitored via Prometheus * Canso Agent * Main pod: Handles deployment of features and AI agents * Proxy pod: Responsible for metric collection and reporting **Interaction:** * Canso Agent Proxy interacts with Airflow API and Prometheus * Collected metrics are securely transmitted to the Control Plane ### 4.3 Communication and Data Flow 1. Canso Agent Proxy initiates metric collection at specified intervals 2. Metrics are collected from Airflow API and Prometheus 3. Collected data is transformed into a standardized format 4. Metrics are published to RabbitMQ in the Control Plane 5. Control Plane services consume metrics for processing and storage ## 5. Benefits and Impact * **Improved Visibility**: Gain deep insights into the health and performance of both Airflow and Spark jobs * **Proactive Management**: Early detection of issues enables faster response times * **Scalability**: Architecture supports monitoring across multiple client infrastructures * **Minimal Overhead**: Separate proxy ensures core functionalities remain unaffected ## 6. Future Roadmap * Implementation of a metrics database in the Control Plane * Advanced analytics and machine learning for predictive maintenance * Expansion of metric collection to cover additional components For any questions, concerns, or support needs, please don't hesitate to Reach out to us [Canso Community](https://github.com/Yugen-ai/gru/blob/main/gru_docs/community.md) # Data Sources A Data Source in Canso is a reference to raw data that is either generated and owned by users or defined by users and owned by Canso's Pre-processing pipelines. Currently, Canso supports tabular data. ## Introduction Data Sources provide an abstraction over datasets owned by users, enabling: * A standardized way of declaring raw data for feature and pre-processing table calculations. * A uniform user experience when defining features, allowing Data Scientists to focus on feature logic without worrying about underlying data, DB connections, or access. * Reusability in defining and materializing features and pre-processing jobs. * Improved understanding of raw data. Data Sources can be of 2 types at the very least 1. Batch Data Sources 2. Streaming Data Sources ## Data Source Types ### Batch Data Sources Batch Data Sources are typically Data Warehouses (BigQuery, Redshift, etc.) or Object Storages (S3, GCS, etc.). In Canso currently we are supporting S3 Data Source only. ### Batch Data Source Attributes (S3 & GCS) S3 and GCS Batch Data Sources, can be described by the following attributes. These attributes define how data is stored and accessed in object storage, allowing for standardized and reusable data processing. | Attribute | Description | Example | | --------------------------- | ---------------------------------------------------------------- | ------------------------------------------ | | `data_source_name` | Unique Name of the datasource | `raw_us_orders` | | `bucket` | Bucket Name | `mycompany_data` | | `base_key` | Fixed Key Component | `raw_txns/orders/us` | | `varying_key_suffix_format` | Varying Time-based Key Component suffixed to Fixed Key Component | `%Y-%m-%d/%H` `%Y-%m-%d/%H-%M` | | `varying_key_suffix_freq` | Frequency of `varying_key_suffix_format` | `30min` `3H` `12H` `1D` | | `time_offset` | Optional time offset in seconds | `3600` | | `file_type` | File Type | `CSV`, `PARQUET` | | `description` | Description of what the data source contains | Daily Raw orders placed by users in the US | | `owner` | Team that owns the data source | `['data_engg@yugen.ai', 'sales@yugen.ai']` | | `schema_path` | Path where PySpark and BQ Schemas of the data are persisted | - | | `event_timestamp_field` | Column indicating the event time | `ordered_at` | | `event_timestamp_format` | Format of the `event_timestamp_field` | PySpark supported data/time formats | | `created_at` | Time when the Data Source was created | `datetime(2021, 1, 1, 0, 0, 0)` | The path/key to data files for a data source are obtained using * `bucket` * `base_key` * `varying_key_suffix_format` * `varying_key_suffix_freq` * `time_offset` Canso uses the concept of Data Spans, which supports various ways in which a user's data in stored in Object Storage. To understand more about how Data Spans are used, see [Data Spans](/feature-store/data-sources/dataspans) ### Streaming Data Sources Canso supports Kafka as a data source. This can be used for real-time data processing and feature generation. Kafka data sources enable: * Real-time data ingestion. * Stream processing for immediate feature extraction. * Continuous updates to machine learning models based on live data streams. ### Streaming Data Source Attributes (Kafka) | Attribute | Description | Example | | ------------------ | ------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `name` | Unique Name of the streaming datasource | `user_activity_stream` | | `description` | Description of what the streaming data source contains | Real-time user activity data from the app | | `owners` | Tenants that owns the data source | `['data_engg@yugen.ai', 'app_analytics@yugen.ai']]` | | `topic` | Kafka topic from which the data is read | `user_activity_topic` | | `schema` | Schema definition of the streaming data | `{"user_id": "STRING", "activity_type": "STRING", "timestamp": "TIMESTAMP"}` | | `timestamp_field` | Field indicating the event time | `timestamp` | | `timestamp_format` | Format of the `timestamp_field` | `yyyy-MM-dd HH:mm:ssXXX` | | `bootstrap_server` | Kafka bootstrap server | 12.345.678.910:9092 | | `read_configs` | Configuration settings for reading from Kafka | `{"datasource": "streaming_data_source", "watermark_delay_threshold": "10 seconds", "starting_timestamp": None, "starting_offsets_by_timestamp": {}, "starting_offsets": "earliest", "fail_on_data_loss": False, "include_headers": True}` | | `cloud_provider` | Cloud provider where Kafka is hosted | `AWS` | ### Special notes on attributes * `read_configs`: The standard set of attributes we support when registering a data source. Users can provide some configurations at the time of feature registration. Currently, this option is not enabled, but we will be adding support for them soon. ## Working with Data Sources Once a DataSource has been defined, it can be 1. Registered for re-usability across different teams 2. Referenced to create ML features ### Example objects of Data Sources * [S3 Data Source](https://github.com/Yugen-ai/gru/blob/c01d1f124605d927bc45312cf86fc3c232fc680a/gru/examples/s3_data_source.py#L32-L48) * [Kafka Data Source](https://github.com/Yugen-ai/gru/blob/c01d1f124605d927bc45312cf86fc3c232fc680a/gru/examples/kafka_source.py#L5-L39) ### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Move forward to see [data-sinks.md](/feature-store/data-sinks) ➑️ # Data Spans For Object-based Data Sources (S3, GCS), Canso uses the concept of Data Spans for loading data from specific keys. Underlying raw data for a data source can have different directory stuctures. ## Introduction For e.g. consider the underlying data to be present in the following directory tree ```console mycompany_bucket/raw_events/orders/ |---2023-01-01 | | |---abc1.parquet | | |---abc2.parquet |---2023-01-02 | | |---abc3.parquet | | |---abc4.parquet |---2023-01-03 | | |---abc5.parquet ... ... |---2023-03-31 | | |---abcm.parquet | | |---abcn.parquet ``` So, the complete path to `abc1.parquet` is ```console s3://mycompany_bucket/raw_events/orders/2023-01-01/abc1.parquet ``` Internally, Canso inteprets this Data Source as below - ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” | β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” | | | | | | | | | | | mycompany_bucket/ | | raw_events/orders/ | | 2023-01-01 | | | | | | | | | | | β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ | | <-------bucket-------> <-------base_key-------> <---varying_key_suffix---> | β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` The `varying_key_suffix` has 2 components - 1. Format - e.g. * `%Y-%m-%d` for `2023-01-01`, * `d=%Y-%m-%d/t=%H-%M` for `d=2023-01-01/t=00-00`, `d=2023-01-01/t=00-15`, `d=2023-01-01/t=06-30` 2. Frequency - e.g. * `30min` * `1H` * `1D` Therefore, in this particular case, `varying_key_suffix_format` and `varying_key_suffix_freq` will be as follows - ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” | varying_key_suffix_format = %Y-%m-%d | | varying_key_suffix_freq = 1D | β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### How Data Spans are used Based on the values of `varying_key_suffix_format` and `varying_key_suffix_freq` provided, Canso internally generates paths that will then be read while materializing features. So, for the above example, the following is generated - ``` s3://mycompany_bucket/raw_events/orders/2023-01-01/ s3://mycompany_bucket/raw_events/orders/2023-01-02/ s3://mycompany_bucket/raw_events/orders/2023-01-03/ s3://mycompany_bucket/raw_events/orders/2023-01-04/ s3://mycompany_bucket/raw_events/orders/2023-01-05/ s3://mycompany_bucket/raw_events/orders/2023-01-06/ s3://mycompany_bucket/raw_events/orders/2023-01-07/ s3://mycompany_bucket/raw_events/orders/2023-01-08/ ... s3://mycompany_bucket/raw_events/orders/2023-03-31/ s3://mycompany_bucket/raw_events/orders/2023-04-01/ ... ``` Now, say a feature `avg_user_spend_l7d` is registered and deployed with a daily schedule frequency i.e. it get's calculated at the beginning of a day and the AVG is based on the last 7 days of data, Canso will automatically calculate the paths based on the feature's execution time. It will load data from those paths and perform the feature computation on the loaded data. To see more examples of how Features are computed, see [Feature Materialization](broken://pages/9IXUX5R37OdWQ47JVgPJ#Feature-Materialization) Here's another example of Data Spans ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” | varying_key_suffix_format = d=%Y-%m-%d/t=%H-%M | | varying_key_suffix_freq = 30min | β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` Canso will generate the following keys in this scenario - ``` s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=00-00/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=00-30/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=01-00/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=01-30/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=02-00/ ... ... s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=23-00/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=23-30/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=00-00/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=00-30/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=01-00/ ... ``` Canso also supports an optional `time_offset` argument, which can be used to displace the paths formed above. For e.g. a `time_offset = 10*60` (10 mins) on the example above will generate the following keys ``` ... s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=00-20/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=00-50/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=01-20/ s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=01-50/ ... ... s3://mycompany_bucket/raw_events/orders/d=2023-01-01/t=23-50/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=00-20/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=00-50/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=01-20/ s3://mycompany_bucket/raw_events/orders/d=2023-01-02/t=01-50/ ... ``` # Data Sinks A Data Sink in Canso is a reference to where processed data is stored. Data Sinks can be used by multiple Batch and Streaming features to save their outputs. Currently, Canso supports two types of data sinks: S3 for offline storage and Redis for online storage. ## Introduction Data Sinks provide a standardized way to store processed data, enabling: * Efficient storage and retrieval of processed data. * Reusability across different batch and streaming features. * Simplified handling of data storage configurations for Data Scientists. * Consistent methods for saving feature and pre-processing table outputs. ## Data Sink Types ### Offline Data Sink Attributes (S3) These attributes define how processed data is stored and accessed in Data Sinks for offline storage. | Attribute | Description | Example | | ------------- | ------------------------------------------ | -------------------------------------------------------------------------------------- | | `name` | Unique Name of the data sink | `processed_sales_orders` | | `description` | Description of what the data sink contains | Processed sales orders stored for analysis | | `owner` | Team that owns the data sink | `['data_engg@yugen.ai', 'sales@yugen.ai']` | | `bucket` | Bucket Name | `mycompany_processed_data` | | `leading_key` | Fixed Key Component | `processed_txns/users` | | `file_type` | File Type | `CSV`, `PARQUET` | | `metadata` | Additional metadata about the offline sink | `{"output_mode": "append", "processing_time": "120 seconds", "output_partitions": 20}` | ### Online Data Sink Attributes (Redis) These attributes define the configuration and usage of sink for low-latency retrieval. | Attribute | Description | Example | | ------------- | -------------------------------------------- | -------------------------------------------------------------------------------------- | | `name` | Unique Name of the data sink | `user_session_data` | | `description` | Description of what the data source contains | Real-time user session data for quick access | | `owner` | Team that owns the data source | `['data_engg@yugen.ai', 'session_mgmt@yugen.ai']` | | `host` | Redis Host | `redis://192.168.1.123:6379` | | `metadata` | Additional metadata about the online sink | `{"output_mode": "append", "processing_time": "120 seconds", "output_partitions": 20}` | ### Example object of Sinks * [S3 Sink](https://github.com/Yugen-ai/gru/blob/c01d1f124605d927bc45312cf86fc3c232fc680a/gru/examples/s3_data_sink.py#L5-L19) * [Redis Sink](https://github.com/Yugen-ai/gru/blob/c01d1f124605d927bc45312cf86fc3c232fc680a/gru/examples/redis_data_sink.py#L3-L11) ## Working with Data Sinks Once a Data Sink is defined, it can be: * Registered for reusability across different teams. * Referenced by Batch and Streaming features to save their outputs. ### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Go back to [data-sources.md](/feature-store/data-sources) ⬅️ * Move forward to see [raw-feature.md](/feature-store/features/raw-feature) ➑️ * Move forward to see [derived-feature.md](/feature-store/features/derived-feature) ➑️ * Move forward to see [streaming-feature.md](/feature-store/features/streaming-feature) ➑️ # ML Features ## Features A [Feature in Machine Learning](https://en.wikipedia.org/wiki/Feature_\(machine_learning\)) is an individual measurable property or characteristic of a phenomenon. Features are used for * Creating Training Data * Inference i.e. making predictions ### Introduction Operationalising Machine Learning is complex. Even before getting to the stage of Training Data, ML teams need to work on multiple aspects. #### 1. Scheduling Features For most Machine Learning problems, features vary with time. For e.g. a user's spend in the last 7 days will keep changing with time. Therefore, Data Scientists often need to calculate their features as per certain schedules. This requires scheduling features as DAGs using workflow management platforms such as Airflow, Luigi etc. More often than not, ML teams have additional requirements that accompany scheduling. #### 1a. Monitoring Users benefit greatly from being able to * visualize historical feature job runs and lineage * go through logs * see job configuration details #### 1b. Alerting Scheduled feature jobs can fail due to multiple reasons, lack of upstream data, availability of computational resources etc. Teams want to be alerted in such cases and have the ability to re-run feature jobs with ease. #### 1c. Pipeline Best Practises Additional efforts have to be invested to make sure feature jobs are [idempotent](https://en.wikipedia.org/wiki/Idempotence). Idempotence ensures self-correction since it prevents duplication of data when pipelines fail. #### 1d. Backfills Once a feature has been scheduled to run i.e has been deployed, a Data Scientist may want these feature jobs to run for a set of days in the past. This is referred to as backfills, a common concept in Data Engineering pipelines. Backfills help Data Scientists re-play the existing feature jobs for past dates with minimal configuration changes and without having to define the feature logic from scratch. #### 2. Need for online retrieval and Mitigate train-serve skew While features are crucial for model training, they are critical for model predictions as well. Assume a model for a recommendation system was trained using multiple features, one of which was clicks on grocery items in an E-commerce website in the last 7 days `user_grocery_clicks_7d`. This feature will also be used to rank items when users show up to the website the next time. This gives rise to 2 considerations for ML Teams 1. Train-Serve Skew Training-Serving skew happens when the feature data distribution while predictions/inference differs from the distribution present in the training data. Lack of consistency in training and serving results in degraded model performance. 2. Low-latency retrieval Model predictions impact user experience. In the scenario above, we would like to recommend products to users in \~100 ms. This mandates that feature values for `user_grocery_clicks_7d` can be queried/fetched in sub-second latencies. #### 3. Re-usability & Feature Sharing Different teams/Data Scientists can end up building the same feature multiple times. Over time, this leads to duplicated pipelines and unnecessary costs incurred in compute, storage (offline and online). For an organization, this reduces productivity and increases time to production since Data Scientists start building features from scratch instead of being able to re-use existing features. ## Working with Features Canso allows users to define features in a Declarative Manner. At a high level, defining a feature involves 3 considerations #### 1. Feature Metadata Feature metadata includes * the name of the feature * a human-readable description for easier understanding * the data-source on top of which this feature needs to be calculated * the datatype * owners of the feature #### 2. Feature Logic Feature Logic is the transformation that is used to compute the feature. Transformations include commonly used aggregations such as SUM, MIN, MAX etc or row-level transformations. #### 3. Feature Scheduling Details Scheduling details include a feature's * computation schedule - common schedules are once a day, once an hour etc. * feature compute start time i.e. the time since when the feature computation should begin * Whether or not the feature's computed values should be ingested to an online store ### Register a Feature A Feature, once defined needs to be registered. Canso persists the feature's metadata, logic and scheduling details for future reference and re-use. ### Deploy a Feature When users deploy a feature, Canso creates a DAG for the feature job. This DAG is automatically scheduled and starts running based on the user's defined schedule. These DAGs compute a feature (also referred to as materialization) and these materialized values are persisted. If the user specifies that ingestion is needed, the materialized values are ingested to an online store as well. To enable online ingestion, set ```python online=True ``` while defining a Feature. ## Types of Features Canso divides Features into 2 categories, Raw and Derived. Raw Features are transformations or aggregations on data sources. Derived Features are defined on top of existing Raw Features. #### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Move forward to see [raw-feature.md](/feature-store/features/raw-feature) ➑️ * Move forward to see [derived-feature.md](/feature-store/features/derived-feature) ➑️ * Move forward to see [streaming-feature.md](/feature-store/features/streaming-feature) ➑️ # Raw ML Batch Feature A Raw Feature in the Canso platform is a fundamental component that executes ML pipelines. It processes data from registered data sources, applies feature logic, and stores the results in data sinks. ## Introduction A Raw Features are the essential building blocks for machine learning models. It provides: * Ensures seamless preparation of data for both training and inference stages. * Performs standardized predefined logics or user-defined functions (UDFs). ## Raw Feature Types ### Feature with Predefined Logic Raw Features with predefined logic utilize built-in transformations and aggregations for ease of use and consistency. Features created using common aggregations like window or sliding window and transformations such as SUM, MIN, MAX, etc. ### Feature with Custom UDF Features created using user-defined functions for more complex and specific transformations. Raw Features with custom UDFs allow for more flexibility and can handle complex transformations not covered by predefined logic. ### Raw Feature Attributes | Attribute | Description | Example | | ---------------------------------------- | ----------------------------------------------------------- | ------------------------------------------------------------------------------------------------- | | **name** | Unique Name of the raw feature | `user_clicks_7d` | | **description** | Description of what the raw feature contains | `Sum of user clicks in the last 7 days` | | **owners** | Team that owns the raw feature | `['data_team@company.com']` | | **entity** | The entity the feature is based on | `user_id` | | **data\_type** | Data type of the raw feature | `FLOAT` | | **data\_sources** | List of data sources used to create the feature | `[survey_telemetry_data]` | | **staging\_sink** | S3 sink where the intermediate data is stored | `[operational_telemetry_data]` | | **online\_sink** | Redis sink for storing the feature online | `["online_telemetry_data"]` | | **online\_sink\_write\_option\_configs** | Configurations for writing to the online sink | `{"online_telemetry_data":{ "file_type_properties": { "type": "PARQUET", "mergeSchema": False}}}` | | **feature\_logic** | Transformation logic to compute the feature | `SlidingWindowAggregation` | | **processing\_engine** | Processing engine used for feature computation | `spark` | | **processing\_engine\_configs** | Configurations for the processing engine | `{"memory": "4g", "cores": 2}` | | **online** | Flag to indicate if the feature should be available online | `True` | | **offline** | Flag to indicate if the feature should be available offline | `True` | | **schedule** | Schedule for feature computation | `1D` | | **active** | Flag to indicate if the feature is active | `True` | | **start\_time** | Start time for feature computation | `datetime.now()` | ### Special notes on attributes * `feature_logic`: The Raw Feature supports sliding windows and window aggregations. * `processing_engine & their configs`: These are the [default set of PySpark configurations](https://github.com/Yugen-ai/gru/blob/main/gru/config/features/default_processing_engine_configs_batch.yaml) used to run the Raw Feature. * `online flag`: If the online flag is enabled, the data will be ingested into the online sink (i.e., Redis cache). * `offline flag`: If the offline flag is enabled, the Raw Feature will ingest the data into the offline sink (i.e., S3 sink). * `Read & Write option configs`: Users can provide some configurations at the time of feature registration. Currently, this option is not enabled, but we will be adding support for them soon. ### Example Object of Raw Features * [Feature with Predefined logic](https://github.com/Yugen-ai/gru/blob/main/gru/examples/create_raw_feature.py#L14-L72) * [Feature with Custom UDF](https://github.com/Yugen-ai/gru/blob/main/gru/examples/create_custom_raw_feature.py#L12-L34) ### Working with Raw Features Once the raw feature is defined: * It can be registered and reused by Derived Features. The output of a Raw Feature becomes the input for a Derived Feature, enabling more complex feature engineering and reducing redundancy. * It can be deployed to execute the defined operation. ### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Go back to [data-sources.md](/feature-store/data-sources) ⬅️ * Go back to [data-sinks.md](/feature-store/data-sinks) ⬅️ * Move forward to see [register-feature.md](/guides/register-feature) ➑️ # Derived ML Batch Feature Derived Features are created by applying transformations on data generated by raw features. They serve as advanced features that build upon the foundational raw features, enabling more complex data processing and feature engineering. ## Introduction Derived Feature is an additional layer on top of raw features. It provides: * Derived Features can utilize multiple raw features as inputs, combining their values to create more meaningful and polished results. * These features are crucial in machine learning workflows as they allow for more sophisticated data transformations and enrichment. * It performs built-in operations and advanced operations on the raw tabular data. ## Derived Feature Types ### Feature with built-in operations Derived Features support a range of built-in operations such as add, subtract, multiply, and safe\_divide. These operations combine the raw data perform the transformation operation and adds new column to the dataframe having the new transformed values. ### Derived Feature Attributes | Attribute | Description | Example | | --------------------------- | ------------------------------------------------------------------------------- | --------------------------------- | | `name` | Unique name of the derived feature | `user_click_rate` | | `description` | Human-readable description for easier understanding | `Click rate of users over time` | | `staging_sink` | Data sink for staging the processed data | `recommendation-data-sink-S3` | | `online_sink` | Data sink for storing the processed data for online retrieval | `redis://192.168.1.100:6379` | | `data_type` | Data type of the derived feature | `FLOAT` | | `owners` | List of team members or teams responsible for the feature | `['data_team@company.com']` | | `schedule` | Schedule for computing the derived feature | `daily` | | `entity` | Entity to which the feature belongs | `user_id` | | `processing_engine` | Engine used for processing the feature logic | `Spark` | | `processing_engine_configs` | Configuration options for the processing engine | `{'num_partitions': 10}` | | `online` | Boolean flag indicating if the feature should be available for online retrieval | `True` | | `offline` | Boolean flag indicating if the feature should be available for offline analysis | `True` | | `transform` | Transformation logic applied to the raw feature values | `add(raw_feature1, raw_feature2)` | | `start_time` | Time since when the feature computation should begin | `2024-01-01 00:00:00` | ### Special notes on attributes * `feature_logic`: The Derived Feature supports operations like add, subtract, multiply, safe\_divide. * `processing_engine & their configs`: These are the [default set of PySpark configurations](https://github.com/Yugen-ai/gru/blob/main/gru/config/features/default_processing_engine_configs_batch.yaml) used to run the Derived Feature. * `online flag`: If the online flag is enabled, the data will be ingested into the online sink (i.e., Redis cache). * `offline flag`: If the offline flag is enabled, the Derived Feature will ingest the data into the offline sink (i.e., S3 sink). * `Read & Write option configs`: Users can provide some configurations at the time of feature registration. Currently, this option is not enabled, but we will be adding support for them soon. ### Example Object of Derived feature * [Feature with built-in logic](https://github.com/Yugen-ai/gru/blob/main/gru/examples/create_derived_feature.py#L11-L38) ### Working with Derived Features Once the derived feature is defined: * It can be used as a standalone feature that combines and transforms raw features or other data sources. * The output of a Derived Feature is used directly for machine learning model training or inference. * It cannot be reused or referenced again in other features. ### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Go back to [data-sources.md](/feature-store/data-sources) ⬅️ * Go back to [data-sinks.md](/feature-store/data-sinks) ⬅️ * Move forward to see [register-feature.md](/guides/register-feature) ➑️ # Raw ML Streaming Feature ## Streaming Feature Streaming Features enable real-time data processing by reading live data from Kafka topics. These features perform predefined or custom logic on the live data and store the results in specified data sinks. ### Introduction Streaming features operate on live data, making them crucial for real-time machine learning applications. It provides: * They allow for immediate processing and analysis of data as it arrives, enabling real-time predictions and insights. * Streaming features are especially important in scenarios where timely data processing is critical, such as fraud detection, recommendation systems, and dynamic pricing models. * By continuously updating features based on the latest data, streaming features ensure that machine learning models are always working with the most current information. * In this context, sliding window operations and custom user-defined functions (UDFs) are supported to provide flexible and powerful data processing capabilities. ## Overview of Streaming Aggregations and Transformations Streaming features enable dynamic real-time data processing by applying a variety of aggregations and transformations on live data streams. These operations are critical for creating powerful, real-time machine learning features. ### Supported Operations | Operation Type | Description | Use Case | Example | | ------------------- | --------------------------------------------------------------------------------------- | ---------------------------------------- | ------------------------------------------------------- | | Window Aggregations | Compute metrics over defined time intervals using sliding, tumbling, or session windows | Time-series analysis, Event monitoring | Calculate average transaction value per 5-minute window | | Filter | Selectively process events based on specified conditions | Data cleaning, Event filtering | Filter out transactions below threshold value | | Map | One-to-one transformation of individual events | Data normalization, Format conversion | Convert temperature from Celsius to Fahrenheit | | FlatMap | One-to-many transformation breaking down events into multiple outputs | Event decomposition, Data expansion | Split compound events into individual components | | Count | Calculate event occurrence frequency | Event frequency analysis, Usage metrics | Count user interactions per session | | Repartition | Redistribute data across specified number of partitions | Performance optimization, Load balancing | Rebalance data across processing nodes | | SelectExpr | SQL-style column transformations and filtering | Column manipulation, Data projection | Extract specific fields using SQL expressions | ### Getting Started 1. Choose the appropriate transformation type based on your use case 2. Configure the transformation parameters 3. Define your StreamingFeature with the chosen transformation 4. Register and deploy your feature using the YugenClient ### Raw Feature Types #### Feature with Predefined Logic Raw Features with predefined logic utilize built-in transformations and aggregations for ease of use and consistency. Features created using common aggregations like window or sliding window and transformations such as SUM, MIN, MAX, etc. #### Feature with Custom UDF Features created using user-defined functions for more complex and specific transformations. Raw Features with custom UDFs allow for more flexibility and can handle complex transformations not covered by predefined logic. #### Streaming Feature Attributes | Attribute | Description | Example | | --------------------------- | ------------------------------------------------------------------------------- | ------------------------------------------- | | `name` | Unique name of the Kafka feature | `real_time_click_rate` | | `description` | Description of what the streaming feature contains | `Real-time click rate of users` | | `data_type` | Data type of the Kafka feature | `FLOAT` | | `data_sources` | List of Kafka topics from which data is read | `['clicks_topic']` | | `staging_sink` | Data sink for staging the processed data | `real_time_clicks_processed_data_s3_bucket` | | `owners` | List of team members or teams responsible for the feature | `['data_team@company.com']` | | `feature_logic` | Transformation logic applied to the live data | `SlidingWindowAggregation` | | `processing_engine` | Engine used for processing the feature logic | `pyspark` | | `processing_engine_configs` | Configuration options for the processing engine | `{'parallelism': 4}` | | `online` | Boolean flag indicating if the feature should be available for online retrieval | `True` | | `offline` | Boolean flag indicating if the feature should be available for offline analysis | `True` | #### Special notes on attributes * `processing_engine & their configs`: These are the [default set of PySpark streaming configurations](https://github.com/Yugen-ai/gru/blob/main/gru/config/features/default_processing_engine_configs_streaming.yaml#L1-L20) used to run the Derived Feature. * `checkpoint_path`: Users can specify a `checkpoint_path` as part of the processing engine configuration to enable checkpointing for stateful recovery in Spark Streaming jobs. If no `checkpoint_path` is provided, the streaming job will run without checkpointing. Users are encouraged to provide a checkpoint path to ensure fault tolerance and state recovery. * For more details, refer to Spark's official documentation on [Checkpointing](https://spark.apache.org/docs/3.5.3/streaming-programming-guide.html#checkpointing). * `online`: If the online flag is enabled, the data will be ingested into the online sink (i.e., Redis cache). * `offline`: If the offline flag is enabled, the Raw Feature will ingest the data into the offline sink (i.e., S3 sink). * `Read option configs`: Users can provide some configurations at the time of feature registration. Currently, this option is not enabled, but we will be adding support for them soon. * `staging_sink_write_option_configs` and `online_sink_write_option_configs`: These are optional configurations that provide flexibility in controlling how data is written to sinks. Currently, this option is not enabled, but we will be adding support for them soon. ### Session Windows Session windows group data based on periods of activity separated by gaps of inactivity. Unlike fixed-time windows, session windows have dynamic lengths that expand as new events arrive and close when no events occur within a specified timeout period. #### When to Use Session Windows Session windows are perfect for analyzing continuous sequences of events that should be grouped together until there's a significant gap in activity. Consider these metrics: * Time spent per active session * Events processed per session * Session-based conversion rates These metrics are valuable in scenarios like user website interactions, where each user action extends the session timeout. If no new actions occur within the timeout period (e.g., 30 minutes), the session closes and a new one begins with the next action. For more details on different types of time windows in Spark Structured Streaming, see the [official documentation](https://spark.apache.org/docs/3.5.3/structured-streaming-programming-guide.html#types-of-time-windows). #### Example Object of Streaming Feature * [Feature with Predefined Logic](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature.py#L13-L43) * [Feature with Custom UDF having pyspark functional logic](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_custom_feature_spark_function.py#L10-L31) * [Feature with Custom UDF having pyspark sql logic](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_custom_feature_spark_sql.py#L10-L31) * [Real-time Maximum Salary Tracker: StreamingFeature with Window Aggregation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_window_max_aggregation.py#L12-L42) * [Real-time Median Salary Tracker: StreamingFeature with Window Aggregation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_window_median_aggregation.py#L12-L42) * [Real-time Distinct Count Salary Tracker: StreamingFeature with Window Aggregation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_window_agg_approx_count_distinct.py#L12-L42) * [StreamingFeature with Filter Transformation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_filter_transformation.py#L16-L41) * [StreamingFeature with Map Transformation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_map_transformation.py#L16-L39) * [StreamingFeature with FlatMap Transformation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_flatmap_transformation.py#L16-L39) * [StreamingFeature with Session Window Aggregation](https://github.com/Yugen-ai/gru/blob/main/gru/examples/kafka_feature_session_window_aggregation.py#L16-L42) #### Working with Kafka Features * Kafka Features are defined and registered similarly to batch features but are continuously updated as new data arrives. * This real-time processing allows for immediate updates to features, which can then be used for live model predictions or analytics. #### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Go back to [data-sources.md](/feature-store/data-sources) ⬅️ * Go back to [data-sinks.md](/feature-store/data-sinks) ⬅️ * Move forward to see [register-feature.md](/guides/register-feature) ➑️ # Custom User Defined Function Often, you may want to implement a feature logic that is not supported by the Aggregations and Transformations that Canso supports out of the box. In such cases, We support custom UDFs that offers greater flexibility, allowing you to apply complex, customized operations to your data, ensuring a more tailored and dynamic processing experience. ## How It Works * Implement your [custom feature logic](#write-custom-udf-script) by developing a Python script that follows the specified UDF contract. * Build a [custom docker image](#build-custom-docker-image) using the base Docker image provided by the canso platform. You can include your UDF script and any additional dependencies required at runtime. * Supply the alias for the image pull secrets in [processing\_engine\_configs](https://github.com/Yugen-ai/gru/blob/125ecd57355fc05237106c267386c1716d6edddf/gru/config/features/default_processing_engine_configs_streaming.yaml) to ensure that your custom Docker image is accessible during feature execution. * Use a [Python client](#define-python-client) to define the necessary arguments for your custom UDF, register it, and deploy it. * Detailed instructions for each step are provided below. {% hint style="warning" %} Currently, custom Docker images can only be pulled from DockerHub repositories, such that these images run in the Canso data plane. Image pull secrets are not supported yet. Support for private container registries (such as Amazon ECR, Google Container Registry, Azure Container Registry) and image pull secrets will be added in an upcoming release very soon. {% endhint %} ## Write Custom UDF Script ### Base Class for UDFs To implement a UDF, you must include this base class in your script. The `apply` method is where you define the custom logic, and `custom_args` allows for the dynamic passing of parameters required during feature execution. ```python from abc import ABC, abstractmethod class CustomOperationHandler(ABC): def __init__(self, custom_args): self.custom_args = custom_args @abstractmethod def apply(self): pass ``` ### Modes of Operation Canso Platform supports two modes for processing custom UDFs * **PySpark SQL Mode**: Allows implementation of custom UDFs using `Spark SQL` approach. * **PySpark Functional Mode**: Allows implementation of custom UDFs using `Spark Functional` programming approach. These modes allow you to implement simple or complex UDFs based on your requirements. ### PySpark SQL Mode * This mode is suitable for executing SQL-based transformations directly in Spark SQL. * In this mode, the DataFrame created from a registered data source is treated as a temporary table in PySpark. You can refer to the data source name directly in your custom UDF. ### Example: Simple Groupby Logic * UDF containing simple logics doesn't need `custom_args`. * For instance, `employee_salary_l3m` is a feature name, and `treasury_data` is a registered data source. ```python class SalaryAggregation(CustomOperationHandler): def apply(self): sql_query = f""" SELECT id, timestamp, SUM(salary) AS employee_salary_l3m FROM treasury_data GROUP BY id, timestamp """ return sql_query ``` ### Example: Complex Set of Operations In more complex cases, dynamic values can be retrieved from `custom_args` to customize the query logic.
Example containing a set of SQL operations ```python class ComplexSalaryAggregation(CustomOperationHandler): def apply(self): # Retrieve parameters from custom_args feature_name = self.custom_args.get("feature_name") table_name = self.custom_args.get("data_source") time_periods = self.custom_args.get("time_periods") weights = self.custom_args.get("weights") salary_increase_bonus = self.custom_args.get("salary_increase_bonus") salary_decrease_penalty = self.custom_args.get("salary_decrease_penalty") top_rank_bonus = self.custom_args.get("top_rank_bonus") top_rank_threshold = self.custom_args.get("top_rank_threshold") time_period_clauses = [ f"AVG(salary) OVER (PARTITION BY id ORDER BY timestamp ROWS BETWEEN {period-1} PRECEDING AND CURRENT ROW) AS avg_salary_l{period}m" for period in time_periods ] weighted_avg_clauses = [ f"COALESCE(avg_salary_l{period}m, 0) * {weight}" for period, weight in zip(time_periods, weights) ] sql_query = f""" WITH salary_stats AS ( SELECT id, timestamp, salary, {', '.join(time_period_clauses)}, LAG(salary) OVER (PARTITION BY id ORDER BY timestamp) AS prev_salary, RANK() OVER (PARTITION BY id ORDER BY salary DESC) AS salary_rank FROM {table_name} ), salary_changes AS ( SELECT *, CASE WHEN salary > prev_salary THEN 1 WHEN salary < prev_salary THEN -1 ELSE 0 END AS salary_change_direction, (salary - prev_salary) / NULLIF(prev_salary, 0) * 100 AS salary_change_percentage FROM salary_stats ) SELECT id, timestamp, salary AS current_salary, {', '.join(f'avg_salary_l{period}m' for period in time_periods)}, salary_rank, salary_change_direction, salary_change_percentage, ( {' + '.join(weighted_avg_clauses)} + CASE WHEN salary_change_direction = 1 THEN {salary_increase_bonus} WHEN salary_change_direction = -1 THEN {salary_decrease_penalty} ELSE 0 END + CASE WHEN salary_rank = 1 THEN {top_rank_bonus} WHEN salary_rank <= {top_rank_threshold} THEN {top_rank_bonus // 2} ELSE 0 END ) AS {feature_name} FROM salary_changes ORDER BY id, timestamp """ return sql_query ```
### PySpark Functional Mode * This mode uses PySpark's DataFrame API for operations like joins, groupBy, and windowing. * The DataFrame created from a registered data source is passed through `custom_feature_kwargs`, allowing the custom logic to use it dynamically. ### Example: Simple Groupby Logic This simple functional operation joins two DataFrames. The values are hardcoded in the `apply` method. ```python from pyspark.sql import functions as F class SalaryAggregation(CustomOperationHandler): def apply(self): data_source_df = self.custom_args.get("clicks_data") feature_name = self.custom_args.get("feature_name") result_df = (data_source_df .groupBy("id", "timestamp") .agg(F.sum("salary").alias(feature_name)) ) return result_df ``` ### Example: Complex Set of DataFrame Operations For more advanced use cases, `custom_args` can be used to dynamically adjust the operation parameters. The logic can be modular by defining helper methods.
Example containing complex functional operations ```python from pyspark.sql import functions as F from pyspark.sql.window import Window class SalaryAggregation(CustomOperationHandler): def apply(self): self.data_source_df = self.custom_args.get("custom_udf_data_source_v1") self.additional_data_df = self.custom_args.get("additional_data_source") self.feature_name = self.custom_args.get("feature_name") self._create_window_specs() joined_df = self._join_data() processed_df = self._process_df(joined_df) result_df = self._create_result_df(processed_df) return result_df def _create_window_specs(self): self.window_6m = Window.partitionBy("id").orderBy("timestamp").rangeBetween(-180, 0) self.window_12m = Window.partitionBy("id").orderBy("timestamp").rangeBetween(-365, 0) def _join_data(self): return (self.data_source_df .join(self.additional_data_df, on="id", how="left") .withColumnRenamed("credit_score", "external_credit_score") ) def _process_df(self, joined_df): return (joined_df .withColumn("salary_6m_avg", F.avg("salary").over(self.window_6m)) .withColumn("salary_12m_avg", F.avg("salary").over(self.window_12m)) .withColumn("expenses_6m_sum", F.sum("expenses").over(self.window_6m)) .withColumn("salary_expense_ratio", F.col("salary") / F.col("expenses")) .withColumn("prev_salary", F.lag("salary").over(Window.partitionBy("id").orderBy("timestamp"))) .withColumn("salary_increase", F.when(F.col("salary") > F.col("prev_salary"), 1).otherwise(0)) .withColumn("credit_utilization", F.col("credit_used") / F.col("credit_limit")) .withColumn("risk_score", F.when(F.col("salary_12m_avg") > 50000, 10) .when(F.col("salary_6m_avg") > 40000, 8) .when(F.col("credit_utilization") < 0.3, 5) .when(F.col("salary_increase") == 1, 3) .otherwise(0)) ) def _create_result_df(self, processed_df): return (processed_df .groupBy("id") .agg( (F.avg("salary_6m_avg") * 0.25 + F.avg("salary_12m_avg") * 0.15 + F.sum("expenses_6m_sum") * -0.1 + F.avg("salary_expense_ratio") * 0.15 + F.sum("salary_increase") * 0.1 + F.avg("credit_utilization") * ```
## Build Custom Docker Image * Example Dockerfile to demonstrates how to add custom Python dependencies and UDF scripts to the base image. * Base image `shaktimaanbot/canso-jobs:v0.0.1-beta` containing basic set of python libraries will be provided. ```dockerfile # Set the base image to use for the Docker image being built FROM --platform=linux/amd64 shaktimaanbot/canso-jobs:v0.0.1-beta # Set the working directory inside container WORKDIR /opt/spark/work-dir/bob # Install additional set of required Python libraries COPY extra_pip_dependencies.txt . RUN pip3 install --no-cache-dir -r extra_pip_dependencies.txt && \ rm -rf /root/.cache/pip # Copy specified set of scripts and modules from the host to current working directory COPY /my/local/path/to/custom_udf.py /opt/spark/work-dir/bob/src/v2/external_udfs/features/my/custom/path/ WORKDIR /opt/spark/work-dir/bob/src/v2/jobs # Change the permission of streaming jobs to make them executable within the container RUN chmod 777 feature_materializer.py ``` ## Define Python Client * This is how users register their custom UDF through the Python client. * User need to specify these things specific to custom UDF: * `custom_class_name`: Executable class containing logic. * `custom_file_path`: Additional path supplied in the dockerfile which contains UDF. * `docker_image`: Custom Docker Image containing UDF script with additional dependencies. * `custom_feature_args`: Feature args required by the UDF at the run time. * `mode`: `spark_sql` to execute sql logic and `spark_function` to execute functional logic. ```python streaming_feature_obj = StreamingFeature( name="employee_salary_l3m", description="Total Employee Salary Sum in the Last 3 months", data_type=DataType.FLOAT, data_sources=["treasury_data"], owners=["custom-user@domain.ai"], feature_logic=CustomFeatureLogic( custom_class_name="SalaryAggregation", custom_file_path="/my/custom/path/custom_udf_spark_sql.py", custom_docker_image="custom-repo/custom-image:v0.0.1-beta", mode = "spark_sql", custom_args={ "timestamp_field": "timestamp", "groupby_keys": ["id", "timestamp"] }, ), processing_engine=ProcessingEngine.PYSPARK_K8S, processing_engine_configs=ProcessingEngineConfigs(spark_streaming_flag=True), online_sink=["salary_agg_sink"], online=True, offline=False, ) ``` # Introduction The Canso AI Agentic System is a platform designed for deploying and managing production-grade AI agents at scale. By abstracting infrastructure complexities, it enables teams to focus on developing AI Agentic workflows while simplifying its deployment, scaling, and integration with dependent components. So you've created an AI Agent using your favourite framework. Now, you want to build an application that serves as an interface for interacting with your agent, allowing you to send prompts and receive responses seamlessly. Additionally, you aim to deploy this application on your cloud and ensure that the deployment is production grade. Beyond the application itself, your deployment must include all supporting components that the agent relies on, such as checkpoint DB, memory, feature stores etc. Traditionally, setting up such a system in a robust and scalable manner could take days or even weeks to get everything up and running. Enter **Canso AI Agentic System** - a solution designed to drastically cut the time needed to take your AI agent from development to production-grade deployment, reducing it from weeks to just hours! In addition, the Canso AI Agentic System has features to empower your AI Agent with the capability to run **long running tasks**. Generally, the tools integrated with AI agents execute simple tasks that are light weight and short lived. But what if you want your agent to be able to execute tasks that could run for hours? How about compute intensive tasks? How would you equip your agent with the capability to execute such tasks? The answer lies within Canso AI Agentic System! Enter **Canso Task Server**. ## Canso Task Server The Canso Task Server is designed to enable your AI Agent to execute **long running** or **compute intensive** tasks. The diagram below gives an overview of how this functionality works: ![Task Server Functionality](/files/JcOiEl2nXmng2sZF7Y7L) Although this involves multiple components, such as a Broker and the Canso Task Server that need to integrate with your AI Agent, the Canso AI Agentic System handles all the complexity for you. Setting up the broker and the Canso Task Server is effortless, requiring just a simple CLI command. Once done, you only need to add the tools from the [Canso Toolkit](/ai-agents/toolkit) into your AI agent, and you’re ready to go! See [Task Server](/ai-agents/concepts/task-server) for more details. ## Next Steps * Read the [Getting Started](/ai-agents/getting-started) guide to learn how to us the Canso AI Agentic System. * Read the [Quickstart](/ai-agents/quickstart) guide to develop and deploy a simple agent. ## Supported Frameworks * [LangGraph](https://langchain-ai.github.io/langgraph/) # Getting Started This guide provides a general overview of the using the Canso AI Agentic System, introducing key concepts and explaining how they interconnect to simplify your AI agent’s development and deployment process. ## Prerequisites The Canso AI Agentic System is built on the foundation of [Canso Architecture](/architecture). Before Proceeding, ensure you have: 1. A [Canso compatible Kubernetes cluster](/getting-started/provision-k8s-cluster) set up. 2. [Canso Helm charts](/getting-started/canso-helm-charts) installed on your cluster. To get started, install Gru by following the instructions [here](/getting-started/canso-py-client) ## Setting up the components Deploying an AI agent involves more than just deploying the agent itself; it also requires deploying the various components the agent depends on for its operation. These may include: 1. A [Checkpoint DB](/ai-agents/concepts/db) to save execution checkpoints, 2. A [Broker](/ai-agents/concepts/broker) and A [Task Server](/ai-agents/concepts/task-server) to support asynchronous execution of long running tasks. To set up the components, 1. define a YAML file containing the configurations for each component to be deployed. Example - config.yaml: ```yaml broker: type: redis name: my-redis checkpoint_db: type: postgres name: my-postgres size: 4Gi task_server: type: celery name: my-task-server replicas: 4 concurrency_per_replica: 1 broker_resource_name: my-redis ``` 2. Run the `gru` command to setup the components ```python gru component setup --cluster-name --config-file config.yaml ``` That's it! The components are now deployed in your cluster and ready to be integrated with your AI Agent. **Note**: You can also choose to set up the components individually by creating a separate YAML file for each component and executing the setup command with the respective files. ## Creating the project bootstrap Set up the scaffold folder for your AI agent project by executing the command: ```bash gru agent create_bootstrap ``` This will prompt you for a set of configurations for deploying your AI Agent. For eg. ```bash agent_name (Agent Name): my-agent agent_framework (Langgraph): Langgraph task_server_name: my-task-server checkpoint_db_name: my-postgres replicas (1): 1 ``` The `task_server_name` and `checkpoint_db_name` specified here correspond to the names assigned when creating these components in the previous step. This ensures the Canso AI Agentic system to connect your agent with the appropriate Checkpoint DB and Task Server. After providing the required inputs, a bootstrap project folder is generated with the following structure: ``` . β”œβ”€β”€ .dockerignore # Files to exclude from Docker build β”œβ”€β”€ .env # Environment variables for the application β”œβ”€β”€ Dockerfile # Docker build file β”œβ”€β”€ README.md # Documentation placeholder β”œβ”€β”€ config.yaml # Agent configuration settings β”œβ”€β”€ requirements.txt # Python dependencies for your agent └── src/ └── main.py # Entry point for the application ``` ## Development and Image Build Inside the created folder, define your AI agent and wrap it using the wrappers provided by Canso. All Python files should be placed inside the `src` folder, with `src/main.py` serving as the entry point for the application. In `src/main.py`, ensure your agent is wrapped with the Canso Agent Wrappers. For instance, if you’re creating the agent using Langgraph, your `src/main.py` should include something like the following: ```python from gru.agents import CansoLanggraphAgent .... Your Agent Code .... canso_agent = CansoLanggraphAgent(stateGraph=) canso_agent.run() ``` Add the environment variables needed by your agent in `.env` file and update configurations in `config.yaml` if needed. Create a Docker image and push it to a container registry: ```bash docker build -t my-agent-image:tag . docker push my-agent-image:tag ``` ## Register and Deploy Agent Run the below commands to register your agent and deploy it ```bash # Register agent gru agent register . --cluster-name --image my-agent-image:tag # Deploy agent gru agent deploy my-agent ``` Your agent is now deployed in your cluster and ready to receive prompts! ## Sending prompts to the agent To send prompts to your agent, create a JSON file containing the prompt and use the following command: ```python gru agent prompt my-agent ``` The prompt is sent to be processed by your AI Agent. ## Next Steps * Read the [Quickstart](/ai-agents/quickstart) guide to develop and deploy a simple agent in your cluster. # Quickstart This guide provides an example of setting up various AI Agentic components, as well as developing and deploying an AI Agent using the Canso AI Agentic System. We'll create a simple `sql-agent` that can execute SQL queries based on natural language prompts. ### Prerequisites Before Proceeding, ensure you have: 1. A [Canso compatible Kubernetes cluster](/getting-started/provision-k8s-cluster) set up. 2. [Canso Helm charts](/getting-started/canso-helm-charts) installed on your cluster. To get started, install Gru by following the instructions [here](/getting-started/canso-py-client) ### Setting up the components Our `sql-agent` utilizes [CansoSQLRunnerTool](/ai-agents/toolkit/sql-runner), which relies on a [Task Server](/ai-agents/concepts/task-server) to execute the SQL queries. For orchestration between the agent and the Task Server, we also need a [Broker](/ai-agents/concepts/broker). In addition, the agent uses [Checkpoint DB](/ai-agents/concepts/db) to save its state. Let us set up these components. To set up the components, we first define a YAML file with the configurations for the components. Save the YAML defined below in a file named `config.yaml`. ```yaml broker: type: redis name: my-redis checkpoint_db: type: postgres name: my-postgres size: 4Gi task_server: type: celery name: my-task-server replicas: 1 concurrency_per_replica: 1 broker_resource_name: my-redis ``` Now we run the gru command to setup the components ```bash gru component setup --cluster-name --config-file config.yaml ``` The Broker, Checkpoint DB and Task Server are now set up in your cluster. ### Creating the project bootstrap Set up the scaffold folder for our `sql-agent` project by executing the command: ```bash gru agent create_bootstrap ``` This will prompt us with a set of configurations for deploying our AI Agent. Provide inputs as specified below: ```bash agent_name (Agent Name): sql-agent agent_framework (Langgraph): Langgraph task_server_name: my-task-server checkpoint_db_name: my-postgres replicas (1): 1 ``` Once done, we get a folder `sql-agent` with the following structure: ``` sql-agent β”œβ”€β”€ .dockerignore # Files to exclude from Docker build β”œβ”€β”€ .env # Environment variables for the application β”œβ”€β”€ Dockerfile # Docker build file β”œβ”€β”€ README.md # Documentation placeholder β”œβ”€β”€ config.yaml # Agent configuration settings β”œβ”€β”€ requirements.txt # Python dependencies for your agent └── src/ └── main.py # Entry point for the application ``` ### Developing the sql-agent `src/main.py` serves a the entrypoint for our application. In this file, we define our AI Agent and wrap it with the `CansoLangraphAgent` wrapper. ```python import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langgraph.graph.message import add_messages from typing import Annotated, Literal, TypedDict from langgraph.prebuilt import ToolNode from gru.agents.framework_wrappers.langgraph.agent import CansoLanggraphAgent from gru.agents.tools.langgraph.sql_runner import CansoSQLRunnerTool from langgraph.graph import END, StateGraph, START load_dotenv() sql_tool = CansoSQLRunnerTool( db_host=os.getenv("DB_HOST"), db_port=os.getenv("DB_PORT"), db_username=os.getenv("DB_USERNAME"), db_password=os.getenv("DB_PASSWORD"), db_name=os.getenv("DB_NAME") ) tools = [sql_tool] tool_node = ToolNode(tools) model = ChatOpenAI(model="gpt-4o", temperature=0, max_tokens=None, timeout=None, max_retries=2,) model = model.bind_tools(tools) class State(TypedDict): messages: Annotated[list, add_messages] def should_continue(state: State) -> Literal["end", "continue"]: messages = state["messages"] last_message = messages[-1] if not last_message.tool_calls: return "end" else: return "continue" async def call_model(state: State): messages = state["messages"] response = await model.ainvoke(messages) return {"messages": [response]} workflow = StateGraph(State) workflow.add_node("agent", call_model) workflow.add_node("action", tool_node) workflow.add_edge(START, "agent") workflow.add_conditional_edges( "agent", should_continue, { "continue": "action", "end": END, }, ) workflow.add_edge("action", "agent") canso_agent = CansoLanggraphAgent(stateGraph=workflow) canso_agent.run() ``` This creates a simple [ReAct Agent with Langgraph](https://langchain-ai.github.io/langgraph/how-tos/react-agent-from-scratch/) that uses `gpt-4o` as the model. Feel free to replace it with any other model of your choice. Note that the details SQL DB are read as environment variables. We provide the values for these environment variables in the `.env` file. ```bash OPENAI_API_KEY= DB_HOST= DB_PORT= DB_USERNAME= DB_PASSWORD= DB_NAME= ``` Now we build the docker image for our Agent using the generated `Dockerfile` and push it to the repository. ```bash docker build -t my-sql-agent:0.0.1 . docker push my-sql-agent:0.0.1 ``` ### Registering and Deploying the sql-agent We run the below commands to register and deploy the `sql-agent` in your cluster. ```bash # Register agent gru agent register . --cluster-name --image my-sql-agent:0.0.1 # Deploy agent gru agent deploy sql-agent ``` `sql-agent` is now deployed in your cluster and ready to receive prompts! ### Prompting the sql-agent To prompt our `sql-agent`, we create a file `prompt.json` with the prompt. ```json { "messages": [ { "type": "human", "content": "Create a database table with name cars. It should have 3 columns: brand which will be a string, model which will also be a string and year which will be an integer." } ] } ``` Now we execute the gru command to prompt the agent. ```bash gru agent prompt sql-agent prompt.json ``` A table name `cars` should be created in your database! Congratulations! You have successfully created and deployed an AI Agent using Canso AI Agentic System! # Use Cases ## Introduction Canso AI Agentic Systems provides a robust platform for deploying AI agents that can automate complex workflows and decision-making processes. At its core, the platform currently offers two powerful tools: * **SQLRunnerTool**: A tool for executing SQL queries against supported databases, enabling data retrieval, analysis, and updates. It provides secure, efficient database operations with built-in connection pooling and error handling. * **KubernetesJobTool**: A tool for managing containerized workloads in isolated environments, allowing parallel processing and scalable computations. It handles resource allocation, job scheduling, and execution monitoring. > ***NOTE:*** You can also create your own custom tools using the platform's extensible framework. This allows you to tailor tools to meet your unique requirements or integrate with specialized systems. The following use cases demonstrate how these tools can be combined to build production-grade AI agent applications, focusing on risk analysis and machine learning operations. ## Risk Analysis and Fraud Detection ### Transaction Analysis AI agents can analyze transaction patterns in real-time to identify potential fraud using a combination of database queries and computational jobs: * Query historical transaction data to establish baseline behavior * Run real-time comparisons against patterns * Execute risk scoring algorithms as Kubernetes jobs * Store results for audit trails Example scenario: An agent monitors credit card transactions, using SQLRunnerTool to query recent transaction history: ```sql SELECT user_id, COUNT(*) as tx_count, AVG(amount) as avg_amount, STDDEV(amount) as std_amount, COUNT(DISTINCT merchant_category) as unique_categories, MAX(amount) - MIN(amount) as amount_range FROM transactions WHERE timestamp >= NOW() - INTERVAL '1 hour' AND user_id IN (SELECT user_id FROM high_risk_users) GROUP BY user_id HAVING COUNT(*) > 10 OR MAX(amount) > 5000 ``` The agent then uses KubernetesJobTool to run risk scoring algorithms on flagged transactions: ```yaml job: name: risk-score-calculation container: image: risk-scoring:v1 resources: memory: "2Gi" cpu: "1" env: - name: TRANSACTION_DATA value: "{{ sql_result }}" - name: RISK_THRESHOLD value: "0.85" volumeMounts: - name: risk-models mountPath: /models volumes: - name: risk-models persistentVolumeClaim: claimName: risk-model-store ``` ### Rule-Based Decision Engine Implement and manage fraud detection rules with dynamic updates and scalable processing: * Store rules in SQL databases * Execute rule evaluation in isolated containers * Scale rule processing based on transaction volume * Update rules dynamically based on new patterns Example scenario: An agent evaluates transaction rules by using SQLRunnerTool to fetch active rules: ```sql WITH rule_parameters AS ( SELECT rule_id, rule_logic, thresholds, priority, last_updated FROM fraud_rules WHERE status = 'active' AND business_unit = 'credit_cards' AND enabled = true ORDER BY priority DESC ) SELECT r.*, m.model_path, m.version FROM rule_parameters r LEFT JOIN rule_models m ON r.rule_id = m.rule_id WHERE m.status = 'deployed' ``` The agent then uses KubernetesJobTool to evaluate these rules against transaction batches: ```yaml job: name: rule-evaluation replicas: "{{ transaction_volume_scale }}" container: image: rule-engine:v2 resources: memory: "4Gi" cpu: "2" env: - name: RULES_CONFIG value: "{{ sql_result }}" - name: BATCH_SIZE value: "1000" volumeMounts: - name: rules-output mountPath: /output volumes: - name: rules-output persistentVolumeClaim: claimName: rules-evaluation-store ``` ## Machine Learning Operations ### Model Deployment Streamline model deployment process with automated validation and monitoring: * Query model performance metrics * Run model validation jobs * Execute A/B tests * Monitor deployment health Example scenario: An agent manages model deployment using SQLRunnerTool to check performance metrics: ```sql WITH model_metrics AS ( SELECT model_id, version, AVG(accuracy) as avg_accuracy, AVG(latency_ms) as avg_latency, COUNT(DISTINCT prediction_id) as prediction_count FROM model_predictions WHERE timestamp >= NOW() - INTERVAL '24 hours' GROUP BY model_id, version ) SELECT m.*, CASE WHEN avg_accuracy < 0.85 OR avg_latency > 100 THEN 'fail' ELSE 'pass' END as health_check FROM model_metrics m ``` The agent then uses KubernetesJobTool to handle model deployment: ```yaml job: name: model-deployment container: image: model-deployer:v1 resources: memory: "8Gi" cpu: "4" gpu: "1" env: - name: MODEL_ID value: "{{ model_id }}" - name: VERSION value: "{{ version }}" - name: DEPLOYMENT_TYPE value: "{{ 'canary' if is_new_model else 'full' }}" volumeMounts: - name: model-storage mountPath: /models - name: deployment-config mountPath: /config volumes: - name: model-storage persistentVolumeClaim: claimName: model-registry - name: deployment-config configMap: name: deployment-parameters ``` ## Future Enhancements The platform roadmap includes enhancements such as: * Additional built-in tools for advanced data preprocessing, real-time analytics, and integration with emerging AI frameworks # Fraud Analyst Agent The Fraud Analyst Agent is an AI-powered assistant designed to support Data Scientists and Data Engineers in their daily fraud analysis tasks. It comes equipped with capabilities such as data reconciliation, analysis, and explanation, helping streamline workflows and improve efficiency. With more capabilities coming soon, the Fraud Analyst Agent aims to make fraud detection and investigation easier than ever. ## Fraud Analyst Agent using Canso AI Agentic System The Canso AI Agentic System enables fast and seamless development and deployment of a Fraud Analyst Agent. Follow the steps below to develop, deploy and interact with a Fraud Analyst Agent. ### Prerequisites Before Proceeding, please ensure you have: 1. A [Canso compatible Kubernetes cluster](/getting-started/provision-k8s-cluster) set up. 2. [Canso Helm charts](/getting-started/canso-helm-charts) installed on your cluster. 3. Canso AI Agent Components - [Broker](/ai-agents/concepts/broker), [Checkpoint DB](/ai-agents/concepts/db) and [Task Server](/ai-agents/concepts/task-server) are set up on your cluster. To get started, install Gru by following the instructions [here](/getting-started/canso-py-client) ### Creating the project bootstrap Set up the scaffold folder for our `fraud-analyst` project by executing the command: ```bash gru agent create_bootstrap ``` This will prompt us with a set of configurations for deploying our AI Agent. Provide inputs as specified below: ```bash [1/7] agent_name (Agent Name): fraud-analyst [2/7] agent_framework (Langgraph): [3/7] version (0.0.1): [4/7] task_server_name: [5/7] checkpoint_db_name: [6/7] replicas (1): [7/7] iam_role_arn: Agent bootstrap project created successfully! ``` Once done, we get a folder `fraud-analyst` with the following structure: ``` fraud-analyst β”œβ”€β”€ Dockerfile β”œβ”€β”€ README.md β”œβ”€β”€ config.yaml β”œβ”€β”€ .env β”œβ”€β”€ requirements.txt └── src └── main.py ``` ### Developing the agent The `requirements.txt` file contains the python requirements of the projects. Let us update it with the required dependencies: ``` gru==0.0.1rc22.dev16 python-dotenv==1.0.1 langchain-openai==0.2.14 pandas==2.2.3 boto3==1.35.95 ``` `src/main.py` serves a the entrypoint for our application. In this file, we define our AI Agent and wrap it with the `CansoLangraphAgent` wrapper. ```python import os from typing import Annotated, TypedDict from dotenv import load_dotenv from io import StringIO import boto3 from gru.agents.tools.langgraph.python_arguments import PythonArgumentsTool from gru.agents.tools.langgraph import GitRepoContentRetriever, PythonCodeRunner, PythonRunStatusChecker from langgraph.graph.message import AnyMessage, add_messages from langchain_core.runnables import Runnable, RunnableConfig from langgraph.prebuilt import tools_condition from langgraph.graph import END, StateGraph, START from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool from langgraph.prebuilt import ToolNode from gru.agents import CansoLanggraphAgent import pandas as pd # This loads the environment variables from .env file. # It is recommended to keep this as the first statement in main.py and not to be removed. load_dotenv() GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") git_content_retriever = GitRepoContentRetriever(GITHUB_TOKEN) python_argument_tool = PythonArgumentsTool(GITHUB_TOKEN) python_code_runner = PythonCodeRunner(GITHUB_TOKEN) python_run_status_tool = PythonRunStatusChecker() def data_explainer_report(df): """ Generates a data explainer report for a Pandas DataFrame. Args: df (pd.DataFrame): The DataFrame to analyze. Returns: dict: A dictionary containing the data quality report. """ report = {} report["describe_stats"] = df.describe().to_dict() report["missing_values"] = df.isnull().sum().to_dict() report["duplicates"] = df.duplicated().sum() report["counts"] = { "rows": len(df), "columns": len(df.columns) } report["basics"] = { "columns": list(df.columns) } return report @tool def check_data_quality(bucket: str, file_path: str) -> str: """ use this for data quality check. Args: bucket : bucket in which the data is stored file_path: path of the file in the bucket """ try: s3_client = boto3.client('s3') csv_obj = s3_client.get_object(Bucket=bucket, Key=file_path) df = pd.read_csv(StringIO(csv_obj['Body'].read().decode('utf-8'))) report = data_explainer_report(df) response = [] response.append(f"Data Quality Report for {file_path}:") response.append(f"- Total rows: {report['counts']['rows']}") response.append(f"- Total columns: {report['counts']['columns']}") response.append(f"- Number of duplicates: {report['duplicates']}") # Report missing values if any exist missing = {k: v for k, v in report['missing_values'].items() if v > 0} if missing: response.append("\nMissing values found in columns:") for col, count in missing.items(): response.append(f"- {col}: {count} missing values") return "\n".join(response) except Exception as e: return f"Error checking data quality: {str(e)}" safe_tools = [python_argument_tool, python_run_status_tool, check_data_quality] sensitive_tools = [git_content_retriever, python_code_runner] sensitive_tool_names = {tool.name for tool in sensitive_tools} class State(TypedDict): messages: Annotated[list[AnyMessage], add_messages] class Assistant: def __init__(self, runnable: Runnable): self.runnable = runnable def __call__(self, state: State, config: RunnableConfig): while True: result = self.runnable.invoke(state) # If the LLM happens to return an empty response, we will re-prompt it # for an actual response. if not result.tool_calls and ( not result.content or isinstance(result.content, list) and not result.content[0].get("text") ): messages = state["messages"] + [("user", "Respond with a real output.")] state = {**state, "messages": messages} else: break return {"messages": result} def route_tools(state: State): """Route to different tool nodes based on the tool being called.""" next_node = tools_condition(state) # If no tools are invoked, return END if next_node == END: return END ai_message = state["messages"][-1] # Handle the first tool call (assuming single tool calls) first_tool_call = ai_message.tool_calls[0] # Route to sensitive tools if the tool name is in sensitive_tool_names if first_tool_call["name"] in sensitive_tool_names: return "sensitive_tools" return "safe_tools" model = ChatOpenAI(model="gpt-4o", temperature=0) assistant_prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are a helpful assistant capable of performing reconciliation activity using the tools provided to you." "Always check the data quality and take confirmation from the user before reconciliation." "Reconciliation is performed by running python code from a repository." "Ask for necessary inputs from the user whenever necessary." ), ("placeholder", "{messages}"), ] ) all_tools = safe_tools + sensitive_tools assistant_runnable = assistant_prompt | model.bind_tools(all_tools) graph = StateGraph(State) graph.add_node("assistant", Assistant(assistant_runnable)) graph.add_node("safe_tools", ToolNode(safe_tools)) graph.add_node("sensitive_tools", ToolNode(sensitive_tools)) graph.add_edge(START, "assistant") graph.add_conditional_edges( "assistant", route_tools, ["safe_tools", "sensitive_tools", END] ) graph.add_edge("safe_tools", "assistant") graph.add_edge("sensitive_tools", "assistant") canso_agent = CansoLanggraphAgent(stateGraph=graph, interrupt_before=["sensitive_tools"]) canso_agent.run() ``` Add the necessary environment variables to the `.env` file. ```bash GITHUB_TOKEN= OPENAI_API_KEY= ``` Now we build the docker image for our Agent using the provided `Dockerfile` and push it to the repository. ```bash docker build -t /fraud-analyst:0.0.1 . docker push /fraud-analyst:0.0.1 ``` ### Registering and Deploying the agent Run the following commands to register and deploy the agent in your cluster. ```bash # Register agent gru agent register . --cluster-name --image /fraud-analyst:0.0.1 --image_pull_secret # Deploy agent gru agent deploy fraud-analyst ``` Congratulations! The Fraud Analyst Agent is successfully deployed and is ready to work for you! ### Interacting with the agent You now interact with the agent using the `gru agent converse ` command. Here's an example conversation with the Fraud Analyst Agent: ``` % gru agent converse fraud-analyst Conversation ID: 3d1d9e67 User: hi Agent: Hello! How can I assist you today? User: perform reconciliation process using code from repository Agent: I'll be calling the tool get_git_repo_contents with following arguments: repository: Do you approve of this action? Type 'y' to continue; otherwise, explain your requested changed. User: y Agent: To perform the reconciliation process using the `recon.py` script from the repository, I need the following inputs from you: 1. **Source Bucket**: The name of the source bucket. 2. **Partner Report**: The path to the partner report file. 3. **Downloads TPAT**: The path to the downloads TPAT file. 4. **Conversion Attempts**: The path to the conversion attempts file. 5. **Destination Bucket**: The name of the destination bucket. 6. **Result Path**: The path where the result should be stored. 7. **Analysis Path**: The path for the analysis output. Please provide these details so I can proceed with running the reconciliation process. User: Agent: I'll be calling the tool run_python_code with following arguments: repository: file_path_to_execute: arguments: run_id: reconciliation_run_001 Do you approve of this action? Type 'y' to continue; otherwise, explain your requested changed. User: y Agent: The reconciliation process has been initiated. You can check the status of the run using the run ID `reconciliation_run_001`. If you need further assistance or want to check the status now, feel free to ask! User: what is the status? Agent: The reconciliation process has been successfully completed. If you need any further assistance or have any questions, feel free to ask! User: good Agent: Great! If you have any more questions or need further assistance in the future, don't hesitate to reach out. Have a wonderful day! ``` # Agent with Memory This guide will walk you through setting up and deploying an AI agent with memory capabilities using the Canso AI Agentic System. We'll create a `memory-agent` that can store and retrieve information from a vector database, enhancing its ability to maintain context across interactions. ## Prerequisites Before proceeding, ensure you have: 1. A [Canso compatible Kubernetes cluster](/getting-started/provision-k8s-cluster)set up 2. [Canso Helm charts](/getting-started/canso-helm-charts) installed on your cluster 3. Gru CLI installed (follow the instructions [here](/getting-started/canso-py-client)) ## Setting Up Components Our `memory-agent` requires several components: * A **Vector Database** for memory storage * A **Broker** for task orchestration * A **Checkpoint DB** for saving agent state * A **Task Server** for executing long-running tasks First, create a file named `config.yaml` with the following configuration: ```yaml vector_db: type: milvus name: my-vector-db size: 4Gi image_pull_secret: docker-secret-cred-agents broker: type: redis name: my-redis checkpoint_db: type: postgres name: my-postgres size: 4Gi task_server: type: celery name: my-task-server replicas: 1 concurrency_per_replica: 1 broker_resource_name: my-redis ``` Next, run the command to set up all components: ```bash gru component setup --cluster-name --config-file config.yaml ``` ## Creating the Project Bootstrap Generate the scaffold for our project: ```bash gru agent create_bootstrap ``` When prompted, provide the following inputs: ``` agent_name (Agent Name): memory-agent agent_framework (Langgraph): Langgraph task_server_name: my-task-server checkpoint_db_name: my-postgres replicas (1): 1 vector_db_name: my-vector-db ``` This creates a folder named `memory-agent` with the following structure: ``` memory-agent/ β”œβ”€β”€ .dockerignore β”œβ”€β”€ .env β”œβ”€β”€ Dockerfile β”œβ”€β”€ README.md β”œβ”€β”€ config.yaml β”œβ”€β”€ requirements.txt └── src/ └── main.py ``` ## Updating Dependencies Update `requirements.txt` with all necessary dependencies: ``` gru==0.0.1rc22.dev16 python-dotenv==1.0.1 langchain-openai==0.2.14 openai==1.14.1 boto3==1.35.95 ``` ## Developing the Memory-Enabled Agent Replace the content of `src/main.py` with the following code: ```python import os from typing import Annotated, TypedDict from dotenv import load_dotenv from gru.agents.tools.langgraph.python_arguments import PythonArgumentsTool from gru.agents.tools.langgraph import GitRepoContentRetriever, PythonCodeRunner, PythonRunStatusChecker, DataQualityTool, MemoryRetrievalTool from langgraph.graph.message import AnyMessage, add_messages from langchain_core.runnables import Runnable, RunnableConfig from langgraph.prebuilt import tools_condition from langgraph.graph import END, StateGraph, START from langchain_openai import ChatOpenAI from openai import OpenAI from langchain_core.prompts import ChatPromptTemplate from langgraph.prebuilt import ToolNode from gru.agents import CansoLanggraphAgent from gru.agents import CansoMemory from gru.agents.tools.core.vector_db.vectordb_factory import VectorDBType from gru.agents.tools.core.embeddings.embedding_factory import EmbeddingType # This loads the environment variables from .env file. # It is recommended to keep this as the first statement in main.py and not to be removed. load_dotenv() GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") OPEN_AI_KEY = os.getenv("OPENAI_API_KEY", "") # Initialize OpenAI client for embeddings embedding_client = OpenAI() # Initialize CansoMemory with embedding client memory = CansoMemory( client=embedding_client, embedding_type=EmbeddingType.OPENAI, vector_db_type=VectorDBType.MILVUS ) # Initialize the LLM model = ChatOpenAI(model="gpt-4o", temperature=0) # Initialize tools git_content_retriever = GitRepoContentRetriever(GITHUB_TOKEN) python_argument_tool = PythonArgumentsTool(GITHUB_TOKEN) python_code_runner = PythonCodeRunner(GITHUB_TOKEN) python_run_status_tool = PythonRunStatusChecker() check_data_quality = DataQualityTool() memory_retrieval_tool = MemoryRetrievalTool(memory=memory) # Group tools by sensitivity safe_tools = [python_argument_tool, python_run_status_tool, check_data_quality, memory_retrieval_tool] sensitive_tools = [git_content_retriever, python_code_runner] sensitive_tool_names = {tool.name for tool in sensitive_tools} class State(TypedDict): messages: Annotated[list[AnyMessage], add_messages] class Assistant: def __init__(self, runnable: Runnable): self.runnable = runnable def __call__(self, state: State, config: RunnableConfig): while True: result = self.runnable.invoke(state) # If the LLM happens to return an empty response, we will re-prompt it # for an actual response. if not result.tool_calls and ( not result.content or isinstance(result.content, list) and not result.content[0].get("text") ): messages = state["messages"] + [("user", "Respond with a real output.")] state = {**state, "messages": messages} else: break return {"messages": result} def route_tools(state: State): """Route to different tool nodes based on the tool being called.""" next_node = tools_condition(state) # If no tools are invoked, return END if next_node == END: return END ai_message = state["messages"][-1] # Handle the first tool call (assuming single tool calls) first_tool_call = ai_message.tool_calls[0] # Route to sensitive tools if the tool name is in sensitive_tool_names if first_tool_call["name"] in sensitive_tool_names: return "sensitive_tools" return "safe_tools" # Define the system prompt assistant_prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are a helpful assistant with memory capabilities. You can store and retrieve information " "from your memory to maintain context across conversations. " "Use the query_memory tool to search for relevant information stored in your memory. " "You can analyze data quality, run code from repositories, and perform other advanced tasks " "using the provided tools. Always ask for confirmation before running sensitive operations." ), ("placeholder", "{messages}"), ] ) # Combine all tools and create the runnable all_tools = safe_tools + sensitive_tools assistant_runnable = assistant_prompt | model.bind_tools(all_tools) # Create the graph graph = StateGraph(State) graph.add_node("assistant", Assistant(assistant_runnable)) graph.add_node("safe_tools", ToolNode(safe_tools)) graph.add_node("sensitive_tools", ToolNode(sensitive_tools)) graph.add_edge(START, "assistant") graph.add_conditional_edges( "assistant", route_tools, ["safe_tools", "sensitive_tools", END] ) graph.add_edge("safe_tools", "assistant") graph.add_edge("sensitive_tools", "assistant") # Initialize the Canso agent with memory canso_agent = CansoLanggraphAgent( stateGraph=graph, interrupt_before=["sensitive_tools"], memory=memory ) canso_agent.run() ``` ## Setting Up Environment Variables Create or update the `.env` file with the following content: ``` OPENAI_API_KEY= GITHUB_TOKEN= ``` ## Building and Pushing the Docker Image Build the Docker image and push it to your container registry: ```bash docker build -t /memory-agent:0.0.1 . docker push /memory-agent:0.0.1 ``` ## Registering and Deploying the Agent Register and deploy your agent with: ```bash # Register agent gru agent register . --cluster-name --image /memory-agent:0.0.1 --image_pull_secret # Deploy agent gru agent deploy memory-agent ``` ## Storing Initial Memory Data Before interacting with your agent, let's store some initial data in its memory. Create a JSON file named `customer_data.json`: ```json { "collection_name": "canso_domain_knowledge", "text": "Customer support policy: All premium customers receive priority support with a response time of maximum 4 hours. Standard customers receive support within 24 hours. For technical issues, we offer screen sharing sessions for premium customers. Refunds are processed within 7 business days.", "data": { "domain": "customer_support", "tags": ["support", "policy", "premium", "standard"], "metadata": { "version": "1.2", "last_updated": "2025-02-10", "created_by": "policy-team", "criticality": "high" } } } ``` Insert this data into agent's memory ```bash gru agent memory insert --agent-name memory-agent --file customer_data.json ``` This command will return the name of the memory where the data was inserted, which you can use for update and delete operations. ```bash Memory registration request queued successfully with memory name = 8a0a3e62-de36-403f-a094-606c15f5a54f ``` ## Updating existing Data in Memory With this operation, you can also update certain attributes of the data you previously added. Create a JSON file named customer\_update\_data.json: ```json { "collection_name": "canso_domain_knowledge", "memory_name": "8a0a3e62-de36-403f-a094-606c15f5a54f", "text": "Customer support policy: All premium customers receive priority support with a response time of maximum 3 hours. Standard customers receive support within 48 hours. For technical issues, we offer screen sharing sessions and dedicated agents for premium customers. Refunds are processed within 5 business days.", "data": { "domain": "customer_success", "tags": ["support", "policy", "premium", "standard", "refund"], "metadata": { "version": "1.3", "last_updated": "2025-03-15", "created_by": "success-team", "criticality": "critical" } } } ``` Update this data in the agent's memory. ```bash gru agent memory update --agent-name memory-agent --file customer_update_data.json ``` ## Delete the existing Data from Memory You can also delete the previously stored elements from memory by. ```bash gru agent memory delete --agent-name memory-agent --file customer_delete_data.json ``` ## Interacting with the Memory-Enabled Agent Start a conversation with your agent: ```bash gru agent converse memory-agent ``` Example interaction: ``` Conversation ID: abcd1234 User: What do you know about our customer support policies? Agent: I'll check my memory for information about customer support policies. I've found some relevant information in my knowledge base about our customer support policies: We have different service levels based on customer type: - Premium customers receive priority support with a maximum response time of 4 hours - Standard customers receive support within 24 hours - Premium customers are eligible for screen sharing sessions for technical issues - All refunds are processed within 7 business days This information was last updated on March 15, 2025 (version 1.2) and is considered high-criticality business information. Is there anything specific about these policies you'd like me to elaborate on? User: Can you also run a data quality check on our customer database? Agent: I'll be calling the tool check_data_quality with following arguments: bucket: customer-data-bucket file_path: customers.csv Do you approve of this action? Type 'y' to continue; otherwise, explain your requested changed. User: y Agent: [Results of data quality check would appear here] ``` ## Additional Resources * [Memory API Reference](/ai-agents/api-summary/memory-api) * [Memory Examples](/ai-agents/use-cases/examples) * [SQL Runner Tool](/ai-agents/toolkit/sql-runner) # Memory command examples This guide provides practical examples of using CansoMemory with Canso AI agents, including storing structured data, having memory-enhanced conversations, and utilizing the SQL Runner tool. ## Prerequisites Before trying these examples, make sure you have: * Deployed a Vector Database * Deployed a Canso AI Agent * Installed the GRU CLI tool ## Storing Data in Memory 1. Storing Database Table Metadata Create a JSON file with detailed table metadata: ```bash cat > customer_table_metadata.json << EOF { "collection_name": "canso_table_metadata", "data": { "table_name": "customers", "schema": "CREATE TABLE customers (\n id INT PRIMARY KEY,\n first_name VARCHAR(50) NOT NULL,\n last_name VARCHAR(50) NOT NULL,\n email VARCHAR(100) UNIQUE NOT NULL,\n phone VARCHAR(20),\n address_line1 VARCHAR(100),\n address_line2 VARCHAR(100),\n city VARCHAR(50),\n state VARCHAR(50),\n postal_code VARCHAR(20),\n country VARCHAR(50),\n status VARCHAR(20) NOT NULL DEFAULT 'active',\n customer_segment VARCHAR(20) CHECK (customer_segment IN ('standard', 'premium', 'vip', 'enterprise')),\n acquisition_source VARCHAR(50),\n lifetime_value DECIMAL(12,2),\n loyalty_points INT DEFAULT 0,\n date_of_birth DATE,\n created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,\n updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,\n last_login_at TIMESTAMP,\n CONSTRAINT valid_email CHECK (email ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$')\n);", } } EOF # Insert the table metadata using GRU CLI gru agent memory insert --agent-name sql-assistant --file customer_table_metadata.json ``` 2. Adding Detailed Table Documentation Create a JSON file with comprehensive table documentation: ```bash cat > customer_documentation.json << EOF { "collection_name": "canso_table_documentation", "data": { "table_name": "customers", "doc_type": "comprehensive", } } EOF # Insert the comprehensive documentation gru agent memory insert --agent-name sql-assistant --file customer_documentation.json ``` 3. Store domain specific knowledge Create a JSON file with domain-specific knowledge: ```bash cat > customer_segment_knowledge.json << EOF { "collection_name": "canso_domain_knowledge", "data": { "domain": "customer_segmentation", "tags": ["customers", "segmentation", "classification", "business-rules"], } } EOF # Insert the domain-specific knowledge gru agent memory insert --agent-name sql-assistant --file customer_segment_knowledge.json ``` 4. Store column metadata information Create a JSON file with the column metadata information: ```bash cat > column_metatdata_info.json << EOF { "collection_name": "canso_column_metadata", "data": { "table_column_composite_key": "alerts|severity", "table_name": "alerts", "column_name": "severity", "candidate_values": [ "high", "medium", "low" ], "metadata": {} } } EOF # Insert the column metadata information gru agent memory insert --agent-name sql-assistant --file column_metatdata_info.json ``` ## Conversation with a Memory-Enhanced Agent On the command line, you can converse with the agent by running the following command: ```bash gru agent converse sql-assistant ``` ``` Conversation ID: a1b2c3d4 User: What information do we have in our customers table? Agent: Based on the schema I have, the customers table contains: - id (INT PRIMARY KEY) - first_name, last_name (VARCHAR(50)) - email (VARCHAR(100), UNIQUE) - phone (VARCHAR(20)) - address fields (address_line1, address_line2, city, state, postal_code, country) - status (VARCHAR(20), DEFAULT 'active') - customer_segment (VARCHAR(20)) with values: standard, premium, vip, enterprise - financial metrics (lifetime_value, loyalty_points) - various timestamps (created_at, updated_at, last_login_at) This is the central table for all customer information in our CRM system and includes personal details, contact information, segmentation data, and account status. User: I need to write a query that shows me all premium and vip customers who have orders in the "SHP" status. What does SHP mean? Agent: SHP stands for "Shipped" - it means the order has left our facility with a carrier. This is one of our primary order status codes in the order processing workflow. For your query, you'll need to join the customers table with the orders table to find premium and vip customers with shipped orders. Here's the query: SELECT c.id, c.first_name, c.last_name, c.email, c.customer_segment, c.lifetime_value, o.id AS order_id, o.order_date, o.total_amount FROM customers c JOIN orders o ON c.id = o.customer_id WHERE c.customer_segment IN ('premium', 'vip') AND o.order_status = 'SHP' ORDER BY c.lifetime_value DESC, o.order_date DESC; Would you like me to run this query for you? User: Can you modify it to also show how long the orders have been in shipped status? Agent: I'll modify the query to show how long orders have been in the shipped status by calculating the difference between the current date and the date when the order status changed to "SHP". SELECT c.id, c.first_name, c.last_name, c.email, c.customer_segment, c.lifetime_value, o.id AS order_id, o.order_date, o.total_amount, osh.status_change_date AS shipped_date, CURRENT_DATE - osh.status_change_date AS days_in_shipped_status FROM customers c JOIN orders o ON c.id = o.customer_id JOIN order_status_history osh ON o.id = osh.order_id AND osh.status = 'SHP' WHERE c.customer_segment IN ('premium', 'vip') AND o.order_status = 'SHP' ORDER BY days_in_shipped_status DESC, c.lifetime_value DESC; This query: - Joins with the order_status_history table to get the date when each order changed to "SHP" status - Calculates days_in_shipped_status by subtracting the ship date from the current date - Orders results by days in shipped status (longest first) and then by customer lifetime value Based on our business rules, you might want to pay special attention to orders that have been in "SHP" status for more than 10 days, as these require investigation and customer outreach. User: Run this query and explain the business impact of the results. Agent: I'll run the query and explain the business impact of the results. Agent: I'll be calling the tool run_sql_query with following arguments: query: SELECT c.id, c.first_name, c.last_name, c.email, c.customer_segment, c.lifetime_value, o.id AS order_id, o.order_date, o.total_amount, osh.status_change_date AS shipped_date, CURRENT_DATE - osh.status_change_date AS days_in_shipped_status FROM customers c JOIN orders o ON c.id = o.customer_id JOIN order_status_history osh ON o.id = osh.order_id AND osh.status = 'SHP' WHERE c.customer_segment IN ('premium', 'vip') AND o.order_status = 'SHP' ORDER BY days_in_shipped_status DESC, c.lifetime_value DESC; Do you approve of this action? Type 'y' to continue; otherwise, explain your requested changed. User: y Agent: Here are the results of the query: | id | first_name | last_name | email | customer_segment | lifetime_value | order_id | order_date | total_amount | shipped_date | days_in_shipped_status | |-----|------------|-----------|--------------------------|------------------|----------------|----------|-------------|--------------|--------------|----------------------| | 143 | Jennifer | Martinez | j.martinez@email.com | VIP | $12,450 | 8834 | 2024-01-15 | $435.99 | 2024-01-16 | 14 | | 267 | David | Chen | d.chen@email.com | Premium | $8,920 | 8847 | 2024-01-17 | $389.50 | 2024-01-18 | 12 | | 189 | Sarah | Williams | s.williams@email.com | VIP | $9,850 | 8856 | 2024-01-20 | $298.75 | 2024-01-21 | 9 | | 312 | Michael | Johnson | m.johnson@email.com | Premium | $5,670 | 8862 | 2024-01-22 | $425.99 | 2024-01-23 | 7 | | 425 | Lisa | Taylor | l.taylor@email.com | Premium | $4,890 | 8869 | 2024-01-23 | $227.25 | 2024-01-24 | 6 | Delivery Exceptions: The first two orders (for Jennifer Martinez and David Chen) have been in shipped status for more than 10 days. According to our business rules, these require immediate investigation and customer outreach. These are high-value customers (especially Jennifer with a lifetime value of over $12,000) where delivery issues could significantly impact customer satisfaction. VIP Customer Focus: Two of the five orders with extended shipping times belong to VIP customers. Given their high lifetime value, any negative delivery experience could put substantial revenue at risk. The Jennifer Martinez order should be your highest priority. Risk Measurement: These five delayed shipments represent approximately $1,777 in order value. However, the customers they belong to represent over $41,000 in lifetime value, showing the disproportionate impact that delivery issues could have on your business. Recommended Actions: Immediate outreach to Jennifer Martinez and David Chen Expedited tracking investigation for all five orders Consider proactive compensation (e.g., partial refund, future discount) for the orders exceeding our 10-day threshold Analyze carrier performance for these specific delivery routes Would you like me to draft a query to identify which carriers are handling these delayed shipments? ``` # Concepts # Task Server The Task Server is a distributed task processing component in Canso's AI Agentic System that empowers AI agents the ability with to execute **long running** or **computationally intensive** tasks asynchronously. The diagram below illustrates how the task server integrates with your AI Agent. ![Task Server Functionality](/files/JcOiEl2nXmng2sZF7Y7L) The [tools](/ai-agents/toolkit) provided by Canso integrate seamlessly with the Task Server. All you need to do is set up the [Broker](/ai-agents/concepts/broker) and the Task Server, which involves executing a simple CLI command, and add the Canso tools to your AI Agent. ## Core Design Philosophy The Task Server implements a fundamental architectural principle: the separation between agent decision-making and task execution. This separation provides several key advantages: 1. **Clean Separation of Concerns** * Agents focus purely on decision-making and workflow orchestration * Task execution is handled independently by specialized workers * Clear boundaries between thinking (agents) and doing (tasks) 2. **Scalability and Resource Optimization** * Agent processes remain lightweight and responsive * Compute-intensive tasks are offloaded to appropriate workers * Independent scaling of agent instances and task workers 3. **Enhanced Reliability** * Task failures don't impact agent stability * Retry mechanisms are handled separately from agent logic * Better error isolation and recovery This architecture enables AI agents to orchestrate complex workflows while maintaining responsiveness and reliability, making it ideal for production deployments. ### Setting up the Task Server To set up the task server, define a YAML file: ```yaml task_server: type: celery name: task_server replicas: 4 concurrency_per_replica: 1 broker_resource_name: redis ``` The table below explains the configuration attribues: | Attribute | Description | Example | | ------------------------- | ------------------------------ | ------------------- | | `type` | Type of task server being used | `celery` | | `name` | Unique name of the task server | `agent-task-server` | | `replicas` | Number of worker replicas | `4` | | `concurrency_per_replica` | Tasks per worker | `1` | | `broker_resource_name` | Associated broker instance | `redis` | **Note**: Before setting up a task server, setting up a Broker is a prerequisite. See [Broker](/ai-agents/concepts/broker) for more details. Run the `gru` command to to set up the task server: ```bash gru component setup --cluster-name --config-file config.yaml ``` ## Tool Tips * See [Broker](/ai-agents/concepts/broker) ➑️ * Learn about [Checkpoint DB](/ai-agents/concepts/db) ➑️ * Explore [Memory](/ai-agents/concepts/conversations) ➑ # Broker Broker is a key component in the Canso AI Agentic System that facilitates orchestration between the AI Agents and the [Canso Task Server](/ai-agents/concepts/task-server). It acts as an intermediary, receiving tasks from AI Agents and distributing them across task workers for processing. ![Role of the Broker](/files/X4FRWBAMTj2WxYVJNO8b) > **Note**: Currently, only Redis is supported as a message broker. Support for RabbitMQ and Kafka will be added in future releases. ## Setting up the Broker To set up the task server, define a YAML file: ```yaml broker: type: redis name: my-redis ``` The table below explains the configuration attribues: | Attribute | Description | Example | | --------- | --------------------------- | ------------------- | | `type` | Type of broker | `redis` | | `name` | Name of the broker instance | `agent-task-broker` | Run the `gru` command to to set up the Broker: ```bash gru component setup --cluster-name --config-file config.yaml ``` ## Tool Tips * Go to [Introduction](/ai-agents/intro) ⬅️ * See [Task Server](/ai-agents/concepts/task-server) ➑️ * Learn about [Checkpoint DB](/ai-agents/concepts/db) ➑️ * Explore [Memory](/ai-agents/concepts/conversations) ➑ # Checkpoint DB The Checkpoint DB is a persistent storage component in Canso's AI Agentic Systems that enables AI agents to save their progress, recover from failures, and maintain context across interactions. By persisting state and execution progress, it ensures reliability, scalability, and auditability for production-grade AI agents. ## Introduction Checkpoint DB provides a reliable way to persist agent state and execution progress, enabling: * Business Continuity: State persistence across agent restarts and failures ensures your operations continue smoothly even after system interruptions * Conversation Intelligence: Context maintenance across multiple interactions enables more intelligent and personalized customer engagements * Operational Visibility: Progress tracking for long-running agent tasks gives you real-time insights into your AI operations * Compliance & Governance: Comprehensive audit trail of agent decisions and actions helps meet regulatory requirements and maintain accountability * Risk Mitigation: Recovery mechanisms for interrupted workflows protect against data loss and ensure business operations can resume from where they left off > **Note**: Currently, only Postgres is supported as a DB. ## Checkpoint DB Attributes These attributes define how agent state and progress data is stored and accessed in the Checkpoint DB: | Attribute | Description | Example | | --------- | -------------------------------------- | ------------------------------------------------- | | `type` | Type of database being used | `postgres` | | `name` | Unique name of the checkpoint database | `agent-checkpoints-db` | | `size` | Storage size allocation | `8Gi` | | `details` | Connection and configuration details | `{"service_url": "...", "admin_password": "..."}` | ## Use Cases Checkpoint DB serves several critical functions in AI Agentic Systems: ### 1. State Persistence * Saves agent state during execution * Maintains conversation history and context * Stores intermediate results and decisions ### 2. Recovery Management * Enables agents to resume from last saved state * Provides failure recovery mechanisms * Maintains consistency during scaling events ### 3. Audit and Tracking * Records agent actions and decisions * Maintains execution history * Enables debugging and performance analysis ## Working with Checkpoint DB ### Setup and Configuration ```yaml checkpoint_db: type: postgres name: postgres size: 8Gi ``` Use the GRU cli to setup the checkpoint DB on your cluster based on the above configs. ```bash gru component setup --cluster-name --config-file config.yaml ``` ### Integration with Agents Agents can interact with the Checkpoint DB to: * Save their current state (e.g., intermediate computation results or workflow checkpoints). * Retrieve previous context for continuity in conversations or tasks. * Record execution progress and decisions for audit trails. * Store intermediate results during long-running operations. ## Deployment Considerations When deploying Checkpoint DB: 1. **Resource Planning** * Allocate sufficient storage based on agent needs * Plan for backup storage ## Related Components * **Task Server**: Interacts with Checkpoint DB to maintain task state * **AI Agents**: Use Checkpoint DB to persist their state and progress ## Tool Tips * Go to [Introduction](/ai-agents/intro) ⬅️ * See [Task Server](/ai-agents/concepts/task-server) ➑️ * Learn about [Broker](/ai-agents/concepts/broker) ➑️ * Explore [Memory](/ai-agents/concepts/conversations) ➑️ # Conversation History # Memory ## Introduction Canso Memory is a powerful memory abstraction within the Canso AI Agentic System that enables AI agents to store, retrieve, update, and delete information from vector databases. This system serves as a long-term memory for AI agents, allowing them to maintain context across conversations and tasks. This guide will help you set up and start using Canso Memory with your Canso AI agents. CansoMemory provides: * A standardized interface for interacting with vector databases * Automatic embedding generation for text content * Support for multiple embedding types and vector database backends * Collection management for organizing different types of memory * Seamless integration with Canso AI agents ## Key Features * Persistent Memory Storage: Store information that persists across agent restarts and sessions * Semantic Search: Retrieve information based on semantic similarity rather than exact matches * Flexible Data Organization: Organize information into collections for different use cases * Simple API: Store and retrieve memory with just a few lines of code * Integration with AI Agents: Seamlessly incorporate memory capabilities into your Canso AI agents ## Supported Technologies | Component | Currently Supported | | ---------------- | ------------------- | | Vector Databases | Milvus | | Embedding Models | OpenAI Embeddings | ## What is Collection in Memory ? A collection in Milvus DB is similar to a table in traditional databases. It's a logical grouping of data entities that serves as the basic unit for data management. Collections help organize and store related information in a structured way that allows for efficient vector similarity search, which is essential for AI Agent memory. ## Why Do We Need Collections ? These collections, for the time being are being used for Text to SQL - helping the agent convert natural language questions into SQL queries by leveraging structured knowledge. Collections allow the AI agent to: 1. **Store and organize different types of information** - Each collection is designed to hold specific types of data relevant to the agent's operations. 2. **Perform semantic search** - By storing vector embeddings alongside text data, the agent can find information based on meaning, not just keywords. 3. **Maintain context awareness** - Collections help the agent understand the database structure, domain knowledge, and query patterns needed to generate accurate responses. ## Supported Collections Below is a summary of the collections we currently support, with detailed field specifications for each: ### 1. Table Metadata Collection **Purpose**: Stores information about your database structure to help the agent understand the schema and generate accurate SQL queries. **Collection Name**: `canso_table_metadata` | Field Name | Data Type | Notes | | ------------------- | -------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `table_name` | `VARCHAR` | Primary key, max length 200 characters, i.e., `"customers"` | | `schema` | `VARCHAR` | JSON representation of table schema, max length 65535 characters, i.e., `{"columns": [{"name": "customer_id", "type": "INT"}, {"name": "name", "type": "VARCHAR"}]}` | | `schema_embeddings` | `FLOAT_VECTOR` | Vector embeddings of table\_name, schema information, dimension depends on embedding model, i.e., `[0.1, 0.2, ..., 0.5]` | **Index**: `IVF_FLAT` with `L2` metric type on schema\_embeddings field [Click here](https://milvus.io/docs/index.md?tab=floating) to learn more about index in Milvus DB ### 2. Domain Knowledge Collection **Purpose**: Contains context-specific information about your business domain to enable the agent to understand domain-specific concepts and translate them into SQL. **Collection Name**: `canso_domain_knowledge` | Field Name | Data Type | Notes | | ------------- | -------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `fact` | `VARCHAR` | Primary key, max length 200 characters, i.e., `"Premium customers receive 10% discount"` | | `explanation` | `VARCHAR` | Detailed explanation of the domain fact, max length 65535 characters, i.e., `"Our loyalty program offers a 10% discount to all customers with Premium status"` | | `logic` | `VARCHAR` | Business logic related to the fact, max length 65535 characters, i.e., `"IF customer.status = 'Premium' THEN apply_discount(0.1)"` | | `embeddings` | `FLOAT_VECTOR` | Vector embeddings of domain knowledge, fact, explanation, logic, dimension depends on embedding model, i.e., `[0.4, 0.1, 0.8, ..., 0.3]` | **Index**: `IVF_FLAT` with `L2` metric type on embeddings field ### 3. Example Queries Collection **Purpose**: Stores successful query patterns and examples to help the agent learn from past interactions and improve future query generation. **Collection Name**: `canso_examples` | Field Name | Data Type | Notes | | ------------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | `name` | `VARCHAR` | Primary key, max length 200 characters, i.e., `"monthly_sales_report"` | | `description` | `VARCHAR` | Description of the example query, max length 65535 characters, i.e., `"Query to generate monthly sales report by product category"` | | `content` | `VARCHAR` | Actual query content, max length 65535 characters, i.e., `"SELECT category, SUM(amount) FROM sales GROUP BY category ORDER BY SUM(amount) DESC"` | | `embeddings` | `FLOAT_VECTOR` | Vector embeddings of example queries, dimension depends on embedding model, i.e., `[0.7, 0.2, 0.1, ..., 0.6]` | **Index**: `IVF_FLAT` with `L2` metric type on embeddings field ### 4. Column Metadata Collection **Purpose**: Stores metadata information for the columns in your database including possible values for columns with low cardinality. This helps the agent to generate more accurate queries when exact values need to be used in queries. The collection also has an optional metadata field which can be used to provide additional information like aliases, synonyms for column values etc. **Collection Name**: `canso_column_metadata` | Field Name | Data Type | Notes | | ---------------------------- | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `table_column_composite_key` | `VARCHAR` | Primary key; A composite of `table_name` and `column_name` for identification; max length 400 characters; Ex `"alerts\|severity"` | | `table_name` | `VARCHAR` | Name of the table; max length 200 characters, Ex "alerts" | | `column_name` | `VARCHAR` | Name of the column; max length 200 characters, Ex "severity" | | `candidate_values` | `ARRAY` | Array of candidate values; each element is VARCHAR, max\_length=100, max\_capacity=2048; Ex \["high","low","medium"] | | `metadata` | `JSON` | Stores additional metadata in JSON format. Since this is a JSON field it can be used as a catchall for any additional info; Ex {"aliases": \["acceptable", "mid"]} | | `embeddings` | `FLOAT_VECTOR` | Vector embeddings of a string concatenation of `table_name` and `column_name`, dimension depends on embedding model; Ex \`\[0.7, 0.2, 0.1, ..., 0.6] | **Index**: `IVF_FLAT` with `L2` metric type on embeddings field ## Collection Limitations and Some Notes 1. **Field Length**: Maximum length for any VARCHAR field in `Milvus DB` is `65535` characters. Similarly for an ARRAY field in `Milvus DB` we can have a maximum of `2048`elements. 2. **Primary Keys**: Each collection has a designated primary key field for unique identification. 3. **Vector Embeddings**: Each collection contains at least one vector field that stores the semantic representation of text data. 4. **Index Types**: Collections use the `IVF_FLAT` index type, which balances search speed and recall rate. ## Memory Tool Integration CansoMemory integrates with various tools including: * Memory Retrieval Tool: Allows agents to search and retrieve relevant memory during conversations * Text to SQL Tool: Uses memory to generate user-specific SQL queries based on schema information and examples stored in memory ## Prerequisites * Access to a supported vector database (currently Milvus) * Appropriate credentials for embedding generation (OpenAI API key) ## Setting Up the Vector Database {% hint style="warning" %} > Before using CansoMemory, you need to deploy a vector database to store the embeddings. Canso currently supports Milvus as its vector database backend. This can also be an external vector database. > {% endhint %} 1. Configure the Vector Database Create a configuration file (e.g., config.yaml) with the following content: ```yaml vector_db: type: milvus name: canso-prod-vdb-4-feb-v2 size: 4Gi image_pull_secret: docker-secret-cred-agents ``` 2. Deploy the Vector Database Use the Canso CLI to deploy the vector database to your cluster: ```bash gru component setup --cluster-name --config-file config.yaml ``` This command will provision a Milvus instance in your cluster with the specified configuration. ## Basic Setup ### 1. Initialize a Memory Instance ```python from gru.agents import CansoMemory from langchain_openai import ChatOpenAI # Initialize the model model = ChatOpenAI(model="gpt-4o", temperature=0) # Create a memory instance memory = CansoMemory(client=model.client) ``` ### 2. Connect Memory to an Agent ```python from gru.agents import CansoLanggraphAgent # ... agent setup code ... # Connect memory to the agent canso_agent = CansoLanggraphAgent( stateGraph=graph, memory=memory, interrupt_before=["sensitive_tools"] ) # Run the agent canso_agent.run() ``` ### Deploying the Agent Register and Deploy the Agent Using the [Deployment Documentation](/ai-agents/getting-started#register-and-deploy-agent) ## Using the CLI for Memory Management CansoMemory can also be managed using the Canso CLI: ### Storing Memory ```bash gru agent memory insert --agent-name my-agent --file data_file.json ``` Where `data_file.json` contains: ```json { "collection": "sql_schemas", "data": {"text": "Table: customers, Description: Stores customer data", "type": "schema"}, "tags": ["SQL", "database"] } ``` ### Updating Memory ```bash gru agent memory update --agent-name my-agent --file memory_file.json ``` ### Deleting Memory ```bash gru agent memory delete --agent-name my-agent --expr --collection ``` ### Converse with Agent with Memory Enhanced Context ```bash gru agent converse --agent-name my-agent ``` ## Detailed Documentation For more detailed information about CansoMemory, please refer to: * [Memory API Reference](/ai-agents/api-summary/memory-api) * [Examples](/ai-agents/use-cases/examples) # How Tos * [Update the AI Agent](/ai-agents/how-tos/update-agent) * [Delete the AI Agent](/ai-agents/how-tos/delete-agent) # Update the AI Agent All the configurations in the AI Agent's [config.yaml](/ai-agents/getting-started#creating-the-project-bootstrap) file along with the AI agent's container image can be updated using the `gru agent update` command. To update the configurations in the `config.yaml` file, make the changes in the file and execute the following command: ```bash gru agent update . ``` To update the container image of the AI Agent, execute the following command: ``` gru agent update . --image= ``` **Note:** `.` represents the context path of the AI Agent folder. Use the path of the AI agent folder if the command is being executed from a different directory. # Delete the AI Agent To delete an AI Agent deployed in your cluster, execute the following command: ```bash gru agent delete ``` **Note:** Please not that the above command deletes only the AI Agent and not its dependent components. # Toolkit Canso Toolkit provides you with tools that you can integrate with your AI Agent. These tools work seamlessly with the [Canso Task Server](/ai-agents/concepts/task-server) to execute long running or compute intensive tasks. They serve as the interface that enables your AI Agent to leverage the capabilities of the task server. Available tools: * [SQL Runner](/ai-agents/toolkit/sql-runner) * [Kubernetes Job Runner](broken://pages/gevaokF9PzqAPLiVJyfG) # SQL Runner The SQL Runner Tool (CansoSQLRunnerTool) is a tool in Canso toolkit that enables AI agents to execute SQL queries against databases. It delegates the execution of the SQL queries to the [Task Server](/ai-agents/concepts/task-server), where the queries are executed, and results are returned to the AI Agent. ### Usage The constructor for the CansoSQLRunnerTool has following parameters: | Parameter | Description | Example | | ------------- | -------------------- | ------------------ | | `db_host` | Database hostname | `"db.example.com"` | | `db_port` | Database port | `"5432"` | | `db_username` | Database username | `"db_user"` | | `db_password` | Database password | `"password123"` | | `db_name` | Target database name | `"my_database"` | The input parameters for the tool, provided at runtime by the workflow or the agent, include: | Parameter | Description | Example | | --------- | -------------------- | ----------------------- | | `query` | Query to be executed | `"select * from table"` | The following code snippet illustrates how the CansoSQLRunnerTool can be integrated with your AI Agent. ```python from gru.tools import CansoSQLRunnerTool from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode sql_tool = CansoSQLRunnerTool( db_host="db.example.com", db_port="5432", db_username="db_user", db_password="password123", db_name="my_database" ) tools = [sql_tool] tool_node = ToolNode(tools) model = ChatOpenAI(model="gpt-4o", temperature=0, max_tokens=None, timeout=None, max_retries=2,) model = model.bind_tools(tools) ``` ## Tool Tips * Go to [Toolkit](/ai-agents/toolkit) ⬅️ * See [Kubernetes Job Tool](broken://pages/gevaokF9PzqAPLiVJyfG) ➑️ * Learn about [Task Server](/ai-agents/concepts/task-server) ➑️ * Explore [Examples](https://github.com/Yugen-ai/gru/blob/main/gru_docs/ai-agents/examples.md) ➑️ # Kubernetes Job # Text-to-SQL The Text-to-SQL Tool (`TextToSQLTool`) is a powerful component in the Canso toolkit that enables AI agents to convert natural language queries into valid SQL statements. It leverages underlying language models to understand user intent and generate appropriate SQL code based on database schema information. ## Overview The Text-to-SQL Tool bridges the gap between natural language understanding and database querying, allowing agents to: * Parse natural language questions about data * Convert these questions into valid, optimized SQL queries * Maintain context about database schemas and relationships * Leverage examples and domain knowledge for better query generation ### Usage The constructor for the `TextToSQLTool` has the following parameters: | Parameter | Description | Required | | --------- | --------------------------------------------------------------------------- | -------- | | `service` | An instance of `TextToSQLService` with configured LLM and context retriever | Yes | The input parameters for the tool, provided at runtime, include: | Parameter | Description | Example | Required | | ------------ | ----------------------------------------------------------------- | ------------------------------------------------------- | -------- | | `query` | Natural language query to convert to SQL | "Show me all customers who made purchases last month" | Yes | | `table_info` | Optional table schema information to initialize or update context | `[{"table": "customers", "schema": "CREATE TABLE..."}]` | No | ## Integration The following code snippet illustrates how the `TextToSQLTool` can be integrated with your AI Agent: ```python from gru.agents.tools.text_to_sql import TextToSQLTool from gru.agents.tools.core.services.text_to_sql import TextToSQLService from gru.agents.tools.core.llm_client.openai import OpenAILLMClient from gru.agents.tools.core.context_retriever.sql import SQLContextRetriever from gru.agents.tools.core.vector_db.milvus import MilvusClient from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode # Initialize dependencies llm_client = OpenAILLMClient(model="gpt-4o") vector_store = MilvusClient() context_retriever = SQLContextRetriever(vector_store=vector_store) # Create the service text_to_sql_service = TextToSQLService( llm_client=llm_client, context_retriever=context_retriever ) # Initialize the tool text_to_sql_tool = TextToSQLTool(service=text_to_sql_service) # Add to your agent's tools tools = [text_to_sql_tool] tool_node = ToolNode(tools) model = ChatOpenAI(model="gpt-4o", temperature=0) model = model.bind_tools(tools) ``` ## Memory Integration The `TextToSQLTool` can leverage CansoMemory to: 1. Store and retrieve database schemas 2. Save example queries and their SQL translations 3. Learn from past query conversions 4. Maintain domain-specific knowledge related to databases Memory-enhanced Text-to-SQL transformations become increasingly accurate as the system learns from more examples and builds a richer context of your database structure. ## Example When a user asks a natural language question: ``` "Which customers spent more than $1000 last quarter and haven't made a purchase this month?" ``` The `TextToSQLTool` can generate a SQL query like: ```sql SELECT c.customer_id, c.first_name, c.last_name, c.email FROM customers c JOIN ( SELECT customer_id, SUM(total_amount) as quarterly_spend FROM orders WHERE order_date BETWEEN '2024-01-01' AND '2024-03-31' GROUP BY customer_id HAVING SUM(total_amount) > 1000 ) q ON c.customer_id = q.customer_id WHERE c.customer_id NOT IN ( SELECT DISTINCT customer_id FROM orders WHERE order_date >= '2024-04-01' ) ORDER BY q.quarterly_spend DESC; ``` ## Tool Tips * Go to [Toolkit](/ai-agents/toolkit) ⬅️ * See [SQL Runner Tool](/ai-agents/toolkit/sql-runner) ➑️ * Learn about [Memory](https://github.com/Yugen-ai/gru/blob/main/gru_docs/ai-agents/memory.md) ➑️ * Explore [Examples](/ai-agents/use-cases/examples) ➑️ # API Documentation ## AI Agents * [Agent](/ai-agents/api-summary/api-doc) * [Memory](/ai-agents/api-summary/memory-api) # Agent ## Setup Agentic Components {% openapi src="/files/RKbQOHjASipz3j1VTfQV" path="/v1/components/" method="post" expanded="true" %} [openapi-ai-agents.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8cec56765ae1ea7389e24bca1c02136299a795ab%2Fopenapi-ai-agents.json?alt=media) {% endopenapi %} ## Register Agent {% openapi src="/files/RKbQOHjASipz3j1VTfQV" path="/v1/agent/register" method="post" expanded="true" %} [openapi-ai-agents.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8cec56765ae1ea7389e24bca1c02136299a795ab%2Fopenapi-ai-agents.json?alt=media) {% endopenapi %} ## Deploy Agent {% openapi src="/files/RKbQOHjASipz3j1VTfQV" path="/v1/agent/{agent\_name}/deploy" method="post" expanded="true" %} [openapi-ai-agents.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8cec56765ae1ea7389e24bca1c02136299a795ab%2Fopenapi-ai-agents.json?alt=media) {% endopenapi %} ## Update Agent {% openapi src="/files/RKbQOHjASipz3j1VTfQV" path="/v1/agent/{agent\_name}" method="patch" expanded="true" %} [openapi-ai-agents.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8cec56765ae1ea7389e24bca1c02136299a795ab%2Fopenapi-ai-agents.json?alt=media) {% endopenapi %} ## Delete Agent {% openapi src="/files/RKbQOHjASipz3j1VTfQV" path="/v1/agent/{agent\_name}" method="delete" expanded="true" %} [openapi-ai-agents.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8cec56765ae1ea7389e24bca1c02136299a795ab%2Fopenapi-ai-agents.json?alt=media) {% endopenapi %} # Memory ## Store Memory {% openapi src="/files/6nxtXgCnE8uXqVUIcoEn" path="/memory" method="post" expanded="true" %} [openapi-ai-memory.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-9737fe9d675f1388da3d71f4f1207d75c6307d51%2Fopenapi-ai-memory.json?alt=media) {% endopenapi %} ## Retrieve Memory {% openapi src="/files/6nxtXgCnE8uXqVUIcoEn" path="/memory" method="get" expanded="true" %} [openapi-ai-memory.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-9737fe9d675f1388da3d71f4f1207d75c6307d51%2Fopenapi-ai-memory.json?alt=media) {% endopenapi %} ## Update Memory {% openapi src="/files/6nxtXgCnE8uXqVUIcoEn" path="/memory" method="patch" expanded="true" %} [openapi-ai-memory.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-9737fe9d675f1388da3d71f4f1207d75c6307d51%2Fopenapi-ai-memory.json?alt=media) {% endopenapi %} ## Delete Memory {% openapi src="/files/6nxtXgCnE8uXqVUIcoEn" path="/memory" method="delete" expanded="true" %} [openapi-ai-memory.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-9737fe9d675f1388da3d71f4f1207d75c6307d51%2Fopenapi-ai-memory.json?alt=media) {% endopenapi %} ## List Collections {% openapi src="/files/6nxtXgCnE8uXqVUIcoEn" path="/memory/collections" method="get" expanded="true" %} [openapi-ai-memory.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-9737fe9d675f1388da3d71f4f1207d75c6307d51%2Fopenapi-ai-memory.json?alt=media) {% endopenapi %} ## Get Collection Info {% openapi src="/files/6nxtXgCnE8uXqVUIcoEn" path="/memory/collections/{collection\_name}" method="get" expanded="true" %} [openapi-ai-memory.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-9737fe9d675f1388da3d71f4f1207d75c6307d51%2Fopenapi-ai-memory.json?alt=media) {% endopenapi %} # Overview ## Introduction Fraud detection & prevention is a combination of tools, technologies and processes to identify and prevent dishonest activities. Fradulent behaviour results in financial losses, disruption in operations and often reputational damage. In banking, this means spotting unusual transactions or behaviors that could point to scams like account takeovers or fake charges. For rewarded ads, fraud detection involves identifying bots pretending to be real users to steal rewards or fake ad clicks to drain advertiser budgets. ### Fraud Detection Systems The first step in combating fraud is to detect it. Fraud detection systems are usually of 2 types, depending on when the system does the detection 1. (Near) Real Time i.e. evaluating a transaction or an event very close to its actual occurence. 2. Post-Event or Historical analysis i.e. investigation of historical data to figure out anomalies. ### Fraud Detection Techniques Fraud detection systems use one or more of these techniques - * Rule-based Systems * Machine Learning Systems (AutoEncoders, GNNs etc) The Canso Platform currently supports Rule based fraud detection systems. Users can easily perform batch & real-time feature engineering, design workflows and define rules within the workflow leveraging features. Real world workflows include detecting Account takeover, UPI Fraud, Money Mules, Bot Fraud and more. Each workflow consists of certain rules. For the Account Takeover Workflow, these could be * login from a new device and country or from a high risk device * VPN detection * No. of failed login attempts in the last 3 hours is greater than 10 or No. of failed login attempts in the last 1 hours is greater than 5 * No. of attributes changes in a user's profile/account settings over the last 12 hours exceeds 3 * behavioral anomaly such as 2x increase in typing speed or click velocity. Canso's Python Client makes it easy for users to develop fraud detection solutions with seamless access to the following key services - * **Workflow Management**: Create worfklows and define rules to detect fraud. * **Feature Management**: Develop & manage features, i.e. aggregated historical behavioral data, for use in rules and decision-making. * **Risk Decisioning Engine**: Deploy workflows as configurable, high performance applications that can evaluate incoming transactions in sub-100 millisecond latency at scale. An exciting release we are working hard to ship in the next couple of months is to build AI agents so that end users do not have to worry about writing code to define rules and workflows. Using simple natural language, you can express business rules, simple and complex, and the AI will figure out the rest. The AI agent is designed to respect and value human inputs and will prioritise human feedback at different stages. *** ## Canso Fraud Management Architecture ![Architecture](/files/UqWAjazmQeIvXknXYIvB) *** ## Dive Right In \
🎯 Risk Workflow ManagementDefine and Manage Risk Workflows and rules/files/X77iRF1IROMHTqCwZ1Qfhttps://docs.canso.ai/risk/workflows
⚑ Real-time Transaction MonitoringDeploy Workflows as real-time services to evaluate transactions/files/X77iRF1IROMHTqCwZ1Qfhttps://docs.canso.ai/risk/txn-evaluation
🧠 Fraud AI AgentsComing Soon!/files/X77iRF1IROMHTqCwZ1Qfhttps://docs.canso.ai/ai-agents/intro
# Workflows and Rules ## Overview The Risk Workflow Management CLI provides tools for managing fraud detection workflows and rules. All commands follow the pattern `canso workflows ...` ## Prerequisites * Set `GRU_TOKEN` environment variable for authentication * Valid JSON configuration files for workflow and rule creation/updates ## Workflow Management Commands ### List Workflows ```bash # List all workflows canso workflows list # List only active workflows canso workflows list --is_active=true # Get specific workflow details canso workflows list --workflow_name= ``` ### Create Workflow ```bash canso workflows create --config ``` Sample workflow\_config.json: ```json { "description": "Workflow for detecting credit card fraud", "initial_rules": [ { "rule_name": "amount_limit_check", "operator": "AND", "sub_rules": [ { "field": "amount", "operator": "<", "redis_key": "transaction_limit", "redis_field": "user:{user_id}" } ] } ] } ``` ### Update Workflow Status ```bash canso workflows update-status --status [ACTIVE|INACTIVE|DEPRECATED] ``` {% hint style="info" %} > Workflow status changes are tracked only in the control plane database. Future versions may implement rollouts in the data plane via ARGOCD. > {% endhint %} ### Deploy Workflow ```bash canso workflows deploy --env [BACKTESTING|STAGING|PRODEXPERIMENT|LIVE] --cluster_name --namespace ``` The following environment variable can be set using the --env\_vars flag: ``` FEATURE_STORE_HOST: str FEATURE_STORE_PORT: int FEATURE_STORE_DB: int FEATURE_STORE_USERNAME: str FEATURE_STORE_PASSWORD: str ``` NOTE: By default, the feature store is connected to the deployed Redis instance along with the Helm chart. Example: ```bash --env_vars FEATURE_STORE_HOST=localhost,FEATURE_STORE_PORT=6379,FEATURE_STORE_DB=1,FEATURE_STORE_USERNAME=admin,FEATURE_STORE_PASSWORD=admin ``` Providing HPA Configuration You can customize the Horizontal Pod Autoscaler (HPA) settings by providing your own configuration file with the --hpa\_configs flag. Below is an example of what the configuration file look like: Sample hpa\_config.json: ```json { "min_replicas": 1, "max_replicas": 8, "target_cpu_utilization_percentage": 80, "target_memory_utilization_percentage": 80 } ``` To deploy a workflow with your custom HPA settings, use the following command: ```bash canso workflows deploy --env [BACKTESTING|STAGING|PRODEXPERIMENT|LIVE] --cluster_name --namespace --hpa_configs ``` If you don’t provide an HPA configuration file, the system will automatically use the default settings. ## Rule Management Commands ### List Rules ```bash # List all rules in a workflow canso workflows rules list # List rules with specific stage canso workflows rules list --stage # Get specific rule details canso workflows rules list --rule_name ``` Available stages: `REGISTERED`, `BACKTESTING`, `STAGING`, `PRODEXPERIMENT`, `LIVE` ### Create Rule ```bash canso workflows rules create --config ``` Sample rules\_config.json: ```json { "rule_name": "amount_limit_check_2", "operator": "AND", "sub_rules": [ { "field": "amount", "operator": "<", "redis_key": "transaction_limit", "redis_field": "user:{user_id}" } ] } ``` ### Update Rule ```bash # Update rule definition canso workflows rules update --config # Update rule status canso workflows rules update --status [ACTIVE|INACTIVE|DEPRECATED] # Update rule stage canso workflows rules update --stage [REGISTERED|BACKTESTING|STAGING|PRODEXPERIMENT|LIVE] ``` Sample rule\_def.json: ```json { "operator": "AND", "sub_rules": [ { "field": "amount", "operator": "<", "redis_key": "transaction_limit", "redis_field": "user:{user_id}" } ] } ``` ## Common Fields ### Rule Operators * `AND`: All sub-rules must pass * `OR`: At least one sub-rule must pass ### Sub-Rule Operators * `<`: Less than * `>`: Greater than * `==`: Equal to * `!=`: Not equal to * `>=`: Greater than or equal to * `<=`: Less than or equal to ### Rule Stages 1. `REGISTERED`: Initial state for new rules 2. `BACKTESTING`: Under testing with historical data 3. `STAGING`: Testing in non-production environment 4. `PRODEXPERIMENT`: Limited production testing 5. `LIVE`: Active in production ### Status Values * `ACTIVE`: Rule/workflow is enabled * `INACTIVE`: Rule/workflow is disabled * `DEPRECATED`: Rule/workflow is no longer in use # Real Time Transaction Monitoring ## Overview The Risk Rule Evaluation Service provides the ability to evaluate transaction in real-time using a REST API. Whenever a workflow is [deployed](/risk/workflows#deploy-workflow), the Canso platform automatically packages all rules in the workflow and spins up the application. This application runs on the data plane i.e. the customer's Kubernetes cluster. The Canso Developer Agent automates the process of deploying the application and any related infra dependencies. Once deployed, this application starts accepting requests to evaluate incoming transactions against eligible workflows rules defined as part of the [Workflow](/risk/workflows#create-workflow). The application is capable of supporting a double-digit millisecond latency (P99 of <100 ms) in most cases and can autoscale by itself within the limits set at the time of deployment. The Txn. Monitoring & Evalution app can be configured such that it integrates with Online Feature stores to retrieve feature values. ## Prerequisites To interact with the Risk Evaluation API, ensure you have the following: 1. **Ingress Host and Path**: * Identify the ingress host and path for your service. * **Ingress Host**: You can find the ingress host by running the following command on your Kubernetes cluster: ```bash kubectl get ingress -n -o yaml ``` Look for the `HOSTS` column to find the appropriate ingress host. * **Ingress Path**: Ensure you also note the `PATH` associated with your service in the ingress output. * Replace `{YOUR_INGRESS_HOST}` and `{YOUR_INGRESS_PATH}` in your API calls with the values retrieved from the above command. 2. **Authentication**: Basic Authentication is required. Use your Canso credentials: * **Authorization Header**: `Authorization: Basic {BASE64_ENCODED_CREDENTIALS}` ## How Rule Evaluation Works 1. **Rule Retrieval**: Fetch active/eligible rules for the specified workflow 2. **Feature Lookups**: Retrieve machine learning feature values from the online feature store 3. **Parallel Evaluation**: Parallel Evaluation of rules in a workflow for better performance ## Monitoring The service exports the following metrics to Prometheus by default. * `rule_evaluation_latency_seconds_count` * `rule_evaluation_requests_total` Prometheus is automatically set up in the Data plane i.e. customer's cluster when the Canso Superchart is installed. Refer to [🚒 Install Canso Helm Charts](/getting-started/canso-helm-charts) for more details. # API Documentation ## Table of Contents * [**Workflow & Rule Management**](#rule-management-service) * [**Transaction Monitoring & Evaluation**](#transaction-monitoring-and-evaluation) ## Rule Management Service #### List all workflows {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows" method="get" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Create workflow {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows" method="post" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Get workflow by name {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}" method="get" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Update workflow status {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/status" method="patch" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### List workflow rules {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/rules" method="get" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Create workflow rule {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/rules" method="post" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Get workflow rule details {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/rules/{rule\_name}" method="get" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Update workflow rule {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/rules/{rule\_name}" method="put" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Update rule status {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/rules/{rule\_name}/status" method="patch" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} #### Update rule stage {% openapi src="/files/3qZhFtv1pGxTSAPQ4CBv" path="/v1/risk/workflows/{workflow\_name}/rules/{rule\_name}/stage" method="patch" expanded="true" %} [workflows.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-ddb15dd422a55dfdb95e80b5f44c859b3dfe21ba%2Fworkflows.json?alt=media) {% endopenapi %} *** ## Transaction Monitoring and Evaluation {% openapi src="/files/lWOYQJ4tiuQMBqhRygQR" path="/v1/risk/evaluate-risk" method="post" expanded="true" %} [txn-eval.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-4c9a43e4768ebf8d4eb6bcc282abde0dc7d2550f%2Ftxn-eval.json?alt=media) {% endopenapi %} # API Documentation ## Health Check {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/health" method="get" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ## Tenant Options ### Get Tenant Options {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-options" method="get" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Add Tenant Options {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-options" method="post" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ## Product Options ### Get Product Options {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/product-options" method="get" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Add Product Options {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/product-options" method="post" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ## Tenant Data Connection Configs ### Get Tenant Data Connections {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-conn-configs" method="get" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Create Tenant Data Connections {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-conn-configs" method="post" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Update Tenant Data Connection {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-conn-configs/{name}" method="patch" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Delete Tenant Data Connection {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-conn-configs/{name}" method="delete" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ## Users ### Get Users {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/users" method="get" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ## Tenant Data ET Configs ### Create Tenant Data ET {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-et-configs" method="post" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Update Tenant Data ET {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-et-configs/{unit\_name}" method="patch" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} ### Delete Tenant Data ET {% openapi src="/files/L2oyKEPcxGy8YZ6auXWI" path="/v1/tenant-data-et-configs/{unit\_name}" method="delete" expanded="true" %} [openapi-spec-fraud-investigation.json](https://4002575704-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FcE4BqP2R5KBg2It3L34D%2Fuploads%2Fgit-blob-8fcb864e2b0546bd5794e621ea1692d30095d143%2Fopenapi-spec-fraud-investigation.json?alt=media) {% endopenapi %} # Registry In the Canso platform, we begin by registering the data sources and sinks, followed by the registration of features. We support S3 and Kafka as data sources, and S3 and Redis as data sinks. Our platform accommodates raw, derived batch, and streaming features. Once a feature is defined, it must be registered to ensure its metadata, logic, and scheduling details are stored for future reference and reuse. ## Steps for Feature Registration ### Create Feature Objects * Define the raw, derived, and streaming features by creating their respective objects with their configurations. ### Generate the access token * To register the feature, generate an access token and store it as an environment variable on your data plane machine. * Only active tokens can be used. When a token expires, generate a new one and update the environment variable. * Take a reference from [this script](https://github.com/Yugen-ai/gru/blob/main/gru/examples/generate_access_token.py) below to generate the access token. ### Register the Features * Register each feature object within the platform to persist their details. * Here the feature names are primary key. You cannot register the feature again with the same name. * Take a reference from [this script](https://github.com/Yugen-ai/gru/blob/main/gru/examples/create_raw_feature.py#L89-L95) to register the features,data sources and sinks. ### Tool Tips * Go to [Top](#top) ⬆️ * Go back to [README.md](/) ⬅️ * Go back to [data-sources.md](/feature-store/data-sources) ⬅️ * Go back to [data-sinks.md](/feature-store/data-sinks) ⬅️ * Go back to [raw-feature.md](/feature-store/features/raw-feature) ⬅️ * Go back to [derived-feature.md](/feature-store/features/derived-feature) ⬅️ * Go back to [streaming-feature.md](/feature-store/features/streaming-feature) ⬅️ * Move forward to see [deploy-feature.md](/guides/deploy-feature) ➑️ # Dry Runs for Batch ML Features A user can do dry run for any registered feature even before doing the actual deployment of the given feature. * It helps end user to do deployment in production more confidently * If any issues ocuurs while doing the dry run can be resolved well before doing the actual deployment. ## Introduction Dry run is a way to test the feature before it gets deployed to production. It helps end users verify the feature logic they're implementing and reduces development and testing time. A user can do a dry run for any registered feature even before doing the actual deployment of the given feature. * It helps end users deploy features in production confidently * If there's any issue with the dry run output then it can be resolved well before doing the actual deployment. ## How to do a dry run? * Same way, user creates a feature and does feature.deploy(), user will call feature.dry\_run() method. * User will have to pass the start date and end date for the Dry run. Internally, an Airflow DAG will be scheduled for the given duration. * Users will also have to specify the MAX\_DRY\_RUN\_DURATION\_DAYS. ## Important Notes * Users are not allowed to do online ingestion in a dry run. Only offline materialization is in scope. ## Examples This \[example]\(#TODO need to add url here) demonstrates how to perform a dry run for a Raw Feature. Once the `dry_run` is executed a new DAG with the name "test\_crf\_sha\_testing\_rows\_4" is generated, which allows users to inspect the job, reference the materialised values, perform any quality checks they want. Upon completion of quality checks, the same feature can be deployed. # Deployment The final step is to deploy the registered feature for execution on your cluster. ### Raw Batch Feature Deployment * First, deploy the registered raw feature. Refer to [this piece of code](https://github.com/Yugen-ai/gru/blob/995ba59b346ea98429ff54bd5292eb7d52dad70a/gru/examples/create_raw_feature.py#L77-L79). * Once deployed, you can view the logs on your data-plane Airflow web UI. ![Raw Feature deployment in airflow](/files/xugNo6TAORPn2KovMAyH) ### Derived Batch Feature Deployment * Next, deploy the derived feature on top of the two deployed raw features. The system will check the successful deployment of the referenced raw features. If both raw features are successfully deployed, the derived feature will proceed; otherwise, it will fail. ![Derived Feature deployment in airflow](/files/ZUdXvBLGJ6Sbt9nil22Y) ### Streaming Feature Deployment * To Deploy the streaming feature refer to [this piece of code](https://github.com/Yugen-ai/gru/blob/995ba59b346ea98429ff54bd5292eb7d52dad70a/gru/examples/kafka_feature.py#L67-L69). * You can trace the logs in your dataplane ArgoCD. ![Streaming Feature deployment in argoCD](/files/JhhLQ6pIskljiycIb0P8) ### Notes * For detailed explanation purposes, feature scripts are divided into multiple parts across several markdown files. However, the standard practice is to create a script containing objects of data sinks, sources, or features, and then register and deploy them all at once. * There is one way to view the logs: through the UI, with Airflow for batch features and ArgoCD for streaming features. However, you can also view the logs from the terminal using the command `kubectl logs -f -n `. The namespace should be the one where the pods running your feature DAGs are located.