diff --git a/streaming-crusoe/.gitignore b/streaming-crusoe/.gitignore new file mode 100644 index 0000000..0a16c91 --- /dev/null +++ b/streaming-crusoe/.gitignore @@ -0,0 +1,4 @@ +.env +__pycache__/ +*.pyc +.DS_Store diff --git a/streaming-crusoe/README.md b/streaming-crusoe/README.md new file mode 100644 index 0000000..dcb03d4 --- /dev/null +++ b/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) diff --git a/streaming-crusoe/requirements.txt b/streaming-crusoe/requirements.txt new file mode 100644 index 0000000..59cc96a --- /dev/null +++ b/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 diff --git a/streaming-crusoe/streaming.py b/streaming-crusoe/streaming.py new file mode 100644 index 0000000..426419f --- /dev/null +++ b/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())