Skip to content

Commit b09f57a

Browse files
remytuyerasremytuyeras
committed
add streaming agents with llm
1 parent 98049a0 commit b09f57a

8 files changed

Lines changed: 633 additions & 5 deletions

File tree

‎README.md‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -186,7 +186,7 @@ python agents/agent_<name>/agent.py --config configs/<specific_config>.json
186186
187187
## Agent Collection
188188
189-
There are 56 agents available in this repo.
189+
There are 58 agents available in this repo.
190190
191191
### Legend
192192
@@ -216,7 +216,7 @@ Click to show the legend.
216216
### Core Messaging Agents
217217

218218
<details>
219-
<summary><b>(Click to expand)</b> Agents of <b>level 1 and 2</b> introducing core messaging primitives like <code>@send</code>, <code>@receive</code>, and <code>@hook</code>.</summary>
219+
<summary><b>(Click to expand)</b> Agents of <b>level 1-3</b> introducing core messaging primitives like <code>@send</code>, <code>@receive</code>, and <code>@hook</code>.</summary>
220220
<br>
221221

222222
<div style="display: flex; justify-content: center;">
@@ -284,7 +284,7 @@ Click to show the legend.
284284
<td><img src="https://img.shields.io/badge/LVL_1-%20?color=2fc56c" alt=""></td>
285285
<td><code>core</code></td>
286286
<td><img src="https://img.shields.io/badge/Redirect-%20?color=482fc5"alt=""></td>
287-
<td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✅</td><td>✅</td><td>✅</td>
287+
<td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✅</td>
288288
</tr>
289289
<tr>
290290
<td><code><strong><a href="agents/agent_EchoAgent_1/">EchoAgent_1</a></strong></code></td>
@@ -302,12 +302,34 @@ Click to show the legend.
302302
<td><img src="https://img.shields.io/badge/Redirect-%20?color=482fc5"alt=""></td>
303303
<td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✅</td><td>✅</td><td>✅</td>
304304
</tr>
305+
<tr>
306+
<td><code><strong><a href="agents/agent_StreamAgent_0/">StreamAgent_0</a></strong></code></td>
307+
<td style="font-size: 0.8em;">Derived from <code>EchoAgent_0</code>; triggers an LLM stream on <code>@receive</code>, buffers token events in an <code>asyncio.Queue</code>, and emits them via <code>@send</code> (queue-wait with timeout).</td>
308+
<td><img src="https://img.shields.io/badge/LVL_3-%20?color=dfa018" alt=""></td>
309+
<td><code>core</code> <code>streaming</code> <code>llm</code></td>
310+
<td><img src="https://img.shields.io/badge/Redirect-%20?color=482fc5" alt=""></td>
311+
<td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✅</td>
312+
</tr>
313+
<tr>
314+
<td><code><strong><a href="agents/agent_StreamAgent_1/">StreamAgent_1</a></strong></code></td>
315+
<td style="font-size: 0.8em;">Same as <code>StreamAgent_0</code> but uses a polling <code>@send</code> loop (<code>sleep</code> + <code>get_nowait</code>) to make the emission cadence easy to rate-limit.</td>
316+
<td><img src="https://img.shields.io/badge/LVL_3-%20?color=dfa018" alt=""></td>
317+
<td><code>core</code> <code>streaming</code> <code>llm</code></td>
318+
<td><img src="https://img.shields.io/badge/Redirect-%20?color=482fc5" alt=""></td>
319+
<td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✗</td><td>✅</td><td>✅</td>
320+
</tr>
305321
</tbody>
306322
</table>
307323
</div>
308324

309325
</details>
310326

327+
<p align="center">
328+
<img width="550px" src="assets/mov2gif/gifs/demo_stream_framed.gif" />
329+
</p>
330+
331+
332+
311333
### Chat Agents
312334

313335
<details>
@@ -470,7 +492,7 @@ Click to show the legend.
470492
### [Graph-Based](https://en.wikipedia.org/wiki/Graph_theory) (or [Category-Based](https://en.wikipedia.org/wiki/Category_theory)) Agents
471493

472494
<details>
473-
<summary><b>(Click to expand)</b> Agents of <b>level 3–5</b> that treat orchestration as a typed state machine: nodes are workflow stages, edges are transitions or decision-making choices, and "edges between edges" encode path-dependent amendments between decisions.</summary>
495+
<summary><b>(Click to expand)</b> Agents of <b>level 3-5</b> that treat orchestration as a typed state machine: nodes are workflow stages, edges are transitions or decision-making choices, and "edges between edges" encode path-dependent amendments between decisions.</summary>
474496
<br>
475497

476498
<div style="display: flex; justify-content: center;">
@@ -604,7 +626,7 @@ Click to show the legend.
604626
### MMO-game agents
605627

606628
<details>
607-
<summary><b>(Click to expand)</b> Agents of <b>level 3–4</b> that implement a shared 2D sandbox: game masters simulate and broadcast world state while players send keyboard input and render a local view using <code>@receive</code>, <code>@send</code>, and <code>@hook</code>.</summary>
629+
<summary><b>(Click to expand)</b> Agents of <b>level 3-4</b> that implement a shared 2D sandbox: game masters simulate and broadcast world state while players send keyboard input and render a local view using <code>@receive</code>, <code>@send</code>, and <code>@hook</code>.</summary>
608630
<br>
609631

610632
<div style="display: flex; justify-content: center;">

‎agents/agent_StreamAgent_0/agent.py‎

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
from summoner.client import SummonerClient
2+
from typing import Any, Union, Optional
3+
import argparse
4+
import asyncio
5+
import json
6+
import uuid
7+
8+
from langchain_openai import ChatOpenAI
9+
from langchain_core.messages import HumanMessage
10+
11+
from dotenv import load_dotenv
12+
load_dotenv()
13+
14+
15+
client = SummonerClient(name="StreamAgent_0")
16+
17+
# LLM with streaming enabled
18+
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
19+
20+
# Initialized in setup()
21+
token_queue: Optional[asyncio.Queue] = None
22+
current_stream_task: Optional[asyncio.Task] = None
23+
24+
25+
async def setup():
26+
global token_queue
27+
token_queue = asyncio.Queue()
28+
29+
30+
async def stream_llm_into_queue(prompt: str, remote_addr: str) -> None:
31+
"""
32+
Streams tokens from the LLM into token_queue as small payloads.
33+
"""
34+
assert token_queue is not None
35+
36+
stream_id = str(uuid.uuid4())
37+
38+
# Optional: tell the server a stream is starting
39+
await token_queue.put(
40+
{"type": "stream_start", "stream_id": stream_id}
41+
)
42+
43+
try:
44+
message = HumanMessage(content=prompt)
45+
46+
# Preferred: async streaming (does not block the event loop)
47+
async for chunk in llm.astream([message]):
48+
if chunk.content:
49+
await token_queue.put(
50+
{
51+
"type": "token",
52+
"stream_id": stream_id,
53+
"token": chunk.content,
54+
}
55+
)
56+
57+
await token_queue.put(
58+
{"type": "stream_end", "stream_id": stream_id}
59+
)
60+
61+
except asyncio.CancelledError:
62+
# If a new prompt arrives and we cancel the current stream
63+
await token_queue.put(
64+
{"type": "stream_cancelled", "stream_id": stream_id}
65+
)
66+
raise
67+
68+
except Exception as e:
69+
await token_queue.put(
70+
{
71+
"type": "stream_error",
72+
"stream_id": stream_id,
73+
"error": str(e),
74+
}
75+
)
76+
77+
78+
@client.receive(route="")
79+
async def receiver_handler(msg: Any) -> None:
80+
"""
81+
Trigger streaming when a message arrives.
82+
"""
83+
global current_stream_task
84+
assert token_queue is not None
85+
86+
# Keep your warning handling
87+
if isinstance(msg, str) and msg.startswith("Warning:"):
88+
client.logger.warning(msg.replace("Warning:", "[From Server]"))
89+
return
90+
91+
# Expect your server-style envelope:
92+
# {"remote_addr": "...", "content": ...}
93+
if not (isinstance(msg, dict) and "remote_addr" in msg and "content" in msg):
94+
client.logger.info(f"Ignored message (unexpected shape): {type(msg)}")
95+
return
96+
97+
remote_addr = str(msg["remote_addr"])
98+
content = msg["content"]
99+
100+
# Decide what prompt text is
101+
# - if content is a string, use it
102+
# - if content is a dict, try "prompt", else dump json
103+
if isinstance(content, str):
104+
prompt = content
105+
elif isinstance(content, dict):
106+
prompt = str(content.get("prompt") or json.dumps(content))
107+
else:
108+
prompt = str(content)
109+
110+
client.logger.info(f"Triggering LLM streaming for remote_addr={remote_addr} prompt={prompt!r}")
111+
112+
# If you want "one active stream at a time", cancel the previous one
113+
if current_stream_task is not None and not current_stream_task.done():
114+
client.logger.warning("Cancelling previous stream (new prompt arrived).")
115+
current_stream_task.cancel()
116+
try:
117+
await current_stream_task
118+
except Exception:
119+
pass
120+
121+
current_stream_task = asyncio.create_task(stream_llm_into_queue(prompt, remote_addr))
122+
123+
def get_token_queue() -> asyncio.Queue:
124+
global token_queue
125+
if token_queue is None:
126+
token_queue = asyncio.Queue()
127+
return token_queue
128+
129+
@client.send(route="")
130+
async def send_handler() -> Union[dict, str, None]:
131+
q = get_token_queue()
132+
try:
133+
return await asyncio.wait_for(q.get(), timeout=0.5)
134+
except asyncio.TimeoutError:
135+
return None
136+
137+
if __name__ == "__main__":
138+
parser = argparse.ArgumentParser(description="Run a Summoner client with a specified config.")
139+
parser.add_argument('--config', dest='config_path', required=False, help='The relative path to the config file (JSON) for the client (e.g., --config configs/client_config.json)')
140+
args = parser.parse_args()
141+
142+
client.loop.run_until_complete(setup())
143+
144+
client.run(host = "127.0.0.1", port = 8888, config_path=args.config_path or "configs/client_config.json")

‎agents/agent_StreamAgent_0/readme.md‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
# `StreamAgent_0`
2+
3+
This agent is a **client-side streaming example** built with the Summoner SDK. It is derived from [`EchoAgent_0`](../agent_EchoAgent_0) and reuses the same receive-then-buffer-then-send pattern, but replaces the static echo payload with an LLM stream. The goal is to show how to trigger a streaming LLM response on `@receive`, buffer streamed tokens in an `asyncio.Queue`, and emit those tokens back to the server via `@send`.
4+
5+
## Behavior
6+
7+
<details>
8+
<summary><b>(Click to expand)</b> The agent goes through these steps:</summary>
9+
<br>
10+
11+
1. The agent connects to the server.
12+
2. When it receives a message of the form:
13+
14+
```json
15+
{"remote_addr": "...", "content": ...}
16+
```
17+
18+
it interprets `content` as a prompt (string or dict) and starts an LLM stream.
19+
3. While the LLM is streaming, the agent pushes events into a queue:
20+
21+
* `{"type": "stream_start", "stream_id": ...}`
22+
* `{"type": "token", "stream_id": ..., "token": ...}`
23+
* `{"type": "stream_end", "stream_id": ...}`
24+
* (optional) `stream_cancelled` if a new prompt arrives while the previous stream is active
25+
* `stream_error` if the stream fails
26+
4. The `@send` route waits for the next queued event and sends it immediately. If no event is available, it returns `None` after 0.5 seconds.
27+
28+
> 📝 **Note:**
29+
>
30+
> * Only one stream is active at a time. A new received prompt cancels the previous stream task.
31+
> * The server is responsible for attaching `remote_addr` and packaging the outbound payload into `content`.
32+
33+
</details>
34+
35+
## SDK Features Used
36+
37+
| Feature | Description |
38+
| ---------------------------- | ------------------------------------------------------------------------ |
39+
| `SummonerClient(name=...)` | Creates and manages the agent instance |
40+
| `@client.receive(route=...)` | Registers an async handler triggered on incoming server messages |
41+
| `@client.send(route=...)` | Registers an async sender that periodically emits payloads to the server |
42+
| `client.run(...)` | Connects the client to the server and initiates the async lifecycle |
43+
44+
## How to Run
45+
46+
First, ensure the Summoner server is running:
47+
48+
```bash
49+
python server.py
50+
```
51+
52+
> [!TIP]
53+
> You can use the option `--config configs/server_config_nojsonlogs.json` for cleaner terminal output and log files.
54+
55+
Set your OpenAI credentials (recommended via `.env` in the agent folder):
56+
57+
```bash
58+
export OPENAI_API_KEY="..."
59+
```
60+
61+
Then run the agent:
62+
63+
```bash
64+
python agents/agent_StreamAgent_0/agent.py
65+
```
66+
67+
If you want to point to a specific client config:
68+
69+
```bash
70+
python agents/agent_StreamAgent_0/agent.py --config configs/client_config.json
71+
```
72+
73+
## Simulation Scenarios
74+
75+
This scenario demonstrates an end-to-end streaming round-trip across three processes:
76+
77+
* **Server**: routes messages between clients and provides the envelope `{remote_addr, content}`.
78+
* **`StreamAgent_0`**: receives a prompt, starts a streaming LLM call, and emits a sequence of streaming events (`stream_start`, `token`, `stream_end`).
79+
* **`InputAgent`**: provides an interactive CLI, sending prompts and printing responses as they arrive.
80+
81+
### 1) Start the three terminals
82+
83+
```sh
84+
# Terminal 1: start the server
85+
python server.py
86+
87+
# Terminal 2: start the streaming agent
88+
python agents/agent_StreamAgent_0/agent.py
89+
90+
# Terminal 3: start the interactive input agent
91+
python agents/agent_InputAgent/agent.py
92+
```
93+
94+
### 2) Enter a prompt in the InputAgent
95+
96+
In Terminal 3 you should see the InputAgent connect, then a prompt:
97+
98+
```log
99+
python agents/agent_InputAgent/agent.py
100+
[DEBUG] Loaded config from: configs/client_config.json
101+
2026-01-30 18:48:39.168 - InputAgent - INFO - Connected to server @(host=127.0.0.1, port=8888)
102+
> How are you?
103+
```
104+
105+
When you type `How are you?` and press Enter:
106+
107+
1. **`InputAgent`** sends the prompt to the server.
108+
2. The **server forwards** it to **`StreamAgent_0`**, packaging it as:
109+
110+
```json
111+
{"remote_addr": "...", "content": "How are you?"}
112+
```
113+
3. **`StreamAgent_0`** begins streaming an LLM response. As tokens arrive, it pushes events into its internal queue.
114+
4. The agent's `@send` loop emits those queued events back to the server as they become available.
115+
5. The **server forwards** those events to **`InputAgent`**, which prints them immediately.
116+
117+
### 3) Observe the streamed events in `InputAgent`
118+
119+
In Terminal 3, you should see a streaming envelope followed by many token events:
120+
121+
```log
122+
[Received] {'type': 'stream_start', 'stream_id': '6e475111-8633-42ea-a456-f146366f131f'}
123+
[Received] {'type': 'token', 'stream_id': '6e475111-8633-42ea-a456-f146366f131f', 'token': "I'm"}
124+
[Received] {'type': 'token', 'stream_id': '6e475111-8633-42ea-a456-f146366f131f', 'token': ' just'}
125+
[Received] {'type': 'token', 'stream_id': '6e475111-8633-42ea-a456-f146366f131f', 'token': ' a'}
126+
...
127+
[Received] {'type': 'token', 'stream_id': '6e475111-8633-42ea-a456-f146366f131f', 'token': '?'}
128+
[Received] {'type': 'stream_end', 'stream_id': '6e475111-8633-42ea-a456-f146366f131f'}
129+
>
130+
```
131+
132+
What to pay attention to:
133+
134+
* **`stream_id`**: all events for a single streamed response share the same `stream_id`. This is what allows the receiver (InputAgent or another client) to group tokens into the right response, even if multiple streams exist in the system.
135+
* **Token granularity**: tokens arrive as small chunks (sometimes including leading spaces). This is normal for streamed generation.
136+
* **Ordering**: you should always see `stream_start` first and `stream_end` last for a given `stream_id`, with one or many `token` events in between.
137+
138+
### 4) Observe StreamAgent_0's logs (trigger confirmation)
139+
140+
In Terminal 2, `StreamAgent_0` logs when it receives a prompt and starts an LLM stream:
141+
142+
```log
143+
python agents/agent_StreamAgent_0/agent.py
144+
[DEBUG] Loaded config from: configs/client_config.json
145+
2026-01-30 18:42:17.003 - StreamAgent_0 - INFO - Connected to server @(host=127.0.0.1, port=8888)
146+
2026-01-30 18:48:42.192 - StreamAgent_0 - INFO - Triggering LLM streaming for remote_addr=127.0.0.1:50490 prompt='How are you?'
147+
```

‎agents/agent_StreamAgent_0/requirements.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
langchain-openai
2+
openai>=1.0.0

0 commit comments

Comments
 (0)