<!-- livebook:{"persist_outputs":true} -->
# Prisms Guide
```elixir
Mix.install(
[
{:lux, ">= 0.5.0"},
{:xml_builder, "~> 2.3"},
{:csv, "~> 3.2"},
{:swoosh, "~> 1.17"}
],
start_applications: false
)
Mix.Task.run("setup", install_deps: false)
Application.put_env(:venomous, :snake_manager, %{
python_opts: [
module_paths: [
Lux.Python.module_path(),
Lux.Python.module_path(:deps)
],
python_executable: "python3"
]
})
Application.ensure_all_started([:lux])
```
## Overview
<a href="https://livebook.dev/run?url=https%3A%2F%2Fgithub.com%2FSpectral-Finance%2Flux%2Fblob%2Fmain%2Flux%2Fguides%2Fprisms.livemd" style="display: none">
<img src="https://livebook.dev/badge/v1/blue.svg" alt="Run in Livebook" />
</a>
Prisms are modular units of functionality that can be composed into workflows. They provide a way to encapsulate business logic, transformations, and integrations into reusable components.
A Prism consists of:
* A unique identifier
* Input and output schemas
* A handler function
* Optional configuration and metadata
## Creating a Prism
Here's a basic example of a Prism:
```elixir
defmodule MyApp.Prisms.TextAnalysis do
use Lux.Prism,
name: "Text Analysis",
description: "Analyzes text for sentiment and key phrases",
input_schema: %{
type: :object,
properties: %{
text: %{type: :string, description: "Text to analyze"},
language: %{type: :string, description: "ISO language code"}
},
required: ["text"]
},
output_schema: %{
type: :object,
properties: %{
sentiment: %{
type: :string,
enum: ["positive", "negative", "neutral"]
},
confidence: %{type: :number},
key_phrases: %{
type: :array,
items: %{type: :string}
}
},
required: ["sentiment", "confidence"]
}
def handler(%{text: ""}, _ctx), do: {:error, :empty_text}
def handler(_input, _ctx) do
# Implementation
{:ok, %{
sentiment: "positive",
confidence: 0.95,
key_phrases: ["great", "awesome"]
}}
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, MyApp.Prisms.TextAnalysis, <<70, 79, 82, 49, 0, 0, 12, ...>>, {:handler, 2}}
```
## Using Prisms
Prisms can be used directly or composed into Beams:
```elixir
# Direct usage
{:ok, result} = MyApp.Prisms.TextAnalysis.run(%{
text: "Great product, highly recommended!",
language: "en"
})
result
```
<!-- livebook:{"output":true} -->
```
%{sentiment: "positive", confidence: 0.95, key_phrases: ["great", "awesome"]}
```
## Prism Types
### Transformation Prisms
Transform data from one format to another:
```elixir
defmodule MyApp.Prisms.DataTransformation do
use Lux.Prism,
name: "Data Transformer",
description: "Transforms data between formats",
input_schema: %{
type: :object,
properties: %{
data: %{type: :object},
format: %{type: :string, enum: ["json", "xml", "csv"]}
}
}
def handler(%{data: data, format: format}, _ctx) do
case format do
"json" -> {:ok, Jason.encode!(data)}
"xml" -> {:ok, XmlBuilder.generate(data)}
"csv" -> {:ok, CSV.encode(data)}
end
end
end
MyApp.Prisms.DataTransformation.run(%{
data: %{a: 1, b: 2},
format: "json"
})
```
<!-- livebook:{"output":true} -->
```
{:ok, "{\"a\":1,\"b\":2}"}
```
### Integration Prisms
Connect to external services:
```elixir
defmodule MyApp.Prisms.EmailSender do
use Lux.Prism,
name: "Email Sender",
description: "Sends emails via SMTP",
input_schema: %{
type: :object,
properties: %{
to: %{type: :string},
subject: %{type: :string},
body: %{type: :string}
},
required: ["to", "subject", "body"]
}
def handler(params, _ctx) do
case Swoosh.Mailer.deliver(build_email(params), []) do
{:ok, _} -> {:ok, %{sent: true}}
{:error, reason} -> {:error, reason}
end
end
defp build_email(%{to: to, subject: subject, body: body}) do
Swoosh.Email.new()
|> Swoosh.Email.to(to)
|> Swoosh.Email.subject(subject)
|> Swoosh.Email.text_body(body)
end
end
```
<!-- livebook:{"output":true} -->
```
{:module, MyApp.Prisms.EmailSender, <<70, 79, 82, 49, 0, 0, 12, ...>>, {:build_email, 1}}
```
### Business Logic Prisms
Implement business rules and workflows:
```elixir
defmodule MyApp.Prisms.OrderProcessor do
use Lux.Prism,
name: "Order Processor",
description: "Processes orders with business rules",
input_schema: %{
type: :object,
properties: %{
order: %{
type: :object,
properties: %{
items: %{type: :array},
total: %{type: :number},
customer: %{type: :object}
}
}
}
}
def handler(%{order: order}, _ctx) do
with :ok <- validate_inventory(order.items),
:ok <- validate_payment(order.total),
{:ok, processed} <- apply_discounts(order) do
{:ok,
%{
order_id: generate_order_id(),
processed_at: DateTime.utc_now(),
final_total: processed.total
}}
end
end
defp validate_inventory(_), do: :ok
defp validate_payment(_), do: :ok
defp apply_discounts(order), do: {:ok, order}
defp generate_order_id(), do: 1
end
MyApp.Prisms.OrderProcessor.run(%{
order: %{
items: [1, 2],
total: 2,
customer: %{}
}
})
```
<!-- livebook:{"output":true} -->
```
{:ok, %{order_id: 1, processed_at: ~U[2025-02-08 18:22:11.925749Z], final_total: 2}}
```
## Best Practices
1. **Input/Output Schemas**
* Define clear, specific schemas
* Document all properties
* Use appropriate types and constraints
* Include examples where helpful
2. **Error Handling**
* Return `{:ok, result}` or `{:error, reason}`
* Provide meaningful error messages
* Handle all error cases
* Use pattern matching for validation
3. **Context Usage**
* Use context for cross-cutting concerns
* Don't rely on global state
* Pass necessary data through context
* Keep context usage minimal
4. **Testing**
* Test happy and error paths
* Mock external dependencies
* Test with various inputs
* Test error conditions
Example test:
<!-- livebook:{"force_markdown":true} -->
```elixir
defmodule MyApp.Prisms.TextAnalysisTest do
use UnitCase, async: true
alias MyApp.Prisms.TextAnalysis
describe "run/2" do
test "analyzes positive text" do
{:ok, result} = TextAnalysis.run(%{
text: "Great product!",
language: "en"
})
assert result.sentiment == "positive"
assert result.confidence > 0.8
assert "great" in result.key_phrases
end
test "handles empty text" do
assert {:error, _} = TextAnalysis.run(%{
text: "",
language: "en"
})
end
end
end
```
## Advanced Topics
### Composable Prisms
Prisms can be composed together:
```elixir
defmodule MyApp.Prisms.Validator do
use Lux.Prism,
name: "Validator",
description: "Validate input"
def handler(input, _ctx), do: {:ok, input}
end
defmodule MyApp.Prisms.Enricher do
use Lux.Prism,
name: "Enricher",
description: "Enrich input"
def handler(input, _ctx), do: {:ok, input}
end
defmodule MyApp.Prisms.Processor do
use Lux.Prism,
name: "Processor",
description: "Process input"
def handler(input, _ctx), do: {:ok, input}
end
defmodule MyApp.Prisms.Pipeline do
use Lux.Prism,
name: "Processing Pipeline",
description: "Chains multiple prisms"
def handler(input, _ctx) do
with {:ok, validated} <- MyApp.Prisms.Validator.run(input),
{:ok, enriched} <- MyApp.Prisms.Enricher.run(validated),
{:ok, processed} <- MyApp.Prisms.Processor.run(enriched) do
{:ok, processed}
end
end
end
MyApp.Prisms.Pipeline.run(%{})
```
<!-- livebook:{"output":true} -->
```
{:ok, %{}}
```
### Async Prisms
Handle long-running operations:
```elixir
defmodule MyApp.Prisms.AsyncProcessor do
use Lux.Prism,
name: "Async Processor",
description: "Handles async operations"
def handler(_input, _ctx) do
task = Task.async(fn ->
# Long running operation
Process.sleep(5000)
{:ok, %{result: "done"}}
end)
case Task.yield(task, :timer.seconds(10)) do
{:ok, result} -> result
nil ->
Task.shutdown(task)
{:error, :timeout}
end
end
end
MyApp.Prisms.AsyncProcessor.run(%{})
```
<!-- livebook:{"output":true} -->
```
{:ok, %{result: "done"}}
```
### Python Integration
Prisms can leverage Python code directly in their handlers using `Lux.Python`. This is particularly useful for machine learning, data processing, or when you need to use Python libraries:
```elixir
defmodule MyApp.Prisms.SentimentAnalyzer do
use Lux.Prism,
name: "Sentiment Analyzer",
description: "Analyzes text sentiment using Python's NLTK",
input_schema: %{
type: :object,
properties: %{
text: %{type: :string, description: "Text to analyze"},
language: %{type: :string, description: "ISO language code"}
},
required: ["text"]
},
output_schema: %{
type: :object,
properties: %{
sentiment: %{type: :string, enum: ["positive", "negative", "neutral"]},
confidence: %{type: :number, minimum: 0, maximum: 1}
},
required: ["sentiment", "confidence"]
}
require Lux.Python
import Lux.Python
def handler(%{text: text, language: lang}, _ctx) do
# Import required Python packages
{:ok, %{"success" => true}} = Lux.Python.import_package("nltk")
# Execute Python code with variable bindings
result = python variables: %{text: text, lang: lang} do
~PY"""
import nltk
from nltk.sentiment import SentimentIntensityAnalyzer
# Download required NLTK data if not already present
try:
nltk.data.find('vader_lexicon')
except LookupError:
nltk.download('vader_lexicon')
# Analyze sentiment
sia = SentimentIntensityAnalyzer()
scores = sia.polarity_scores(text)
# Convert scores to our format
compound = scores['compound']
if compound >= 0.05:
sentiment = "positive"
elif compound <= -0.05:
sentiment = "negative"
else:
sentiment = "neutral"
# Return result
{
"sentiment": sentiment,
"confidence": abs(compound)
}
"""
end
{:ok, result}
end
end
MyApp.Prisms.SentimentAnalyzer.run(%{
text: "hello",
language: "en"
})
```
<!-- livebook:{"output":true} -->
```
{:ok, %{"confidence" => 0.0, "sentiment" => "neutral"}}
```
```elixir
defmodule MyApp.Prisms.CryptoAddressValidator do
use Lux.Prism,
name: "Crypto Address Validator",
description: "Validates cryptocurrency addresses",
input_schema: %{
type: :object,
properties: %{
address: %{type: :string, description: "The cryptocurrency address"},
chain: %{type: :string, enum: ["ethereum", "bitcoin"], description: "Chain type"}
},
required: ["address", "chain"]
}
require Lux.Python
import Lux.Python
def handler(%{address: address, chain: "ethereum"}, _ctx) do
# Import required packages
{:ok, %{"success" => true}} = Lux.Python.import_package("web3")
{:ok, %{"success" => true}} = Lux.Python.import_package("eth_utils")
result = python variables: %{address: address} do
~PY"""
from eth_utils import is_address, to_checksum_address
checksum_address = to_checksum_address(address)
is_valid = is_address(checksum_address)
{"is_valid": is_valid, "normalized_address": checksum_address}
"""
end
{:ok, result}
end
end
MyApp.Prisms.CryptoAddressValidator.run(%{
address: "0xd3CdA913deB6f67967B99D67aCDFa1712C293601",
chain: "ethereum"
})
```
<!-- livebook:{"output":true} -->
```
{:ok, %{"is_valid" => true, "normalized_address" => "0xd3CdA913deB6f67967B99D67aCDFa1712C293601"}}
```
#### Pure Python Prisms
You can define a prism entirely in Python. Extend the [Prism](https://github.com/Spectral-Finance/lux/blob/main/lux/priv/python/lux/prism.py) abstract class from `lux` package and implement `handler` and `new` methods:
```python
from lux import Prism
class SimplePrism(Prism):
id='aaace2b1-8089-4d4e-b68e-2ce904da12f0'
name='Simple Prism'
description='A very simple prism that greets you'
def handler(self, input, context):
return {
"success": True,
"data": {
"message": f'Hello, {input.get("name", "prism")}!',
},
}
@staticmethod
def new():
return SimplePrism()
```
You can load Python prism using `Lux.Prism.view/1` with file path. Please keep in mind that your Python prism runs in the `venv` set up during initial setup.
Only the packages listed in [`pyproject.toml`](../priv/python/pyproject.toml) are available to your Python prisms.
While you access the prism with `Lux.Prism.view/1`, `SimplePrism` module is dynamically defined and available in Elixir runtime.
<!-- livebook:{"force_markdown":true} -->
```elixir
prism = Lux.Prism.view("path/to/prism.py")
```
You can also access the Python prism via dynamically defined module `SimplePrism` to Elixir:
<!-- livebook:{"force_markdown":true} -->
```elixir
SimplePrism.view()
```
In the agent module, use `{:python, "path/to/prism.py"}` tuple to indicate that the prism is implemented in Python.
<!-- livebook:{"force_markdown":true} -->
```elixir
defmodule MyApp.Agent do
use Lux.Agent,
prisms: [
{:python, "path/to/prism.py"}
],
signal_handlers: [
{:python, "path/to/prism.py"}
]
end
```
The Python integration supports:
* Direct Python code execution with `~PY` sigils
* Variable binding between Elixir and Python
* Package management with `import_package/1`
* Error handling and timeouts
* Multi-line Python code with proper indentation
* Access to the full Python ecosystem
Best practices for Python integration:
1. Always handle package imports explicitly
2. Use proper error handling for Python code execution
3. Keep Python code focused and modular
4. Leverage Python's scientific and ML libraries when appropriate
5. Use type hints and docstrings in Python code
6. Follow both Elixir and Python style guides
Want to learn more about Python integration? Check out the [Python language support doc](language_support/python.livemd).
