feat: Perplexity AI Integration (#97)

lux/lib/lux/llm/perplexity.ex

@@ -0,0 +1,167 @@
+defmodule Lux.LLM.Perplexity do
+  @moduledoc """
+  Perplexity AI LLM implementation with search-grounded responses and citation support.
+
+  Perplexity uses an OpenAI-compatible API but does not support function/tool calling.
+  Supports search domain filtering, citations, and streaming.
+  """
+
+  @behaviour Lux.LLM
+
+  alias Lux.LLM.ResponseSignal
+
+  require Logger
+
+  @endpoint "https://api.perplexity.ai/chat/completions"
+
+  defmodule Config do
+    @moduledoc """
+    Configuration module for Perplexity AI.
+    """
+    @type t :: %__MODULE__{
+            endpoint: String.t(),
+            model: String.t(),
+            api_key: String.t(),
+            temperature: float(),
+            top_p: float(),
+            max_tokens: integer(),
+            presence_penalty: float(),
+            frequency_penalty: float(),
+            return_citations: boolean(),
+            search_domain_filter: [String.t()],
+            return_images: boolean(),
+            stream: boolean(),
+            receive_timeout: integer(),
+            user: String.t(),
+            messages: [map()]
+          }
+
+    defstruct endpoint: "https://api.perplexity.ai/chat/completions",
+              model: "sonar-pro",
+              api_key: nil,
+              temperature: 0.7,
+              top_p: nil,
+              max_tokens: nil,
+              presence_penalty: 0.0,
+              frequency_penalty: 1.0,
+              return_citations: true,
+              search_domain_filter: nil,
+              return_images: false,
+              stream: false,
+              receive_timeout: 60_000,
+              user: nil,
+              messages: []
+  end
+
+  @impl true
+  def call(prompt, _tools, config) do
+    config =
+      struct(
+        Config,
+        Map.merge(
+          %{
+            model: (Application.get_env(:lux, :perplexity_models) || %{})[:default],
+            api_key: Application.get_env(:lux, :api_keys, [])[:perplexity]
+          },
+          config
+        )
+      )
+
+    messages = config.messages ++ build_messages(prompt)
+
+    body =
+      %{
+        model: Lux.Config.resolve(config.model),
+        messages: messages,
+        temperature: config.temperature
+      }
+      |> maybe_add(:top_p, config.top_p)
+      |> maybe_add(:max_tokens, config.max_tokens)
+      |> maybe_add(:presence_penalty, config.presence_penalty)
+      |> maybe_add(:frequency_penalty, config.frequency_penalty)
+      |> maybe_add(:return_citations, config.return_citations)
+      |> maybe_add(:search_domain_filter, config.search_domain_filter)
+      |> maybe_add(:return_images, config.return_images)
+      |> maybe_add(:stream, config.stream)
+
+    [
+      url: config.endpoint,
+      json: body,
+      headers: [
+        {"Authorization", "Bearer #{Lux.Config.resolve(config.api_key)}"},
+        {"Content-Type", "application/json"}
+      ]
+    ]
+    |> Keyword.merge(Application.get_env(:lux, __MODULE__, []))
+    |> Req.new()
+    |> Req.post()
+    |> case do
+      {:ok, %{status: 200} = response} ->
+        handle_response(response)
+
+      {:ok, %{status: 401}} ->
+        {:error, :invalid_api_key}
+
+      {:ok, %{status: status, body: %{"error" => %{"message" => message}}}} ->
+        {:error, {status, message}}
+
+      {:ok, %{status: status, body: body}} ->
+        {:error, {status, body}}
+
+      {:error, error} ->
+        handle_error(error)
+    end
+  end
+
+  defp build_messages(prompt) do
+    [%{role: "user", content: prompt}]
+  end
+
+  defp maybe_add(body, _key, nil), do: body
+  defp maybe_add(body, key, value), do: Map.put(body, key, value)
+
+  defp handle_response(%{body: body}) do
+    with %{"choices" => [choice | _]} <- body,
+         %{"message" => message, "finish_reason" => finish_reason} <- choice,
+         {:ok, content} <- parse_content(message["content"]) do
+      citations = Map.get(body, "citations", [])
+
+      payload = %{
+        content: content,
+        model: body["model"],
+        finish_reason: finish_reason,
+        tool_calls: nil,
+        tool_calls_results: nil
+      }
+
+      metadata = %{
+        id: body["id"],
+        created: body["created"],
+        usage: body["usage"],
+        citations: citations
+      }
+
+      %{
+        schema_id: ResponseSignal,
+        payload: payload,
+        metadata: metadata
+      }
+      |> Lux.Signal.new()
+      |> ResponseSignal.validate()
+    end
+  end
+
+  defp parse_content(nil), do: {:ok, nil}
+
+  defp parse_content(content) when is_binary(content) do
+    case Jason.decode(content) do
+      {:ok, decoded} -> {:ok, decoded}
+      {:error, _} -> {:ok, %{"text" => content}}
+    end
+  end
+
+  defp handle_error(error) do
+    Logger.error("Perplexity AI API error: #{inspect(error)}")
+    {:error, "Perplexity AI API error: #{inspect(error)}"}
+  end
+end

lux/lib/lux/nodejs.ex

@@ -17,7 +17,19 @@ defmodule Lux.NodeJS do
       ...>   '''
       ...> end
       42
+
+  ## Error Handling
+
+  All public functions return `{:ok, result}` on success or `{:error, reason}`
+  on failure. Possible error reasons include:
+
+    * `:timeout` - Node.js execution exceeded the specified timeout.
+      When a timeout occurs, the Node.js supervisor is restarted to
+      prevent resource leaks from the un-canceled Node.js process.
+    * `:invalid_code` - The provided code is empty or invalid.
+    * `string()` - Other error messages from the Node.js runtime.
   """
+
   @type eval_option ::
           {:variables, map()}
           | {:timeout, pos_integer()}
@@ -38,33 +50,40 @@ defmodule Lux.NodeJS do
     * `:variables` - A map of variables to bind in the Node.js context
     * `:timeout` - Timeout in milliseconds for Node.js execution
 
+  ## Returns
+
+    * `{:ok, result}` - Successfully evaluated, returns the result
+    * `{:error, :timeout}` - Execution exceeded the timeout
+    * `{:error, :invalid_code}` - Code is empty or invalid
+    * `{:error, reason}` - Other error from the Node.js runtime
+
   ## Examples
 
       iex> Lux.NodeJS.eval("export const main = ({x}) => x * 2", variables: %{x: 21})
       {:ok, 42}
 
-      iex> Lux.NodeJS.eval("export const main = () => os.getenv('TEST')", env: %{"TEST" => "value"})
-      {:ok, "value"}
+      iex> Lux.NodeJS.eval("export const main = () => 42", timeout: 5000)
+      {:ok, 42}
   """
-  @spec eval(String.t(), eval_options()) :: {:ok, term()} | {:error, String.t()}
+  @spec eval(String.t(), eval_options()) :: {:ok, term()} | {:error, term()}
   def eval(code, opts \\ []) do
-    {variables, opts} = Keyword.pop(opts, :variables, %{})
+    with {:ok, code} <- validate_code(code) do
+      {variables, opts} = Keyword.pop(opts, :variables, %{})
 
-    code
-    |> do_eval(variables, opts, &NodeJS.call/3)
-    |> case do
-      {:ok, result} -> {:ok, result}
-      {:error, "Call timed out."} -> {:error, :timeout}
-      {:error, error} -> {:error, error}
+      code
+      |> do_eval(variables, opts, &NodeJS.call/3)
+      |> handle_eval_result()
     end
   end
 
   @doc """
   Same as `eval/2`, but raises an error.
   """
   def eval!(code, opts \\ []) do
-    {variables, opts} = Keyword.pop(opts, :variables, %{})
-    do_eval(code, variables, opts, &NodeJS.call!/3)
+    with {:ok, code} <- validate_code(code) do
+      {variables, opts} = Keyword.pop(opts, :variables, %{})
+      do_eval(code, variables, opts, &NodeJS.call!/3)
+    end
   end
 
   @doc """
@@ -94,19 +113,7 @@ defmodule Lux.NodeJS do
 
     {"lux.mjs", "importPackage"}
     |> NodeJS.call([package_name, %{update_lock_file: update_lock_file}], opts)
-    |> case do
-      {:ok, %{"success" => true} = result} ->
-        {:ok, result}
-
-      {:ok, %{"error" => "ERR_MODULE_NOT_FOUND"}} ->
-        {:error, "Cannot import package: #{package_name}"}
-
-      {:ok, %{"error" => error}} ->
-        {:error, error}
-
-      {:error, error} ->
-        {:error, error}
-    end
+    |> handle_import_result()
   end
 
   @doc """
@@ -124,6 +131,16 @@ defmodule Lux.NodeJS do
     quote do: unquote(string)
   end
 
+  defp validate_code(""), do: {:error, :invalid_code}
+  defp validate_code(code) when is_binary(code) do
+    if String.trim(code) == "" do
+      {:error, :invalid_code}
+    else
+      {:ok, code}
+    end
+  end
+  defp validate_code(_), do: {:error, :invalid_code}
+
   defp do_eval(code, variables, opts, fun) do
     filename = create_file_name(code)
 
@@ -133,6 +150,30 @@ defmodule Lux.NodeJS do
     end
   end
 
+  defp handle_eval_result({:ok, result}), do: {:ok, result}
+  defp handle_eval_result({:error, "Call timed out."}) do
+    restart_nodejs_supervisor()
+    {:error, :timeout}
+  end
+  defp handle_eval_result({:error, error}), do: {:error, error}
+
+  defp handle_import_result({:ok, %{"success" => true} = result}) do
+    {:ok, result}
+  end
+  defp handle_import_result({:ok, %{"error" => "ERR_MODULE_NOT_FOUND"}}) do
+    {:error, "Cannot import package: #{package_name}"}
+  end
+  defp handle_import_result({:ok, %{"error" => error}}) do
+    {:error, error}
+  end
+  defp handle_import_result({:error, "Call timed out."}) do
+    restart_nodejs_supervisor()
+    {:error, :timeout}
+  end
+  defp handle_import_result({:error, error}) do
+    {:error, error}
+  end
+
   defp ensure_module_path(filename) do
     filepath = Path.join(@module_path, filename)
     module_path = Path.dirname(filepath)
@@ -148,4 +189,16 @@ defmodule Lux.NodeJS do
     hash = :sha |> :crypto.hash(code) |> Base.encode16(case: :lower)
     Path.join(["node_modules", "lux", "#{hash}.mjs"])
   end
+
+  defp restart_nodejs_supervisor do
+    Lux.Supervisor
+    |> Supervisor.which_children()
+    |> Enum.find_value(fn {_id, pid, _type, modules} ->
+      if pid != :undefined and modules == [NodeJS.Supervisor], do: pid, else: nil
+    end)
+    |> case do
+      nil -> :ok
+      pid -> Process.exit(pid, :kill)
+    end
+  end
 end

lux/priv/node/lux.mjs

@@ -1,28 +1,54 @@
 import { fileURLToPath } from 'url';
 import { dirname, join } from 'path';
-import { writeFile, readFile } from 'fs/promises';
+import { writeFile, readFile, unlink } from 'fs/promises';
 import { ensureDependencyInstalled } from "nypm";
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = dirname(__filename);
 
+function validatePackageName(packageName) {
+  if (typeof packageName !== 'string' || packageName.trim().length === 0) {
+    return { success: false, error: "ERR_INVALID_PACKAGE_NAME", message: "Package name must be a non-empty string" };
+  }
+  return null;
+}
+
+async function readPackageFiles() {
+  const packageJsonPath = join(__dirname, 'package.json');
+  const packageLockPath = join(__dirname, 'package-lock.json');
+  const [packageJson, packageLock] = await Promise.all([
+    readFile(packageJsonPath, 'utf8'),
+    readFile(packageLockPath, 'utf8').catch(() => null),
+  ]);
+  return { packageJson, packageLock, packageLockPath };
+}
+
+async function restorePackageFiles(original, options) {
+  if (options.update_lock_file) return;
+  const { packageJson, packageLock, packageLockPath } = original;
+  if (packageLock === null) {
+    await unlink(packageLockPath).catch(() => {});
+  } else {
+    await writeFile(packageLockPath, packageLock, 'utf8');
+  }
+}
+
 export const importPackage = async (packageName, options = {}) => {
-  const packageJson = await readFile(join(__dirname, 'package.json'), 'utf8');
-  const packageLock = await readFile(join(__dirname, 'package-lock.json'), 'utf8');
+  const validationError = validatePackageName(packageName);
+  if (validationError) return validationError;
+
+  const original = await readPackageFiles();
 
   try {
     await ensureDependencyInstalled(packageName, {
       cwd: __dirname,
       silent: true
     });
     await import(packageName);
-    return {success: true};
+    return { success: true };
   } catch (error) {
-    return {success: false, error: error.code};
+    return { success: false, error: error.code || "ERR_UNKNOWN", message: error.message };
   } finally {
-    if (!options.update_lock_file) {
-      await writeFile(join(__dirname, 'package.json'), packageJson, 'utf8');
-      await writeFile(join(__dirname, 'package-lock.json'), packageLock, 'utf8');
-    }
+    await restorePackageFiles(original, options);
   }
-};
+};