Example Walkthroughs

This guide walks through each example in detail, explaining what happens at each step.

Simple Agent Example
RAG Agent Example
Context Management Example
Building Your Own

How the examples are run

Flink is a provided dependency of the root module, so java -cp target/agentic-flink-1.0.0-SNAPSHOT-uber.jar ... fails with NoClassDefFoundError: org/apache/flink/configuration/ReadableConfig. The supported way to run a class from src/main/java/org/agentic/flink/example/ is the examples Maven profile, which forks a JVM with the test-scope classpath (Flink included):

./mvnw -q -f ports/jagentic-core/pom.xml install -DskipTests   # once per clone
./mvnw -q -P examples compile exec:exec -Dexec.mainClass=<fully qualified class>

1. Simple Agent Example

What it demonstrates: Basic agent workflow with tools, validation, and completion.

File: SimpleAgentExample.java

Running It

./mvnw -q -P examples compile exec:exec \
  -Dexec.mainClass=org.agentic.flink.example.SimpleAgentExample

Status: this command compiles and submits the job but currently fails before any event is processed with InvalidProgramException: Object org.agentic.flink.stream.AgentExecutionStream$$Lambda ... is not serializable (AgentExecutionStream registers a lambda that captures the non-serializable stream builder). The walk-through below describes the intended flow of the legacy AgentExecutionStream DSL; the failure is tracked in docs/audit-backlog.md under AGS-40. For an example that runs end to end see Building Your Own or the storage job in docs/guides/storage-quickstart.md.

What Happens (Step-by-Step)

Step 1: Setup Flink Environment

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);  // Run on single thread for simplicity

What this does: Creates the Flink execution environment (like starting a factory).

Step 2: Create Agent Configuration

AgentConfig config = new AgentConfig();
config.setAgentId("simple-agent-001");
config.setMaxIterations(10);
config.setValidationEnabled(true);

What this does:

Creates an agent named "simple-agent-001"
Allows up to 10 retry attempts if something fails
Enables result validation

Step 3: Define Tools

// Calculator tool
ToolDefinition calculator = new ToolDefinition();
calculator.setToolId("calculator");
calculator.setName("Calculator");
calculator.setDescription("Performs mathematical calculations");
calculator.addInputParameter("operation", "string", "Math operation (+, -, *, /)");
calculator.addInputParameter("a", "number", "First number");
calculator.addInputParameter("b", "number", "Second number");
config.addTool(calculator);

// Web search tool
ToolDefinition webSearch = new ToolDefinition();
webSearch.setToolId("web_search");
webSearch.setName("Web Search");
webSearch.setDescription("Searches the web for information");
webSearch.addInputParameter("query", "string", "Search query");
config.addTool(webSearch);

// Data analyzer tool
ToolDefinition dataAnalyzer = new ToolDefinition();
dataAnalyzer.setToolId("data_analyzer");
dataAnalyzer.setName("Data Analyzer");
dataAnalyzer.setDescription("Analyzes data and provides insights");
dataAnalyzer.addInputParameter("data", "array", "Data to analyze");
config.addTool(dataAnalyzer);

What this does: Gives the agent 3 tools it can use, defining what each tool needs as input.

Step 4: Create Sample Events

// Event 1: Calculator task
AgentEvent calcEvent = createToolCallEvent(
    "flow-001", "user-001", "simple-agent-001",
    "calculator",
    Map.of("operation", "+", "a", 2, "b", 40)
);

// Event 2: Web search task
AgentEvent searchEvent = createToolCallEvent(
    "flow-002", "user-002", "simple-agent-001",
    "web_search",
    Map.of("query", "Apache Flink documentation")
);

// Event 3: Data analysis task
AgentEvent analysisEvent = createToolCallEvent(
    "flow-003", "user-003", "simple-agent-001",
    "data_analyzer",
    Map.of("data", List.of(10, 20, 30, 40, 50))
);

What this does: Creates 3 different tasks for the agent to complete.

Step 5: Create Data Stream

DataStream<AgentEvent> events = env.fromElements(
    calcEvent,
    searchEvent,
    analysisEvent
);

What this does: Creates a stream (conveyor belt) of events.

Step 6: Print Results

events.print();

What this does: Shows the events as they're processed.

Step 7: Execute

env.execute("Simple Agent Example");

What this does: Starts the Flink job.

Output Explained

When you run this, you'll see:

1> AgentEvent(flowId=flow-001, eventType=TOOL_CALL_REQUESTED, toolId=calculator)

Meaning: Event 1 arrived, requesting the calculator tool.

[INFO] Agent simple-agent-001: Executing tool=calculator with params={operation=+, a=2, b=40}

Meaning: The agent is using the calculator tool.

[INFO] Tool execution complete: result=42

Meaning: Calculator finished: 2 + 40 = 42.

[INFO] Validation passed for flow-001

Meaning: The result was validated and is correct.

2> AgentEvent(flowId=flow-001, eventType=TOOL_CALL_COMPLETED, success=true)

Meaning: Task completed successfully!

Try It Yourself

Modify the calculation:

Open SimpleAgentExample.java

Find line with:

Map.of("operation", "+", "a", 2, "b", 40)

Change to:

Map.of("operation", "*", "a", 10, "b", 5)

Rebuild: mvn clean package
Run again - should now calculate 10 * 5 = 50!

2. RAG Agent Example

What it demonstrates: Working with documents, embeddings, and semantic search.

File: RagAgentExample.java

Prerequisites: Ollama at http://localhost:11434 with nomic-embed-text:latest pulled, and Qdrant (podman run -p 6333:6333 -p 6334:6334 qdrant/qdrant). The tool executors read ConfigKeys.DEFAULT_OLLAMA_BASE_URL and ConfigKeys.DEFAULT_QDRANT_HOST/DEFAULT_QDRANT_PORT. Without both services the job fails at the first tool call; the walk-through below was not re-run against live services.

Running It

./mvnw -q -P examples compile exec:exec \
  -Dexec.mainClass=org.agentic.flink.example.RagAgentExample

What Happens (Step-by-Step)

Step 1: Setup

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// Create tool registry
ToolExecutorRegistry registry = new ToolExecutorRegistry();
registry.register(new DocumentIngestionToolExecutor(ollamaConfig, qdrantConfig));
registry.register(new SemanticSearchToolExecutor(ollamaConfig, qdrantConfig));
registry.register(new RagToolExecutor(ollamaConfig, qdrantConfig));
registry.register(new EmbeddingToolExecutor(ollamaConfig));

What this does:

Sets up Flink
Registers 4 RAG tools
Configures Ollama (AI) and Qdrant (database)

Step 2: Create Documents to Ingest

// Document 1: About Flink state
String doc1 = "Apache Flink provides stateful stream processing with " +
    "exactly-once guarantees. State can be keyed or operator state.";

// Document 2: About state backends
String doc2 = "Flink supports multiple state backends: RocksDB, " +
    "HashMapStateBackend, and EmbeddedRocksDB for production workloads.";

// Document 3: About checkpointing
String doc3 = "Flink uses asynchronous snapshots for checkpointing " +
    "without stopping processing, enabling fault tolerance.";

What this does: Creates 3 documents about Apache Flink to store in the database.

Step 3: Ingest Documents

AgentEvent ingestEvent1 = createDocumentIngestionEvent(
    "flow-ingest-1", "user-001", doc1
);
// Similar for doc2 and doc3

What this does: Creates events to store each document.

Behind the scenes:

Text is split into chunks (manageable pieces)
Each chunk is converted to an embedding (a list of numbers that represents meaning)
Embeddings are stored in Qdrant with the original text

Example:

"Apache Flink provides stateful stream processing..."
   ↓ Embedding Model
[0.23, 0.45, 0.12, ..., 0.89]  ← 768 numbers representing the meaning
   ↓ Store in Qdrant
Stored with ID: doc-chunk-1

Step 4: Search Documents

AgentEvent searchEvent = createSemanticSearchEvent(
    "flow-search-1", "user-001",
    "How does Flink manage state?"  // Query
);

What this does: Searches for documents related to "Flink state management".

Behind the scenes:

Query is converted to an embedding
Qdrant finds similar embeddings (similar meaning)
Returns the most relevant text chunks

Example:

Query: "How does Flink manage state?"
   ↓ Embedding
[0.25, 0.44, 0.13, ..., 0.87]
   ↓ Compare to stored embeddings
Most similar: doc-chunk-1 (95% match)
   ↓ Return
"Apache Flink provides stateful stream processing..."

Step 5: RAG Query (Retrieval + Generation)

AgentEvent ragEvent = createRagQueryEvent(
    "flow-rag-1", "user-001",
    "Explain Flink checkpointing"
);

What this does: Uses retrieved documents to generate an answer.

Behind the scenes:

Search for relevant documents
Send documents + question to AI
AI generates answer using the documents

Example:

Question: "Explain Flink checkpointing"
   ↓ Search documents
Found: "Flink uses asynchronous snapshots..."
   ↓ Send to AI with context
AI: "Flink checkpointing works by taking asynchronous snapshots
     of the application state without stopping processing. This
     ensures fault tolerance..."

Step 6: Generate Embeddings

AgentEvent embeddingEvent = createEmbeddingEvent(
    "flow-embed-1", "user-001",
    "stateful stream processing"
);

What this does: Converts text to a vector embedding.

Use case: If you want to build custom search or similarity comparisons.

Output Explained

[INFO] Ingesting document: 500 characters, chunk_size=500
[INFO] Created 1 chunks, stored 1 embeddings
[INFO] Document ingested successfully: doc-1

Meaning: Document 1 was broken into chunks and stored.

[INFO] Searching for: "How does Flink manage state?"
[INFO] Found 3 matches with scores: [0.92, 0.87, 0.75]
[INFO] Top result: "Apache Flink provides stateful..."

Meaning: Search found 3 relevant chunks, most relevant one has 92% similarity.

[INFO] RAG query: "Explain Flink checkpointing"
[INFO] Retrieved 2 context chunks
[INFO] Generating answer with LLM...
[INFO] Answer: "Flink checkpointing works by..."

Meaning: RAG found context, sent to AI, got an answer.

Try It Yourself

Add your own document:

Add after the existing documents:

String myDoc = "Your document text here...";
AgentEvent myIngest = createDocumentIngestionEvent(
    "flow-ingest-my", "user-001", myDoc
);

Add to the event list
Rebuild and run
Your document is now searchable!

3. Context Management Example

What it demonstrates: Memory management, prioritization, and compaction.

File: ContextManagementExample.java

Running It

./mvnw -q -P examples compile exec:exec \
  -Dexec.mainClass=org.agentic.flink.example.ContextManagementExample

Status: this command compiles and submits the job but currently fails at job graph construction with IllegalStateException: Auto generated UIDs have been disabled but no UID or hash has been assigned to operator compaction-requests (the example disables PipelineOptions.AUTO_GENERATE_UIDS and then omits a uid on one source). The walk-through below describes the intended flow; the failure is tracked in docs/audit-backlog.md under AGS-40.

What Happens (Step-by-Step)

Step 1: Create Context with Size Limits

AgentContext context = new AgentContext(
    "agent-001",    // Agent ID
    "flow-001",     // Flow ID
    "user-001",     // User ID
    500,            // Max 500 tokens
    20              // Max 20 items
);

What this does: Creates a memory space with limits (like a notebook with limited pages).

Step 2: Add Important Facts (MUST Keep)

context.addContext(createItem(
    "Apache Flink provides stateful stream processing with exactly-once guarantees.",
    ContextPriority.MUST,          // Never forget!
    "state-management"
));

context.addContext(createItem(
    "Flink supports RocksDB, HashMapStateBackend, and EmbeddedRocksDB.",
    ContextPriority.MUST,
    "state-management"
));

What this does: Adds critical information that should never be forgotten.

Step 3: Add Useful Info (SHOULD Keep)

context.addContext(createItem(
    "RocksDB state backend is recommended for large state.",
    ContextPriority.SHOULD,        // Important but can compress
    "state-management"
));

What this does: Adds important but not critical information.

Step 4: Add Nice-to-Have Info (COULD Keep)

context.addContext(createItem(
    "The HashMapStateBackend stores state in memory as Java objects.",
    ContextPriority.COULD,         // Can forget if needed
    "state-backends"
));

What this does: Adds supplementary information.

Step 5: Add Irrelevant Info (WONT Keep)

context.addContext(createItem(
    "Docker containers provide application isolation.",
    ContextPriority.WONT,          // Not relevant, delete!
    "unrelated"
));

What this does: Adds irrelevant information that should be removed immediately.

Step 6: Exceed Memory Limits

By adding many items, the context exceeds its token limit (500 tokens).

Current usage: 612/500 tokens (122% full!)
Status: OVERFLOW - Compaction needed!

Step 7: Trigger Compaction

CompactionRequest request = new CompactionRequest(
    context,
    "Understand Apache Flink state management",
    CompactionRequest.CompactionReason.TOKEN_LIMIT_EXCEEDED
);

What this does: Requests automatic cleanup.

Step 8: 5-Phase Compaction Process

Phase 1: Remove WONT Items

Before: 2 WONT items (80 tokens)
Action: Delete all WONT items
After: 0 WONT items (saved 80 tokens)

Phase 2: Score Relevancy

Calculate relevancy for each item:
- Semantic similarity to intent (50%)
- How recent (20%)
- How often accessed (15%)
- Priority level (15%)

Phase 3: Remove Low-Relevancy COULD Items

COULD items with relevancy < 0.5:
- "Docker info" → relevancy=0.2 → DELETE
Saved: 40 tokens

Phase 4: Compress SHOULD Items

Sort SHOULD items by relevancy
Keep top 50%, remove bottom 50%
Saved: 60 tokens

Phase 5: Promote to Long-Term

High-relevancy MUST items (score ≥ 0.7):
- "Flink provides stateful processing" → PROMOTE
Action: Store in Qdrant for future retrieval

Step 9: Final Result

Original: 612 tokens, 12 items
After compaction: 432 tokens, 8 items
Saved: 180 tokens (29% reduction)
Promoted to long-term: 2 items

Output Explained

[INFO] Context size: 612/500 tokens (OVERFLOW)
[INFO] Starting compaction...

Meaning: Memory is full, cleanup starting.

[INFO] Phase 1: Removed 2 WONT items (80 tokens)

Meaning: Deleted irrelevant items.

[INFO] Phase 2: Scored 10 items for relevancy

Meaning: Calculated importance scores.

[INFO] Phase 3: Removed 1 COULD items (40 tokens)

Meaning: Deleted low-value supplementary info.

[INFO] Phase 4: Compressed 3 SHOULD items (60 tokens)

Meaning: Removed less relevant "important" items.

[INFO] Phase 5: Promoted 2 MUST items to long-term storage

Meaning: Archived high-value items to Qdrant.

[INFO] Compaction complete: 432/500 tokens (86% usage)

Meaning: Memory now has room!

Try It Yourself

Experiment with priorities:

Change some items from MUST to SHOULD
Notice how they might get removed during compaction

Change context limits to see different behaviors:

AgentContext context = new AgentContext(
    "agent-001", "flow-001", "user-001",
    200,   // Very small limit - more aggressive compaction
    10     // Fewer items
);

4. Building Your Own

Let's build a complete example from scratch: A weather assistant agent!

Step 1: Create the Tool

Create src/main/java/org/agentic/flink/tools/WeatherTool.java. AbstractToolExecutor takes the tool id and description in its constructor and provides a typed getRequiredParameter(params, key, type):

package org.agentic.flink.tools;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

public class WeatherTool extends AbstractToolExecutor {

    public WeatherTool() {
        super("weather", "Gets current weather for a location");
    }

    @Override
    public CompletableFuture<Object> execute(Map<String, Object> params) {
        String location = getRequiredParameter(params, "location", String.class);

        // In a real application this would call a weather API.
        Map<String, Object> weather = new HashMap<>();
        weather.put("location", location);
        weather.put("temperature", 72);
        weather.put("condition", "Sunny");
        weather.put("humidity", 45);

        return CompletableFuture.completedFuture(weather);
    }

    @Override
    public boolean validateParameters(Map<String, Object> params) {
        return params.containsKey("location");
    }
}

Step 2: Create the Example

Create src/main/java/org/agentic/flink/example/WeatherAgentExample.java. AgentConfig, AgentEvent and ToolCallRequest are Lombok @Data classes: construct them with the constructors shown and use the generated setters (setMaxIterations, setEnableValidation, setCurrentStage); there is no AgentConfig.addTool or ToolCallRequest.addParameter. The requested tool and its parameters travel in the event's data map, which is how SimpleAgentExample encodes tool calls as well:

package org.agentic.flink.example;

import java.util.Map;
import java.util.Optional;
import org.agentic.flink.core.AgentConfig;
import org.agentic.flink.core.AgentEvent;
import org.agentic.flink.core.AgentEventType;
import org.agentic.flink.tools.ToolExecutor;
import org.agentic.flink.tools.ToolExecutorRegistry;
import org.agentic.flink.tools.WeatherTool;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WeatherAgentExample {

    public static void main(String[] args) throws Exception {
        // 1. Set up Flink
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. Register the tool executor
        ToolExecutorRegistry registry = new ToolExecutorRegistry();
        registry.register(new WeatherTool());

        // 3. Configure the agent; only allowed tools may be called
        AgentConfig config = new AgentConfig("weather-agent", "Weather Agent");
        config.addAllowedTool("weather");
        config.setMaxIterations(3);
        config.setEnableValidation(true);

        // 4. Create one TOOL_CALL_REQUESTED event per city
        DataStream<AgentEvent> events =
            env.fromElements(
                createWeatherEvent("New York"),
                createWeatherEvent("London"),
                createWeatherEvent("Tokyo"));

        // 5. Execute the requested tool for each event and print the result
        events.map(new ToolCallFunction(registry, config)).print();

        env.execute("Weather Agent Example");
    }

    private static AgentEvent createWeatherEvent(String city) {
        AgentEvent event =
            new AgentEvent("flow-" + city, "user-001", "weather-agent", AgentEventType.TOOL_CALL_REQUESTED);
        event.setCurrentStage("TOOL_CALL");
        event.putData("toolId", "weather");
        event.putData("parameters", Map.of("location", city));
        return event;
    }

    /** Runs the tool named by the event; ToolExecutorRegistry and AgentConfig are Serializable. */
    public static class ToolCallFunction implements MapFunction<AgentEvent, String> {
        private final ToolExecutorRegistry registry;
        private final AgentConfig config;

        public ToolCallFunction(ToolExecutorRegistry registry, AgentConfig config) {
            this.registry = registry;
            this.config = config;
        }

        @Override
        @SuppressWarnings("unchecked")
        public String map(AgentEvent event) throws Exception {
            String toolId = event.getData("toolId", String.class);
            if (!config.isToolAllowed(toolId)) {
                return event.getFlowId() + ": tool " + toolId + " is not allowed";
            }
            Optional<ToolExecutor> executor = registry.getExecutor(toolId);
            if (executor.isEmpty()) {
                return event.getFlowId() + ": no executor registered for " + toolId;
            }
            Map<String, Object> parameters = (Map<String, Object>) event.getData("parameters");
            Object result = executor.get().execute(parameters).get();
            return event.getFlowId() + ": " + result;
        }
    }
}

Both listings are compiled by python docs/tools/check_java_snippets.py against the root module classpath.

Step 3: Build and Run

./mvnw -q -P examples compile exec:exec \
  -Dexec.mainClass=org.agentic.flink.example.WeatherAgentExample

Expected Output

The job runs to completion with exit code 0 and prints one line per event (SLF4J warnings about a missing provider precede them):

flow-New York: {condition=Sunny, temperature=72, humidity=45, location=New York}
flow-London: {condition=Sunny, temperature=72, humidity=45, location=London}
flow-Tokyo: {condition=Sunny, temperature=72, humidity=45, location=Tokyo}

Enhance It!

Add validation:

// Inside ToolCallFunction.map, after the tool has returned: org.agentic.flink.serde.ValidationResult
// carries the flow/user/agent ids plus error and warning lists.
Map<String, Object> weather = (Map<String, Object>) result;
int temp = (int) weather.get("temperature");

ValidationResult validation =
    new ValidationResult(event.getFlowId(), event.getUserId(), event.getAgentId());
if (temp < -100 || temp > 150) {
    validation.addError("Temperature out of realistic range");
} else {
    validation.setValid(true);
}

Add memory:

// Remember previously queried locations
AgentContext context = new AgentContext(/* ... */);
context.addContext(new ContextItem(
    "User asked about New York weather",
    ContextPriority.SHOULD,
    MemoryType.SHORT_TERM
));

Summary

You've now seen:

Simple Agent: Basic workflow
RAG Agent: Document processing and search
Context Agent: Memory management
Custom Agent: Build your own from scratch

Next Steps

Experiment - Modify the examples
Combine - Use RAG + Context + Your Tools together
Build - Create your own agent for your use case
Read - Check agent-framework.md for advanced topics

Happy building!

FilesExpand file tree

examples.md

Latest commit

History

examples.md

File metadata and controls

Example Walkthroughs

Table of Contents

How the examples are run

1. Simple Agent Example

Running It

What Happens (Step-by-Step)

Step 1: Setup Flink Environment

Step 2: Create Agent Configuration

Step 3: Define Tools

Step 4: Create Sample Events

Step 5: Create Data Stream

Step 6: Print Results

Step 7: Execute

Output Explained

Try It Yourself

2. RAG Agent Example

Running It

What Happens (Step-by-Step)

Step 1: Setup

Step 2: Create Documents to Ingest

Step 3: Ingest Documents

Step 4: Search Documents

Step 5: RAG Query (Retrieval + Generation)

Step 6: Generate Embeddings

Output Explained

Try It Yourself

3. Context Management Example

Running It

What Happens (Step-by-Step)

Step 1: Create Context with Size Limits

Step 2: Add Important Facts (MUST Keep)

Step 3: Add Useful Info (SHOULD Keep)

Step 4: Add Nice-to-Have Info (COULD Keep)

Step 5: Add Irrelevant Info (WONT Keep)

Step 6: Exceed Memory Limits

Step 7: Trigger Compaction

Step 8: 5-Phase Compaction Process

Step 9: Final Result

Output Explained

Try It Yourself

4. Building Your Own

Step 1: Create the Tool

Step 2: Create the Example

Step 3: Build and Run

Expected Output

Enhance It!

Summary

Next Steps