feat: real-time streaming output examples for Crusoe Managed Inference

streaming-crusoe/.gitignore

@@ -0,0 +1,4 @@
+.env
+__pycache__/
+*.pyc
+.DS_Store

streaming-crusoe/README.md

@@ -0,0 +1,104 @@
+# Streaming Output × Crusoe AI
+
+Real-time token streaming from [Crusoe Managed Inference](https://www.crusoe.ai/cloud/managed-inference) — sync streaming, async streaming, callback handlers, and concurrent multi-prompt streaming.
+
+## What this demonstrates
+
+| Demo | What it shows |
+|------|--------------|
+| Basic streaming | Stream tokens to stdout as they arrive |
+| Callback handler | Stream via `StreamingStdOutCallbackHandler` |
+| Async streaming | Non-blocking token generation with `astream` |
+| Concurrent streaming | 3 prompts streamed simultaneously in 0.50s |
+
+## Prerequisites
+
+- Python 3.10+
+- A Crusoe Cloud account → [console.crusoecloud.com](https://console.crusoecloud.com)
+- Inference API key (Intelligence API Keys section under Security)
+
+## Setup
+
+```bash
+pip install -r requirements.txt
+export CRUSOE_API_KEY="your-api-key"
+```
+
+## Run all demos
+
+```bash
+python streaming.py
+```
+
+## Local testing (no Crusoe account needed)
+
+```bash
+pip install langchain-groq
+export GROQ_API_KEY="your-groq-key"  # free at console.groq.com
+python streaming.py
+```
+
+## How streaming works
+
+### Sync streaming
+
+```python
+from langchain_crusoe import ChatCrusoe
+from langchain_core.messages import HumanMessage
+
+llm = ChatCrusoe(model="meta-llama/Llama-3.3-70B-Instruct")
+
+for chunk in llm.stream([HumanMessage(content="Explain KV cache sharing.")]):
+    print(chunk.content, end="", flush=True)
+```
+
+### Async streaming
+
+```python
+import asyncio
+from langchain_crusoe import ChatCrusoe
+from langchain_core.messages import HumanMessage
+
+llm = ChatCrusoe(model="meta-llama/Llama-3.3-70B-Instruct")
+
+async def main():
+    async for chunk in llm.astream([HumanMessage(content="Explain KV cache sharing.")]):
+        print(chunk.content, end="", flush=True)
+
+asyncio.run(main())
+```
+
+### Concurrent streaming (multiple prompts at once)
+
+```python
+import asyncio
+from langchain_crusoe import ChatCrusoe
+from langchain_core.messages import HumanMessage
+
+llm = ChatCrusoe(model="meta-llama/Llama-3.3-70B-Instruct")
+
+async def stream_one(prompt: str):
+    result = ""
+    async for chunk in llm.astream([HumanMessage(content=prompt)]):
+        result += chunk.content
+    return result
+
+async def main():
+    results = await asyncio.gather(
+        stream_one("What is a vector database?"),
+        stream_one("What is RAG?"),
+        stream_one("What is a LangGraph agent?"),
+    )
+    for r in results:
+        print(r)
+
+asyncio.run(main())
+```
+
+## Related
+
+- [langchain-crusoe](../langchain-crusoe/) — LangChain integration for Crusoe Managed Inference
+- [langgraph-crusoe](../langgraph-crusoe/) — Multi-node agentic pipelines on Crusoe
+- [rag-crusoe](../rag-crusoe/) — RAG pipeline with Qdrant on Crusoe
+- [structured-output-crusoe](../structured-output-crusoe/) — Structured output and tool calling on Crusoe
+- [Crusoe Managed Inference Docs](https://docs.crusoecloud.com/managed-inference/overview)

streaming-crusoe/requirements.txt

@@ -0,0 +1,4 @@
+langchain-crusoe>=0.1.0
+langchain-groq>=1.1.0
+langchain-core>=1.0.0
+python-dotenv

streaming-crusoe/streaming.py

@@ -0,0 +1,129 @@
+"""
+Real-time streaming output from Crusoe Managed Inference.
+Demonstrates token streaming, async streaming, and streaming with callbacks.
+Tested locally with Groq as a drop-in replacement for Crusoe.
+"""
+import os
+import asyncio
+import time
+from dotenv import load_dotenv
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.callbacks import StreamingStdOutCallbackHandler
+
+load_dotenv()
+
+
+def get_llm(streaming: bool = False, callbacks=None):
+    if os.getenv("CRUSOE_API_KEY"):
+        from langchain_crusoe import ChatCrusoe
+        return ChatCrusoe(
+            model="meta-llama/Llama-3.3-70B-Instruct",
+            temperature=0.7,
+            max_tokens=512,
+            streaming=streaming,
+            callbacks=callbacks or [],
+        )
+    else:
+        from langchain_groq import ChatGroq
+        return ChatGroq(
+            model="llama-3.3-70b-versatile",
+            temperature=0.7,
+            max_tokens=512,
+            streaming=streaming,
+            callbacks=callbacks or [],
+        )
+
+
+def demo_basic_streaming():
+    """Demo 1: Stream tokens to stdout as they arrive."""
+    print("=" * 60)
+    print("DEMO 1: Basic token streaming")
+    print("=" * 60)
+    llm = get_llm()
+    prompt = "Explain how GPU clusters accelerate deep learning training in 5 steps."
+
+    print(f"Prompt: {prompt}\n")
+    print("Response (streaming):")
+
+    start = time.time()
+    token_count = 0
+    for chunk in llm.stream([HumanMessage(content=prompt)]):
+        print(chunk.content, end="", flush=True)
+        if chunk.content:
+            token_count += 1
+    elapsed = time.time() - start
+    print(f"\n\nTokens streamed: {token_count} | Time: {elapsed:.2f}s")
+
+
+def demo_streaming_with_callback():
+    """Demo 2: Stream using a callback handler."""
+    print("\n" + "=" * 60)
+    print("DEMO 2: Streaming with callback handler")
+    print("=" * 60)
+    print("Response (via StreamingStdOutCallbackHandler):\n")
+
+    llm = get_llm(
+        streaming=True,
+        callbacks=[StreamingStdOutCallbackHandler()]
+    )
+    llm.invoke([
+        SystemMessage(content="You are a concise technical writer."),
+        HumanMessage(content="What is MemoryAlloy KV cache sharing and why does it matter for LLM inference?")
+    ])
+    print()
+
+
+async def demo_async_streaming():
+    """Demo 3: Async streaming for non-blocking token generation."""
+    print("\n" + "=" * 60)
+    print("DEMO 3: Async streaming")
+    print("=" * 60)
+    llm = get_llm()
+    prompt = "List 5 best practices for deploying LLMs in production."
+
+    print(f"Prompt: {prompt}\n")
+    print("Response (async streaming):")
+
+    start = time.time()
+    async for chunk in llm.astream([HumanMessage(content=prompt)]):
+        print(chunk.content, end="", flush=True)
+    elapsed = time.time() - start
+    print(f"\n\nTime: {elapsed:.2f}s")
+
+
+async def demo_concurrent_streaming():
+    """Demo 4: Stream multiple prompts concurrently."""
+    print("\n" + "=" * 60)
+    print("DEMO 4: Concurrent async streaming (3 prompts at once)")
+    print("=" * 60)
+
+    prompts = [
+        "In one sentence: what is a vector database?",
+        "In one sentence: what is retrieval-augmented generation?",
+        "In one sentence: what is a LangGraph agent?",
+    ]
+
+    llm = get_llm()
+
+    async def stream_one(i: int, prompt: str):
+        result = ""
+        async for chunk in llm.astream([HumanMessage(content=prompt)]):
+            result += chunk.content
+        return i, result
+
+    start = time.time()
+    results = await asyncio.gather(*[stream_one(i, p) for i, p in enumerate(prompts)])
+    elapsed = time.time() - start
+
+    for i, result in sorted(results):
+        print(f"\nQ{i+1}: {prompts[i]}")
+        print(f"A{i+1}: {result}")
+
+    print(f"\n3 concurrent streams completed in {elapsed:.2f}s")
+
+
+if __name__ == "__main__":
+    demo_basic_streaming()
+    demo_streaming_with_callback()
+    asyncio.run(demo_async_streaming())
+    asyncio.run(demo_concurrent_streaming())