👌 IMPROVE: Examples

Author: arre-ankit

examples/agent/agent.run.stream.py

@@ -1,70 +1,64 @@
 """
-Run Agent Streaming
+Run Agent Streaming with get_runner
 
-This example demonstrates how to run an agent with streaming response.
+This example demonstrates how to run an agent with streaming response using get_runner.
 """
 
 import os
-import sys
 
 from dotenv import load_dotenv
 
-from langbase import Langbase
-from langbase.helper import stream_text
+from langbase import Langbase, get_runner
 
 load_dotenv()
 
 
 def main():
     # Check for required environment variables
     langbase_api_key = os.environ.get("LANGBASE_API_KEY")
-    llm_api_key = os.environ.get("LLM_API_KEY")
+    api_key = os.environ.get("LLM_API_KEY")
 
     if not langbase_api_key:
         print("❌ Missing LANGBASE_API_KEY in environment variables.")
         print("Please set: export LANGBASE_API_KEY='your_langbase_api_key'")
         exit(1)
 
-    if not llm_api_key:
-        print("❌ Missing LLM_API_KEY in environment variables.")
-        print("Please set: export LLM_API_KEY='your_llm_api_key'")
-        exit(1)
-
     # Initialize Langbase client
     langbase = Langbase(api_key=langbase_api_key)
 
-    # Run the agent with streaming
-    response = langbase.agent_run(
-        stream=True,
-        model="openai:gpt-4.1-mini",
-        api_key=llm_api_key,
-        instructions="You are a helpful assistant that help users summarize text.",
-        input=[{"role": "user", "content": "Who is an AI Engineer?"}],
-    )
-
-    print("Stream started.\n")
-
-    # Process the streaming response
-    for line in response.iter_lines():
-        if line:
-            line_str = line.decode("utf-8")
-            if line_str.startswith("data: "):
-                data = line_str[6:]  # Remove 'data: ' prefix
-                if data.strip() == "[DONE]":
-                    print("\nStream ended.")
-                    break
-                try:
-                    import json
-
-                    json_data = json.loads(data)
-                    if "choices" in json_data and len(json_data["choices"]) > 0:
-                        delta = json_data["choices"][0].get("delta", {})
-                        if "content" in delta:
-                            sys.stdout.write(delta["content"])
-                            sys.stdout.flush()
-                except json.JSONDecodeError:
-                    # Skip invalid JSON lines
-                    continue
+    try:
+        # Get readable stream - equivalent to const {stream} = await langbase.agent.run(...)
+        response = langbase.agent_run(
+            stream=True,
+            model="openai:gpt-4.1-mini",
+            instructions="You are a helpful assistant that help users summarize text.",
+            input=[{"role": "user", "content": "Who is an AI Engineer?"}],
+            api_key=api_key
+        )
+
+        # Convert the stream to a stream runner - equivalent to getRunner(stream)
+        runner = get_runner(response)
+
+        # Event-like handling in Python
+        # Method 1: Using iterator pattern (Python equivalent of event listeners)
+        
+        # Equivalent to runner.on('connect', ...)
+        print("Stream started.\n")
+
+        try:
+            # Equivalent to runner.on('content', content => {...})
+            for content in runner.text_generator():
+                print(content, end="", flush=True)
+            
+            # Equivalent to runner.on('end', ...)
+            print("\nStream ended.")
+            
+        except Exception as error:
+            # Equivalent to runner.on('error', error => {...})
+            print(f"Error: {error}")
+
+    except Exception as e:
+        print(f"Error: {e}")
 
 
 if __name__ == "__main__":

examples/chunker/chunker.py

@@ -4,6 +4,7 @@
 
 import os
 import pathlib
+import json
 
 from dotenv import load_dotenv
 
@@ -37,15 +38,10 @@ def main():
         and tools for each part of the job and provides developers with a zero-config composable 
         AI infrastructure."""
 
-        # Alternative: Read content from a file
-        # document_path = pathlib.Path(__file__).parent.parent / "parse" / "composable-ai.md"
-        # with open(document_path, "r", encoding="utf-8") as file:
-        #     content = file.read()
-
         # Chunk the content
         chunks = lb.chunker(content=content, chunk_max_length=1024, chunk_overlap=256)
 
-        print(chunks)
+        print(json.dumps(chunks, indent=2))
 
     except Exception as e:
         print(f"Error chunking content: {e}")

examples/embed/embed.py

@@ -1,7 +1,7 @@
 # Experimental upcoming beta AI primitve.
 # Please refer to the documentation for more information: https://langbase.com/docs for more information.
 import os
-
+import json
 from dotenv import load_dotenv
 
 from langbase import Langbase
@@ -22,7 +22,7 @@ def main():
         ],
         embedding_model="openai:text-embedding-3-large",
     )
-    print(response)
+    print(json.dumps(response, indent=2))
 
 
 if __name__ == "__main__":

examples/memory/memory.docs.delete.py

@@ -21,11 +21,11 @@ def main():
 
     # Memory name and document ID to delete
     memory_name = "product-knowledge"  # Replace with your memory name
-    document_id = "doc-123"  # Replace with the document ID you want to delete
+    document_name = "name.txt"  # Replace with the document name you want to delete
 
     # Delete the document
     try:
-        response = lb.memories.docs.delete(name=memory_name, document_id=document_id)
+        response = lb.memories.documents.delete(memory_name=memory_name, document_name=document_id)
 
         print(
             f"Document '{document_id}' deleted successfully from memory '{memory_name}'"

examples/memory/memory.docs.list.py

@@ -24,8 +24,8 @@ def main():
 
     # List documents in the memory
     try:
-        response = lb.memories.docs.list(
-            name=memory_name, limit=10  # Limit the number of documents returned
+        response = lb.memories.documents.list(
+            memory_name=memory_name
         )
 
         print(f"Documents in memory '{memory_name}':")

examples/memory/memory.docs.retry-embed.py

@@ -21,10 +21,11 @@ def main():
 
     # Memory name to retry embedding for
     memory_name = "product-knowledge"  # Replace with your memory name
+    document_name="name.txt" # Replace with document name
 
     # Retry embedding for failed documents
     try:
-        response = lb.memories.docs.retry_embed(name=memory_name)
+        response = lb.memories.documents.embeddings.retry(memory_name=memory_name,document_name=document_name)
 
         print(f"Retry embedding initiated for memory '{memory_name}'")
         print(json.dumps(response, indent=2))

examples/memory/memory.docs.upload.py

@@ -24,22 +24,17 @@ def main():
 
     # Upload documents to the memory
     try:
-        response = lb.memories.docs.upload(
-            name=memory_name,
-            documents=[
-                {
-                    "content": "Langbase is a powerful platform for building AI applications with composable AI.",
-                    "metadata": {"source": "documentation", "section": "introduction"},
-                },
-                {
-                    "content": "The platform supports various AI models and provides tools for memory management.",
-                    "metadata": {"source": "documentation", "section": "features"},
-                },
-            ],
+        # Example 1: Upload string content as bytes
+        content1 = "Langbase is a powerful platform for building AI applications with composable AI."
+        response1 = lb.memories.documents.upload(
+            memory_name=memory_name,
+            document_name="intro.txt",
+            document=content1.encode('utf-8'),  # Convert string to bytes
+            content_type="text/plain",
+            meta={"source": "documentation", "section": "introduction"}
         )
-
-        print("Documents uploaded successfully!")
-        print(json.dumps(response, indent=2))
+        print("Document 1 uploaded successfully!")
+        print(f"Status: {response1.status_code}")
 
     except Exception as e:
         print(f"Error uploading documents: {e}")

examples/memory/memory.retrieve.py

@@ -1,5 +1,8 @@
 """
 Example demonstrating how to retrieve memories in Langbase.
+
+This example shows how to retrieve memories using a query. The memory parameter
+expects a list of dictionaries with 'name' keys specifying which memories to search.
 """
 
 import json
@@ -21,12 +24,16 @@ def main():
 
     # Retrieve memories using a query
     memory_name = "product-knowledge"  # Replace with your memory name
-    query = "What are the main features of the product?"
+    query = "What is Langbase?"
 
     try:
         response = lb.memories.retrieve(
-            name=memory_name,
             query=query,
+            memory=[
+                {
+                    "name": memory_name
+                }
+            ],
             top_k=5,  # Number of relevant memories to retrieve
         )
 

examples/pipes/pipes.create.py

@@ -19,24 +19,19 @@ def main():
     # Initialize the client
     lb = Langbase(api_key=langbase_api_key)
 
-    # Define pipe configuration
-    pipe_config = {
-        "name": "my-summary-pipe",  # Replace with your desired pipe name
-        "description": "A pipe for text summarization",
-        "system_prompt": "You are a helpful assistant that summarizes text clearly and concisely.",
-        "model": "openai:gpt-4-turbo-preview",
-        "variables": [
-            {
-                "name": "text_to_summarize",
-                "description": "The text that needs to be summarized",
-                "type": "string",
-            }
-        ],
-    }
-
     # Create the pipe
     try:
-        response = lb.pipes.create(**pipe_config)
+        response = lb.pipes.create(
+            name="summary-agent",
+            description="A pipe for text summarization",
+            messages=[
+                {
+                    "role": "system",
+                    "content": "You are a helpful assistant that summarizes text clearly and concisely.",
+                }
+            ],
+            upsert=True
+        )
 
         print("Pipe created successfully!")
         print(json.dumps(response, indent=2))

examples/pipes/pipes.run.py

@@ -20,15 +20,19 @@ def main():
     # Initialize the client
     lb = Langbase(api_key=langbase_api_key)
 
-    # Name of the pipe to run
-    pipe_name = "summary-agent-14"  # Replace with your pipe name
-
-    # Define messages for the conversation
-    messages = [{"role": "user", "content": "Who is an AI Engineer?"}]
 
     # Run the pipe with explicit stream=False
     try:
-        response = lb.pipes.run(name=pipe_name, messages=messages, stream=False)
+        response = lb.pipes.run(
+            name="summary-agent", 
+            messages=[
+                {
+                    "role": "user",
+                    "content": "Who is an AI Engineer?"
+                }
+            ], 
+            stream=False
+        )
 
         # Print the entire response as is
         print(json.dumps(response, indent=2))