Chuyển tới nội dung chính

Quick start

In this walkthrough you register an agent, call it through the same command surface an API server uses, and read the frames of a streaming call. Everything runs in one process: an in-process worker client replaces NATS, and in-memory stores replace Postgres.

1. Create the project​

Follow Installation to check out the workspace, then create a binary crate next to it:

cargo new my-agent-service
my-agent-service/Cargo.toml
[dependencies]
beam_core = { path = "../riverbeam/crates/beam_core" }
beam_domain = { path = "../riverbeam/crates/beam_domain" }
beam_stream = { path = "../riverbeam/crates/beam_stream" }
beam_types = { path = "../riverbeam/crates/beam_types" }
beam_worker = { path = "../riverbeam/crates/beam_worker" }

serde_json = "1"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
uuid = { version = "1", features = ["v4", "serde"] }

2. Wire the pipeline​

my-agent-service/src/main.rs
use std::sync::Arc;

use beam_core::{BeamConfig, RiverBeam};
use beam_domain::{BeamDomain, InvokePayload, LocalWorkerClient};
use beam_stream::{BeamStream, BroadcastTransport};
use beam_types::{ContentPart, StreamChunkType};
use beam_worker::BeamWorker;
use serde_json::Value;
use uuid::Uuid;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Stream frames fan out to every broadcast subscriber.
    let transport = Arc::new(BroadcastTransport::new(64));
    let mut frames = transport.subscribe();

    // The pipeline, with one agent that echoes its input.
    let mut beam = RiverBeam::new(BeamConfig::default());
    beam.set_stream(Arc::new(BeamStream::new(transport.clone())));
    beam.register_echo_agent("assistant");

    // A worker, and an in-process client in place of NATS.
    let worker = Arc::new(BeamWorker::new(Arc::new(beam)));
    let domain = BeamDomain::new(LocalWorkerClient::new(worker));

    let session_id = Uuid::new_v4();
    let payload = |text: &str| InvokePayload {
        agent_name: "assistant".into(),
        content: vec![ContentPart::text(text)],
        session_id,
        model_override: None,
        metadata: Value::Null,
        stream: false,
        auth_context: None,
    };

    // 3. Synchronous invoke: waits for the agent result.
    let result = domain.invoke(payload("hello")).await?;
    println!("invoke -> {}", result["content"][0]["text"]);

    // 4. Streaming invoke: returns an acknowledgement at once.
    let ack = domain.stream(payload("stream please")).await?;
    println!("stream accepted, request_id = {}", ack.request_id);

    while let Ok(published) = frames.recv().await {
        println!("{:?} on {}", published.frame.chunk_type, published.channel);
        if published.frame.chunk_type == StreamChunkType::Done {
            break;
        }
    }
    Ok(())
}

3. Run it​

cargo run
invoke -> "Echo: hello"
stream accepted, request_id = 6b0c…
Chunk on ws.channel.<session_id>:<request_id>
Done on ws.channel.<session_id>:<request_id>

What happened​

  1. domain.invoke built an InvokeRequest and sent the invoke_agent task to the worker with request, a round trip.
  2. The worker ran the invoke pipeline: it opened a session, interaction and run in the in-memory store, saved your message, built the context, and called the echo agent.
  3. domain.stream sent the same task with send, which does not wait. The pipeline published the agent's frames to the channel ws.channel.{session_id}:{request_id}. The last frame has type done and carries the full AgentResult.

Both calls used the same session_id, so they belong to one conversation, and the second call sees the first turn in its history.

Next steps​