LiveView Integration with Nous AI

Comprehensive guide for integrating Nous AI agents with Phoenix LiveView, covering real-time streaming, multi-user coordination, and production patterns.

Table of Contents

1. Quick Start
2. Basic LiveView Integration
3. Streaming Patterns
4. GenServer Agent Management
5. PubSub Multi-User Coordination
6. Production Patterns
7. Error Handling & Recovery
8. Testing Strategies
9. Performance Optimization
10. Complete Examples

Quick Start

The fastest way to get AI agents working in LiveView:

defmodule MyAppWeb.ChatLive do
  use Phoenix.LiveView

  def mount(_params, _session, socket) do
    # Initialize agent once on mount and keep it in state
    agent = Nous.new("anthropic:claude-3-5-sonnet",
      instructions: "You are a helpful assistant."
    )

    socket = assign(socket,
      agent: agent,
      messages: [],
      input: "",
      streaming: false
    )

    {:ok, socket}
  end

  def handle_event("send_message", %{"message" => message}, socket) do
    if not socket.assigns.streaming do
      # Spawn streaming in background, passing the agent from state
      parent = self()
      spawn_link(fn -> stream_response(parent, socket.assigns.agent, message) end)

      socket = assign(socket, :streaming, true, :input, "")
      {:noreply, socket}
    else
      {:noreply, socket}
    end
  end

  def handle_info({:stream_chunk, text}, socket) do
    # Update UI with streaming text
    {:noreply, push_event(socket, "append_text", %{text: text})}
  end

  def handle_info({:stream_complete, final_message}, socket) do
    messages = socket.assigns.messages ++ [final_message]
    socket = assign(socket, :messages, messages, :streaming, false)
    {:noreply, socket}
  end

  defp stream_response(parent, agent, prompt) do
    # Use the agent passed as parameter instead of creating a new one
    Nous.run_stream(agent, prompt)
    |> Stream.each(fn
      {:text_delta, text} -> send(parent, {:stream_chunk, text})
      {:finish, result} -> send(parent, {:stream_complete, %{role: :assistant, content: result.output}})
    end)
    |> Stream.run()
  end

  def render(assigns) do
    # Your LiveView template here
  end
end

Basic LiveView Integration

Simple Agent Interaction

The most basic pattern for LiveView + AI:

defmodule MyAppWeb.SimpleChatLive do
  use Phoenix.LiveView

  def mount(_params, _session, socket) do
    # Initialize agent on mount
    agent = Nous.new("anthropic:claude-3-5-sonnet",
      instructions: "You are a helpful assistant in a web chat."
    )

    socket = assign(socket,
      agent: agent,
      messages: [],
      input: "",
      loading: false
    )

    {:ok, socket}
  end

  def handle_event("send_message", %{"message" => message}, socket) do
    if socket.assigns.loading do
      {:noreply, socket}
    else
      # Add user message
      user_msg = %{role: :user, content: message, timestamp: DateTime.utc_now()}
      messages = socket.assigns.messages ++ [user_msg]

      # Start async AI response
      task = Task.async(fn ->
        case Nous.run(socket.assigns.agent, message) do
          {:ok, result} -> {:ok, result.output}
          {:error, error} -> {:error, "AI Error: #{inspect(error)}"}
        end
      end)

      socket =
        socket
        |> assign(:messages, messages)
        |> assign(:input, "")
        |> assign(:loading, true)
        |> assign(:current_task, task)

      {:noreply, socket}
    end
  end

  def handle_event("update_input", %{"value" => value}, socket) do
    {:noreply, assign(socket, :input, value)}
  end

  def handle_info({ref, result}, socket) when socket.assigns.current_task.ref == ref do
    # Task completed
    Process.demonitor(ref, [:flush])

    {messages, socket} = case result do
      {:ok, ai_response} ->
        ai_msg = %{role: :assistant, content: ai_response, timestamp: DateTime.utc_now()}
        {socket.assigns.messages ++ [ai_msg], socket}

      {:error, error_msg} ->
        error_msg = %{role: :error, content: error_msg, timestamp: DateTime.utc_now()}
        {socket.assigns.messages ++ [error_msg], socket}
    end

    socket =
      socket
      |> assign(:messages, messages)
      |> assign(:loading, false)
      |> assign(:current_task, nil)

    {:noreply, socket}
  end

  def handle_info({:DOWN, ref, :process, _pid, reason}, socket) do
    if socket.assigns.current_task && socket.assigns.current_task.ref == ref do
      # Task crashed
      error_msg = %{role: :error, content: "Request failed: #{inspect(reason)}", timestamp: DateTime.utc_now()}
      messages = socket.assigns.messages ++ [error_msg]

      socket =
        socket
        |> assign(:messages, messages)
        |> assign(:loading, false)
        |> assign(:current_task, nil)

      {:noreply, socket}
    else
      {:noreply, socket}
    end
  end
end

Form Handling and Validation

defmodule MyAppWeb.SmartFormLive do
  use Phoenix.LiveView
  import Ecto.Changeset

  def mount(_params, _session, socket) do
    agent = Nous.new("anthropic:claude-3-5-sonnet",
      instructions: """
      You help users fill out forms intelligently.
      Suggest improvements, catch errors, and provide helpful hints.
      Be concise and actionable in your suggestions.
      """
    )

    changeset = change(%{name: "", email: "", message: ""})

    socket = assign(socket,
      agent: agent,
      changeset: changeset,
      ai_suggestions: [],
      validating: false
    )

    {:ok, socket}
  end

  def handle_event("validate", %{"form" => params}, socket) do
    changeset = validate_form(params)

    socket =
      socket
      |> assign(:changeset, changeset)
      |> maybe_get_ai_suggestions(params)

    {:noreply, socket}
  end

  def handle_event("submit", %{"form" => params}, socket) do
    changeset = validate_form(params)

    if changeset.valid? do
      # Process form submission
      {:noreply, put_flash(socket, :info, "Form submitted successfully!")}
    else
      {:noreply, assign(socket, :changeset, changeset)}
    end
  end

  defp validate_form(params) do
    types = %{name: :string, email: :string, message: :string}

    {%{}, types}
    |> cast(params, [:name, :email, :message])
    |> validate_required([:name, :email, :message])
    |> validate_format(:email, ~r/@/)
    |> validate_length(:message, min: 10)
  end

  defp maybe_get_ai_suggestions(socket, params) do
    if should_suggest?(params) and not socket.assigns.validating do
      parent = self()

      spawn_link(fn ->
        prompt = """
        User is filling out a contact form with:
        Name: "#{params["name"]}"
        Email: "#{params["email"]}"
        Message: "#{params["message"]}"

        Provide helpful, concise suggestions for improvement.
        Only suggest if there are clear issues or improvements.
        """

        case Nous.run(socket.assigns.agent, prompt) do
          {:ok, result} -> send(parent, {:ai_suggestions, result.output})
          {:error, _} -> send(parent, {:ai_suggestions, ""})
        end
      end)

      assign(socket, :validating, true)
    else
      socket
    end
  end

  def handle_info({:ai_suggestions, suggestions}, socket) do
    socket =
      socket
      |> assign(:ai_suggestions, parse_suggestions(suggestions))
      |> assign(:validating, false)

    {:noreply, socket}
  end

  defp should_suggest?(params) do
    # Only suggest when user has entered substantial content
    String.length(params["message"] || "") > 20
  end

  defp parse_suggestions(""), do: []
  defp parse_suggestions(text) do
    text
    |> String.split("\n")
    |> Enum.filter(&(String.trim(&1) != ""))
    |> Enum.take(3)  # Limit to 3 suggestions
  end
end

Streaming Patterns

Basic Text Streaming

defmodule MyAppWeb.StreamingChatLive do
  use Phoenix.LiveView

  def mount(_params, _session, socket) do
    socket = assign(socket,
      messages: [],
      input: "",
      streaming: false,
      current_response: "",
      stream_task: nil
    )

    {:ok, socket}
  end

  def handle_event("send_message", %{"message" => message}, socket) do
    if not socket.assigns.streaming do
      user_msg = %{role: :user, content: message, id: generate_id()}
      messages = socket.assigns.messages ++ [user_msg]

      # Start streaming
      parent = self()
      task = Task.async(fn ->
        stream_ai_response(parent, message)
      end)

      socket =
        socket
        |> assign(:messages, messages)
        |> assign(:input, "")
        |> assign(:streaming, true)
        |> assign(:current_response, "")
        |> assign(:stream_task, task)

      {:noreply, socket}
    else
      {:noreply, socket}
    end
  end

  def handle_event("stop_streaming", _params, socket) do
    if socket.assigns.stream_task do
      Task.shutdown(socket.assigns.stream_task, :brutal_kill)
    end

    # Save partial response
    if socket.assigns.current_response != "" do
      partial_msg = %{
        role: :assistant,
        content: socket.assigns.current_response <> "\n\n_(stopped)_",
        id: generate_id()
      }
      messages = socket.assigns.messages ++ [partial_msg]

      socket = assign(socket, :messages, messages)
    end

    socket =
      socket
      |> assign(:streaming, false)
      |> assign(:current_response, "")
      |> assign(:stream_task, nil)

    {:noreply, socket}
  end

  # Stream chunk received
  def handle_info({:stream_chunk, text}, socket) do
    new_response = socket.assigns.current_response <> text

    socket =
      socket
      |> assign(:current_response, new_response)
      |> push_event("stream_text", %{text: text})

    {:noreply, socket}
  end

  # Stream completed
  def handle_info({:stream_complete, usage}, socket) do
    final_msg = %{
      role: :assistant,
      content: socket.assigns.current_response,
      id: generate_id(),
      usage: usage
    }

    messages = socket.assigns.messages ++ [final_msg]

    socket =
      socket
      |> assign(:messages, messages)
      |> assign(:streaming, false)
      |> assign(:current_response, "")
      |> assign(:stream_task, nil)
      |> push_event("scroll_to_bottom", %{})

    {:noreply, socket}
  end

  # Stream error
  def handle_info({:stream_error, error}, socket) do
    error_msg = %{
      role: :error,
      content: "Stream error: #{inspect(error)}",
      id: generate_id()
    }

    messages = socket.assigns.messages ++ [error_msg]

    socket =
      socket
      |> assign(:messages, messages)
      |> assign(:streaming, false)
      |> assign(:current_response, "")
      |> assign(:stream_task, nil)

    {:noreply, socket}
  end

  # Task completion/crash handling
  def handle_info({ref, _result}, socket) when socket.assigns.stream_task.ref == ref do
    Process.demonitor(ref, [:flush])
    {:noreply, socket}
  end

  def handle_info({:DOWN, ref, :process, _pid, reason}, socket) do
    if socket.assigns.stream_task && socket.assigns.stream_task.ref == ref do
      error_msg = %{
        role: :error,
        content: "Connection lost: #{inspect(reason)}",
        id: generate_id()
      }

      messages = socket.assigns.messages ++ [error_msg]

      socket =
        socket
        |> assign(:messages, messages)
        |> assign(:streaming, false)
        |> assign(:current_response, "")
        |> assign(:stream_task, nil)

      {:noreply, socket}
    else
      {:noreply, socket}
    end
  end

  defp stream_ai_response(parent, prompt) do
    agent = Nous.new("anthropic:claude-3-5-sonnet")

    try do
      Nous.run_stream(agent, prompt)
      |> Stream.each(fn
        {:text_delta, text} ->
          send(parent, {:stream_chunk, text})

        {:finish, result} ->
          usage = if result, do: Map.from_struct(result.usage), else: %{}
          send(parent, {:stream_complete, usage})

        {:error, error} ->
          send(parent, {:stream_error, error})

        _ -> :ok
      end)
      |> Stream.run()
    rescue
      error ->
        send(parent, {:stream_error, error})
    end
  end

  defp generate_id, do: :crypto.strong_rand_bytes(8) |> Base.url_encode64()
end

Advanced Streaming with Buffering

defmodule MyAppWeb.BufferedStreamingLive do
  use Phoenix.LiveView

  @buffer_size 50  # Characters to buffer before UI update
  @buffer_timeout 100  # Max ms to wait before flushing buffer

  def mount(_params, _session, socket) do
    socket = assign(socket,
      messages: [],
      streaming: false,
      buffer: "",
      buffer_timer: nil,
      chars_buffered: 0
    )

    {:ok, socket}
  end

  def handle_info({:stream_chunk, text}, socket) do
    new_buffer = socket.assigns.buffer <> text
    chars_buffered = socket.assigns.chars_buffered + String.length(text)

    socket = assign(socket, :buffer, new_buffer, :chars_buffered, chars_buffered)

    cond do
      # Buffer is full - flush immediately
      chars_buffered >= @buffer_size ->
        flush_buffer(socket)

      # First chunk - start timer
      socket.assigns.buffer_timer == nil ->
        timer = Process.send_after(self(), :flush_buffer, @buffer_timeout)
        assign(socket, :buffer_timer, timer)

      # Buffer accumulating - wait for timer or full buffer
      true ->
        socket
    end
    |> then(&{:noreply, &1})
  end

  def handle_info(:flush_buffer, socket) do
    {:noreply, flush_buffer(socket)}
  end

  def handle_info({:stream_complete, usage}, socket) do
    # Final flush and cleanup
    socket = flush_buffer(socket)

    # Mark streaming as complete
    socket = assign(socket, :streaming, false)
    {:noreply, socket}
  end

  defp flush_buffer(socket) do
    if socket.assigns.buffer != "" do
      # Cancel existing timer
      if socket.assigns.buffer_timer do
        Process.cancel_timer(socket.assigns.buffer_timer)
      end

      # Send buffered text to UI
      socket =
        socket
        |> push_event("append_text", %{text: socket.assigns.buffer})
        |> assign(:buffer, "")
        |> assign(:buffer_timer, nil)
        |> assign(:chars_buffered, 0)

      socket
    else
      socket
    end
  end
end

GenServer Agent Management

Stateful Agent Server

defmodule MyApp.AgentServer do
  @moduledoc """
  GenServer that manages a persistent AI agent with conversation history.

  Features:
  - Persistent conversation history
  - Configurable agent settings
  - Automatic context management
  - Usage tracking
  """

  use GenServer
  require Logger

  defstruct [
    :agent,
    :conversation_id,
    :user_id,
    messages: [],
    total_tokens: 0,
    last_activity: nil,
    settings: %{}
  ]

  ## Client API

  def start_link(opts) do
    {user_id, opts} = Keyword.pop!(opts, :user_id)
    {conversation_id, opts} = Keyword.pop!(opts, :conversation_id)

    name = via_tuple({user_id, conversation_id})
    GenServer.start_link(__MODULE__, {user_id, conversation_id, opts}, name: name)
  end

  def send_message(user_id, conversation_id, message, opts \\ []) do
    case get_or_start_agent(user_id, conversation_id) do
      {:ok, pid} ->
        GenServer.call(pid, {:send_message, message, opts}, 30_000)

      {:error, reason} ->
        {:error, reason}
    end
  end

  def send_message_stream(user_id, conversation_id, message, stream_to, opts \\ []) do
    case get_or_start_agent(user_id, conversation_id) do
      {:ok, pid} ->
        GenServer.call(pid, {:send_message_stream, message, stream_to, opts}, 30_000)

      {:error, reason} ->
        {:error, reason}
    end
  end

  def get_conversation(user_id, conversation_id) do
    case get_agent(user_id, conversation_id) do
      {:ok, pid} ->
        GenServer.call(pid, :get_conversation)

      {:error, :not_found} ->
        {:ok, []}

      {:error, reason} ->
        {:error, reason}
    end
  end

  def update_settings(user_id, conversation_id, settings) do
    case get_or_start_agent(user_id, conversation_id) do
      {:ok, pid} ->
        GenServer.call(pid, {:update_settings, settings})

      {:error, reason} ->
        {:error, reason}
    end
  end

  def stop_agent(user_id, conversation_id) do
    case get_agent(user_id, conversation_id) do
      {:ok, pid} ->
        GenServer.stop(pid)

      {:error, _} ->
        :ok
    end
  end

  ## Server Implementation

  def init({user_id, conversation_id, opts}) do
    # Load existing conversation if any
    messages = load_conversation(user_id, conversation_id)

    # Create agent with default or custom settings
    agent_settings = Keyword.get(opts, :agent_settings, default_agent_settings())
    agent = create_agent(agent_settings)

    state = %__MODULE__{
      agent: agent,
      user_id: user_id,
      conversation_id: conversation_id,
      messages: messages,
      total_tokens: calculate_total_tokens(messages),
      last_activity: DateTime.utc_now(),
      settings: agent_settings
    }

    # Schedule periodic cleanup
    schedule_cleanup()

    Logger.info("Started agent server for user=#{user_id}, conversation=#{conversation_id}")
    {:ok, state}
  end

  def handle_call({:send_message, message, opts}, _from, state) do
    # Add user message to history
    user_msg = %{
      role: :user,
      content: message,
      timestamp: DateTime.utc_now(),
      id: generate_id()
    }

    messages = state.messages ++ [user_msg]

    # Get AI response
    case Nous.run(state.agent, messages) do
      {:ok, result} ->
        ai_msg = %{
          role: :assistant,
          content: result.output,
          timestamp: DateTime.utc_now(),
          id: generate_id(),
          usage: result.usage
        }

        new_messages = messages ++ [ai_msg]
        new_total = state.total_tokens + (result.usage.total_tokens || 0)

        # Persist conversation
        save_conversation(state.user_id, state.conversation_id, new_messages)

        new_state = %{state |
          messages: new_messages,
          total_tokens: new_total,
          last_activity: DateTime.utc_now()
        }

        {:reply, {:ok, ai_msg}, new_state}

      {:error, reason} ->
        Logger.error("Agent error: #{inspect(reason)}")
        {:reply, {:error, reason}, state}
    end
  end

  def handle_call({:send_message_stream, message, stream_to, opts}, _from, state) do
    # Add user message
    user_msg = %{
      role: :user,
      content: message,
      timestamp: DateTime.utc_now(),
      id: generate_id()
    }

    messages = state.messages ++ [user_msg]

    # Start streaming in separate process
    parent = self()
    stream_id = generate_id()

    spawn_link(fn ->
      stream_response(parent, stream_to, state.agent, messages, stream_id)
    end)

    # Return immediately with stream ID
    new_state = %{state |
      messages: messages,
      last_activity: DateTime.utc_now()
    }

    {:reply, {:ok, stream_id}, new_state}
  end

  def handle_call({:update_settings, settings}, _from, state) do
    new_agent = create_agent(settings)

    new_state = %{state |
      agent: new_agent,
      settings: settings,
      last_activity: DateTime.utc_now()
    }

    {:reply, :ok, new_state}
  end

  def handle_call(:get_conversation, _from, state) do
    conversation = %{
      messages: state.messages,
      total_tokens: state.total_tokens,
      last_activity: state.last_activity,
      settings: state.settings
    }

    {:reply, {:ok, conversation}, state}
  end

  def handle_info({:stream_complete, stream_id, ai_message, usage}, state) do
    # Add completed message to history
    new_messages = state.messages ++ [ai_message]
    new_total = state.total_tokens + (usage.total_tokens || 0)

    # Persist updated conversation
    save_conversation(state.user_id, state.conversation_id, new_messages)

    new_state = %{state |
      messages: new_messages,
      total_tokens: new_total,
      last_activity: DateTime.utc_now()
    }

    {:noreply, new_state}
  end

  def handle_info(:cleanup_check, state) do
    # Check if agent should be stopped due to inactivity
    idle_time = DateTime.diff(DateTime.utc_now(), state.last_activity, :millisecond)

    if idle_time > idle_timeout() do
      Logger.info("Stopping idle agent: user=#{state.user_id}, conversation=#{state.conversation_id}")
      {:stop, :normal, state}
    else
      schedule_cleanup()
      {:noreply, state}
    end
  end

  def handle_info(_msg, state) do
    {:noreply, state}
  end

  ## Private Functions

  defp via_tuple({user_id, conversation_id}) do
    {:via, Registry, {MyApp.AgentRegistry, {user_id, conversation_id}}}
  end

  defp get_agent(user_id, conversation_id) do
    case Registry.lookup(MyApp.AgentRegistry, {user_id, conversation_id}) do
      [{pid, _}] when is_pid(pid) -> {:ok, pid}
      [] -> {:error, :not_found}
    end
  end

  defp get_or_start_agent(user_id, conversation_id) do
    case get_agent(user_id, conversation_id) do
      {:ok, pid} ->
        {:ok, pid}

      {:error, :not_found} ->
        case MyApp.AgentSupervisor.start_agent(user_id: user_id, conversation_id: conversation_id) do
          {:ok, pid} -> {:ok, pid}
          {:error, {:already_started, pid}} -> {:ok, pid}
          {:error, reason} -> {:error, reason}
        end
    end
  end

  defp create_agent(settings) do
    model = Map.get(settings, :model, "anthropic:claude-3-5-sonnet")
    instructions = Map.get(settings, :instructions, "You are a helpful assistant.")

    Nous.new(model,
      instructions: instructions,
      model_settings: Map.get(settings, :model_settings, %{})
    )
  end

  defp stream_response(parent, stream_to, agent, messages, stream_id) do
    accumulated = ""

    try do
      Nous.run_stream(agent, messages)
      |> Stream.reduce("", fn
        {:text_delta, text}, acc ->
          new_acc = acc <> text
          send(stream_to, {:stream_chunk, stream_id, text})
          new_acc

        {:finish, result}, acc ->
          ai_message = %{
            role: :assistant,
            content: acc,
            timestamp: DateTime.utc_now(),
            id: generate_id(),
            usage: result.usage
          }

          send(stream_to, {:stream_complete, stream_id, ai_message})
          send(parent, {:stream_complete, stream_id, ai_message, result.usage})
          acc

        {:error, error}, acc ->
          send(stream_to, {:stream_error, stream_id, error})
          acc
      end)
    rescue
      error ->
        send(stream_to, {:stream_error, stream_id, error})
    end
  end

  # Configuration
  defp default_agent_settings do
    %{
      model: "anthropic:claude-3-5-sonnet",
      instructions: "You are a helpful assistant.",
      model_settings: %{temperature: 0.7}
    }
  end

  defp idle_timeout, do: 30 * 60 * 1000  # 30 minutes

  defp schedule_cleanup do
    Process.send_after(self(), :cleanup_check, 5 * 60 * 1000)  # Check every 5 minutes
  end

  # Persistence (implement based on your storage)
  defp load_conversation(_user_id, _conversation_id), do: []

  defp save_conversation(_user_id, _conversation_id, _messages), do: :ok

  defp calculate_total_tokens(messages) do
    messages
    |> Enum.filter(&Map.has_key?(&1, :usage))
    |> Enum.map(& &1.usage.total_tokens)
    |> Enum.sum()
  end

  defp generate_id, do: :crypto.strong_rand_bytes(8) |> Base.url_encode64()
end

Agent Supervisor

defmodule MyApp.AgentSupervisor do
  @moduledoc """
  DynamicSupervisor for managing AI agent processes.
  """

  use DynamicSupervisor

  def start_link(init_arg) do
    DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  def start_agent(opts) do
    child_spec = {MyApp.AgentServer, opts}
    DynamicSupervisor.start_child(__MODULE__, child_spec)
  end

  def stop_agent(user_id, conversation_id) do
    MyApp.AgentServer.stop_agent(user_id, conversation_id)
  end

  def list_agents do
    DynamicSupervisor.which_children(__MODULE__)
  end

  def init(_init_arg) do
    DynamicSupervisor.init(strategy: :one_for_one, max_children: 1000)
  end
end

Nous.PubSub Integration

Nous provides a built-in PubSub abstraction (Nous.PubSub) that unifies all agent event broadcasting. Configure it once and all AgentServer instances, Callbacks, and Research Coordinator will broadcast events automatically.

Configuration

# config/config.exs
config :nous, pubsub: MyApp.PubSub

With this configuration, you no longer need to pass pubsub: MyApp.PubSub to every AgentServer — it's read automatically.

Using AgentServer with Nous.PubSub

defmodule MyAppWeb.ChatLive do
  use Phoenix.LiveView

  def mount(%{"session_id" => session_id}, _session, socket) do
    if connected?(socket) do
      # Subscribe to agent events (topic: "agent:{session_id}")
      Nous.PubSub.subscribe(MyApp.PubSub, "agent:#{session_id}")
    end

    # pubsub defaults to Nous.PubSub.configured_pubsub()
    {:ok, pid} = Nous.AgentServer.start_link(
      session_id: session_id,
      agent_config: %{model: "openai:gpt-4", instructions: "Be helpful."}
    )

    {:ok, assign(socket, agent: pid, session_id: session_id)}
  end

  # Events arrive automatically via PubSub
  def handle_info({:agent_delta, text}, socket), do: ...
  def handle_info({:agent_complete, result}, socket), do: ...
  def handle_info({:tool_call, call}, socket), do: ...
end

Async HITL Approval

For tools that require human approval, Nous.PubSub.Approval provides an async approval handler compatible with Nous.Plugins.HumanInTheLoop:

# In your agent setup:
deps = %{
  hitl_config: %{
    tools: ["send_email", "delete_record"],
    handler: Nous.PubSub.Approval.handler(
      session_id: session_id,
      timeout: :timer.minutes(5)
    )
  }
}

# In your LiveView:
def handle_info({:approval_required, info}, socket) do
  # info: %{tool_call_id: ..., name: ..., arguments: ..., session_id: ...}
  {:noreply, assign(socket, pending_approval: info)}
end

def handle_event("approve", _params, socket) do
  info = socket.assigns.pending_approval
  Nous.PubSub.Approval.respond(MyApp.PubSub, info.session_id, info.tool_call_id, :approve)
  {:noreply, assign(socket, pending_approval: nil)}
end

def handle_event("reject", _params, socket) do
  info = socket.assigns.pending_approval
  Nous.PubSub.Approval.respond(MyApp.PubSub, info.session_id, info.tool_call_id, :reject)
  {:noreply, assign(socket, pending_approval: nil)}
end

Topic Builders

Nous provides standardized topic builders:

Note: AgentServer uses the topic "agent:{session_id}" (without the nous: prefix) for backward compatibility.

PubSub Multi-User Coordination

Room-Based Chat with AI Moderation

defmodule MyAppWeb.RoomChatLive do
  use Phoenix.LiveView
  alias Phoenix.PubSub

  def mount(%{"room_id" => room_id}, %{"user_id" => user_id}, socket) do
    if connected?(socket) do
      # Subscribe to room updates
      PubSub.subscribe(MyApp.PubSub, "room:#{room_id}")
      PubSub.subscribe(MyApp.PubSub, "room:#{room_id}:user:#{user_id}")

      # Join the room
      MyApp.RoomManager.join_room(room_id, user_id)
    end

    socket = assign(socket,
      room_id: room_id,
      user_id: user_id,
      messages: [],
      participants: [],
      input: "",
      ai_agent_active: true,
      user_typing: MapSet.new()
    )

    {:ok, socket}
  end

  def handle_event("send_message", %{"message" => message}, socket) do
    if String.trim(message) != "" do
      # Broadcast message to room
      MyApp.RoomManager.send_message(
        socket.assigns.room_id,
        socket.assigns.user_id,
        message
      )

      {:noreply, assign(socket, :input, "")}
    else
      {:noreply, socket}
    end
  end

  def handle_event("typing", %{"typing" => typing}, socket) do
    # Broadcast typing indicator
    PubSub.broadcast(MyApp.PubSub, "room:#{socket.assigns.room_id}", {
      :user_typing,
      socket.assigns.user_id,
      typing
    })

    {:noreply, socket}
  end

  def handle_event("toggle_ai", _params, socket) do
    new_status = not socket.assigns.ai_agent_active

    MyApp.RoomManager.toggle_ai_agent(socket.assigns.room_id, new_status)

    {:noreply, assign(socket, :ai_agent_active, new_status)}
  end

  # New message in room
  def handle_info({:new_message, message}, socket) do
    messages = socket.assigns.messages ++ [message]
    {:noreply, assign(socket, :messages, messages)}
  end

  # AI response
  def handle_info({:ai_response, message}, socket) do
    messages = socket.assigns.messages ++ [message]
    {:noreply, assign(socket, :messages, messages)}
  end

  # Stream chunk from AI
  def handle_info({:ai_stream_chunk, chunk_id, text}, socket) do
    {:noreply, push_event(socket, "ai_stream_chunk", %{chunk_id: chunk_id, text: text})}
  end

  # AI stream complete
  def handle_info({:ai_stream_complete, message}, socket) do
    messages = socket.assigns.messages ++ [message]
    {:noreply, assign(socket, :messages, messages)}
  end

  # User typing indicators
  def handle_info({:user_typing, user_id, typing}, socket) do
    user_typing = if typing do
      MapSet.put(socket.assigns.user_typing, user_id)
    else
      MapSet.delete(socket.assigns.user_typing, user_id)
    end

    {:noreply, assign(socket, :user_typing, user_typing)}
  end

  # Participants updated
  def handle_info({:participants_updated, participants}, socket) do
    {:noreply, assign(socket, :participants, participants)}
  end

  def terminate(_reason, socket) do
    if socket.assigns[:room_id] && socket.assigns[:user_id] do
      MyApp.RoomManager.leave_room(socket.assigns.room_id, socket.assigns.user_id)
    end
    :ok
  end
end

Room Manager with AI Agent

defmodule MyApp.RoomManager do
  @moduledoc """
  Manages chat rooms with AI agent integration.
  """

  use GenServer
  alias Phoenix.PubSub

  defstruct [
    :room_id,
    participants: MapSet.new(),
    messages: [],
    ai_agent_active: true,
    ai_agent: nil,
    message_count: 0,
    last_ai_response: nil
  ]

  ## Client API

  def start_link(room_id) do
    GenServer.start_link(__MODULE__, room_id, name: via_tuple(room_id))
  end

  def join_room(room_id, user_id) do
    ensure_room_exists(room_id)
    GenServer.call(via_tuple(room_id), {:join, user_id})
  end

  def leave_room(room_id, user_id) do
    case get_room(room_id) do
      {:ok, pid} -> GenServer.call(pid, {:leave, user_id})
      {:error, _} -> :ok
    end
  end

  def send_message(room_id, user_id, content) do
    ensure_room_exists(room_id)
    GenServer.call(via_tuple(room_id), {:send_message, user_id, content})
  end

  def toggle_ai_agent(room_id, active) do
    ensure_room_exists(room_id)
    GenServer.call(via_tuple(room_id), {:toggle_ai_agent, active})
  end

  ## Server Implementation

  def init(room_id) do
    # Create AI agent for this room
    agent = Nous.new("anthropic:claude-3-5-sonnet",
      instructions: """
      You are an AI assistant in a group chat room.
      - Be helpful and engaging
      - Only respond when directly addressed or when the conversation needs guidance
      - Keep responses concise for chat format
      - Be respectful and moderate if needed
      - Use the participants' names when possible
      """
    )

    state = %__MODULE__{
      room_id: room_id,
      ai_agent: agent
    }

    {:ok, state}
  end

  def handle_call({:join, user_id}, _from, state) do
    if not MapSet.member?(state.participants, user_id) do
      new_participants = MapSet.put(state.participants, user_id)

      # Broadcast participant update
      broadcast_participants(state.room_id, new_participants)

      # Send recent messages to new participant
      send_message_history(user_id, state.room_id, state.messages)

      {:reply, :ok, %{state | participants: new_participants}}
    else
      {:reply, :ok, state}
    end
  end

  def handle_call({:leave, user_id}, _from, state) do
    new_participants = MapSet.delete(state.participants, user_id)

    broadcast_participants(state.room_id, new_participants)

    new_state = %{state | participants: new_participants}

    # Stop room if empty
    if MapSet.size(new_participants) == 0 do
      {:stop, :normal, :ok, new_state}
    else
      {:reply, :ok, new_state}
    end
  end

  def handle_call({:send_message, user_id, content}, _from, state) do
    message = %{
      id: generate_id(),
      user_id: user_id,
      content: content,
      timestamp: DateTime.utc_now(),
      type: :user
    }

    new_messages = state.messages ++ [message]
    new_message_count = state.message_count + 1

    # Broadcast message to room
    broadcast_message(state.room_id, message)

    new_state = %{state |
      messages: new_messages,
      message_count: new_message_count
    }

    # Check if AI should respond
    new_state = maybe_trigger_ai_response(new_state, message)

    {:reply, :ok, new_state}
  end

  def handle_call({:toggle_ai_agent, active}, _from, state) do
    new_state = %{state | ai_agent_active: active}
    {:reply, :ok, new_state}
  end

  def handle_info({:ai_stream_chunk, chunk_id, text}, state) do
    # Broadcast AI stream chunk
    PubSub.broadcast(MyApp.PubSub, "room:#{state.room_id}", {
      :ai_stream_chunk,
      chunk_id,
      text
    })

    {:noreply, state}
  end

  def handle_info({:ai_stream_complete, message, usage}, state) do
    # Add AI message to history
    new_messages = state.messages ++ [message]

    # Broadcast complete message
    broadcast_message(state.room_id, message)

    new_state = %{state |
      messages: new_messages,
      last_ai_response: DateTime.utc_now()
    }

    {:noreply, new_state}
  end

  ## Private Functions

  defp via_tuple(room_id) do
    {:via, Registry, {MyApp.RoomRegistry, room_id}}
  end

  defp get_room(room_id) do
    case Registry.lookup(MyApp.RoomRegistry, room_id) do
      [{pid, _}] when is_pid(pid) -> {:ok, pid}
      [] -> {:error, :not_found}
    end
  end

  defp ensure_room_exists(room_id) do
    case get_room(room_id) do
      {:ok, _pid} -> :ok
      {:error, :not_found} ->
        MyApp.RoomSupervisor.start_room(room_id)
        :ok
    end
  end

  defp maybe_trigger_ai_response(state, message) do
    if should_ai_respond?(state, message) do
      trigger_ai_response(state, message)
    else
      state
    end
  end

  defp should_ai_respond?(state, message) do
    state.ai_agent_active and (
      # AI mentioned by name
      String.contains?(String.downcase(message.content), ["ai", "assistant", "help"]) or
      # Question directed at group
      String.contains?(message.content, "?") or
      # First message in a while
      time_since_last_ai_response(state) > 300_000  # 5 minutes
    )
  end

  defp trigger_ai_response(state, trigger_message) do
    # Get recent conversation context
    recent_messages = Enum.take(state.messages, -10)

    # Create context prompt
    context = build_ai_context(recent_messages, state.participants)

    # Start streaming AI response
    parent = self()
    chunk_id = generate_id()

    spawn_link(fn ->
      stream_ai_response(parent, state.ai_agent, context, chunk_id)
    end)

    state
  end

  defp build_ai_context(messages, participants) do
    participant_list = participants |> Enum.join(", ")

    conversation = messages
    |> Enum.map(fn msg ->
      "#{msg.user_id}: #{msg.content}"
    end)
    |> Enum.join("\n")

    """
    You're in a chat room with participants: #{participant_list}

    Recent conversation:
    #{conversation}

    Respond helpfully and naturally to the conversation.
    """
  end

  defp stream_ai_response(parent, agent, context, chunk_id) do
    accumulated = ""

    Nous.run_stream(agent, context)
    |> Stream.reduce("", fn
      {:text_delta, text}, acc ->
        new_acc = acc <> text
        send(parent, {:ai_stream_chunk, chunk_id, text})
        new_acc

      {:finish, result}, acc ->
        ai_message = %{
          id: generate_id(),
          user_id: "AI Assistant",
          content: acc,
          timestamp: DateTime.utc_now(),
          type: :ai,
          usage: result.usage
        }

        send(parent, {:ai_stream_complete, ai_message, result.usage})
        acc

      _, acc -> acc
    end)
  end

  defp broadcast_message(room_id, message) do
    PubSub.broadcast(MyApp.PubSub, "room:#{room_id}", {:new_message, message})
  end

  defp broadcast_participants(room_id, participants) do
    participant_list = MapSet.to_list(participants)
    PubSub.broadcast(MyApp.PubSub, "room:#{room_id}", {:participants_updated, participant_list})
  end

  defp send_message_history(user_id, room_id, messages) do
    # Send recent messages to new participant
    recent_messages = Enum.take(messages, -20)

    Enum.each(recent_messages, fn message ->
      PubSub.broadcast(MyApp.PubSub, "room:#{room_id}:user:#{user_id}", {:new_message, message})
    end)
  end

  defp time_since_last_ai_response(state) do
    if state.last_ai_response do
      DateTime.diff(DateTime.utc_now(), state.last_ai_response, :millisecond)
    else
      999_999_999  # Very large number if no previous response
    end
  end

  defp generate_id, do: :crypto.strong_rand_bytes(8) |> Base.url_encode64()
end