Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 0 additions & 129 deletions examples/03_pii_detection/basic_pii_detection.py

This file was deleted.

91 changes: 0 additions & 91 deletions examples/04_input_scanner/basic_input_scanning.py

This file was deleted.

31 changes: 7 additions & 24 deletions examples/05_llm_tracing/cohere.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@

from netra import Netra
from netra.decorators import workflow
from netra.pii import get_default_detector

# --- Configuration ---

Expand Down Expand Up @@ -72,14 +71,12 @@ def initialize_sdks() -> Any:


@workflow(name="cohere_chat_workflow") # type: ignore[arg-type]
async def get_cohere_response_with_pii_protection(
client: Any, messages: List[Dict[str, str]], model: str
) -> Optional[Any]:
async def get_cohere_response(client: Any, messages: List[Dict[str, str]], model: str) -> Optional[Any]:
"""
Sends messages to the Cohere chat API with PII protection.
Sends messages to the Cohere chat API.

This function is wrapped with Netra's `@workflow` decorator. It scans the
latest user message for PII and masks it before sending the payload to Cohere.
This function is wrapped with Netra's `@workflow` decorator so Netra can
monitor its execution.

Args:
client: The initialized Cohere AsyncClient.
Expand All @@ -94,24 +91,10 @@ async def get_cohere_response_with_pii_protection(
return None

# Separate the latest message from the previous chat history
latest_message_content = messages[-1].get("message", "")
message_to_send = messages[-1].get("message", "")
history_for_api = messages[:-1]

logging.info("Scanning latest user message for PII.")

# 1. PII Detection and Masking
pii_detector = get_default_detector(action_type="MASK")
pii_result = pii_detector.detect(latest_message_content)

if pii_result.has_pii:
logging.warning("PII detected. Using masked text for the API call.")
message_to_send = pii_result.masked_text
logging.info("Masked input: '%s'", message_to_send)
else:
logging.info("No PII detected in the latest message.")
message_to_send = latest_message_content

# 2. Call Cohere API
# Call Cohere API
try:
logging.info(f"Sending request to Cohere model: {model}")
response = await client.chat(
Expand Down Expand Up @@ -159,7 +142,7 @@ async def main() -> None:

# Get response from Cohere
with console.status("[bold cyan]Cohere is thinking...[/bold cyan]"):
response = await get_cohere_response_with_pii_protection(
response = await get_cohere_response(
client=cohere_client, messages=chat_history, model=args.model
) # type: ignore[misc]

Expand Down
37 changes: 10 additions & 27 deletions examples/05_llm_tracing/gemini.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@

from netra import Netra
from netra.decorators import workflow
from netra.pii import get_default_detector

# --- Configuration ---

Expand All @@ -29,8 +28,8 @@ def initialize_netra_sdk() -> None:
"""
Initializes the Netra SDK with configuration details.

This setup is crucial for enabling Netra's monitoring and PII protection
features within the application.
This setup is crucial for enabling Netra's monitoring features
within the application.
"""
try:
netra_api_key = os.environ["NETRA_API_KEY"]
Expand All @@ -54,13 +53,12 @@ def initialize_netra_sdk() -> None:


@workflow(name="translator_workflow") # type: ignore[arg-type]
async def translate_text_with_pii_protection(text_to_translate: str) -> Any:
async def translate_text(text_to_translate: str) -> Any:
"""
Translates English text to French using Gemini, with PII protection.
Translates English text to French using Gemini.

This function is wrapped with the Netra `@workflow` decorator, which allows
Netra to monitor its execution. Before translation, it scans the input
for PII. If PII is found, it's masked before being sent to the Gemini API.
Netra to monitor its execution.

Args:
text_to_translate: The string of English text to be translated.
Expand All @@ -74,21 +72,9 @@ async def translate_text_with_pii_protection(text_to_translate: str) -> Any:

logging.info("Starting translation workflow for: '%s'", text_to_translate)

# 1. PII Detection using Netra's default PII detector
logging.info("Scanning for PII in the input text.")
pii_detector = get_default_detector(action_type="MASK")
pii_result = pii_detector.detect(text_to_translate)
input_for_model = text_to_translate

if pii_result.has_pii:
logging.warning("PII detected. Using masked text for translation.")
# Use the text with PII masked (e.g., "My name is [PERSON_0]")
input_for_model = pii_result.masked_text
logging.info("Masked input: '%s'", input_for_model)
else:
logging.info("No PII detected. Using original text.")
input_for_model = text_to_translate

# 2. Translation using Google Gemini API
# Translation using Google Gemini API
try:
logging.info("Sending request to Gemini API.")
google_api_key = os.environ["GOOGLE_API_KEY"]
Expand Down Expand Up @@ -131,17 +117,15 @@ async def main() -> None:
Main function to parse command-line arguments and run the translation.
"""
# Setup command-line argument parsing to get input text from the user
parser = argparse.ArgumentParser(
description="Translate English text to French using Gemini with Netra PII protection."
)
parser = argparse.ArgumentParser(description="Translate English text to French using Gemini with Netra tracing.")
parser.add_argument("message", type=str, help="The English text to translate. Please wrap in quotes.")
args = parser.parse_args()

# Initialize the Netra SDK before running the workflow
initialize_netra_sdk()

# Run the translation function and get the result
translated_message = await translate_text_with_pii_protection(args.message) # type: ignore[misc]
translated_message = await translate_text(args.message) # type: ignore[misc]

if translated_message:
print("\n--- Translation Result ---")
Expand All @@ -154,8 +138,7 @@ async def main() -> None:

if __name__ == "__main__":
# Example Usage from command line:
# python your_script_name.py "Hello, my name is John and my email is john.doe@example.com"
# python your_script_name.py "This is a test without any personal data."
# python your_script_name.py "Hello, my name is John."

# Run the main asynchronous function
asyncio.run(main())
12 changes: 1 addition & 11 deletions examples/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,17 +18,7 @@ This directory contains examples demonstrating how to integrate Netra SDK into y
- **`class_decorators.py`** - Advanced class-level instrumentation patterns


### 🔒 03_pii_detection/

- **`basic_pii_detection.py`** - Comprehensive PII detection and protection strategies


### 🛡️ 04_input_scanner/

- **`basic_input_scanning.py`** - Advanced security scanning for malicious inputs


### 🤖 05_llm_tracing/

- **`gemini.py`** - Google Gemini API integration with PII protection
- **`gemini.py`** - Google Gemini API integration and tracing
- **`cohere.py`** - Cohere API monitoring and tracing
Loading