📦 NEW: Pipe Examples

Author: arre-ankit

examples/pipes/pipe.run.chat.py

@@ -0,0 +1,43 @@
+"""
+Example demonstrating how to have a conversation using a pipe in non-streaming mode.
+"""
+
+import json
+import os
+
+from dotenv import load_dotenv
+
+from langbase import Langbase
+
+
+def main():
+    load_dotenv()
+
+    langbase = Langbase(api_key=os.getenv("LANGBASE_API_KEY"))
+
+    # Message 1: Tell something to the LLM.
+    response1 = langbase.pipes.run(
+        stream=False,
+        name="summary-agent",
+        messages=[{"role": "user", "content": "My company is called Langbase"}],
+    )
+
+    print(json.dumps(response1, indent=2))
+
+    # Message 2: Continue the conversation in the same thread
+    # Pass the complete conversation history including the new message
+    response2 = langbase.pipes.run(
+        name="summary-agent",
+        stream=False,
+        thread_id=response1["threadId"],
+        messages=[{"role": "user", "content": "Tell me the name of my company?"}],
+    )
+
+    print(json.dumps(response2, indent=2))
+    # You'll see any LLM will know the company is `Langbase`
+    # since it's the same chat thread. This is how you can
+    # continue a conversation in the same thread.
+
+
+if __name__ == "__main__":
+    main()

examples/pipes/pipe.run.pipe.key.py

@@ -0,0 +1,37 @@
+"""
+Example demonstrating how to run a pipe with a pipe API key.
+"""
+
+import os
+
+from dotenv import load_dotenv
+
+from langbase import Langbase, get_runner
+
+
+def main():
+    load_dotenv()
+
+    langbase = Langbase(api_key=os.getenv("LANGBASE_API_KEY"))
+
+    user_msg = "Who is an AI Engineer?"
+
+    # Get readable stream
+    response = langbase.pipes.run(
+        messages=[{"role": "user", "content": user_msg}],
+        stream=True,
+        raw_response=True,
+        api_key=os.getenv("PIPE_API_KEY"),
+    )
+
+    runner = get_runner(response)
+    print("Stream started.\n")
+    # Use text_generator() to stream content
+    for content in runner.text_generator():
+        print(content, end="", flush=True)
+
+    print("\n\nStream ended!")  # Add a newline after first response
+
+
+if __name__ == "__main__":
+    main()

examples/pipes/pipe.run.stream.chat.py

@@ -0,0 +1,53 @@
+"""
+Example demonstrating how to have a conversation using a pipe in streaming mode.
+"""
+
+import os
+import sys
+
+from dotenv import load_dotenv
+
+from langbase import Langbase, get_runner
+
+
+def main():
+    load_dotenv()
+
+    langbase = Langbase(api_key=os.getenv("LANGBASE_API_KEY"))
+
+    # Message 1: Tell something to the LLM.
+    response1 = langbase.pipes.run(
+        name="summary-agent",
+        stream=True,
+        messages=[{"role": "user", "content": "My company is called Langbase"}],
+    )
+
+    runner1 = get_runner(response1)
+
+    # Use text_generator() to stream content
+    for content in runner1.text_generator():
+        print(content, end="", flush=True)
+
+    print("\n\nStream ended!")  # Add a newline after first response
+
+    # Message 2: Ask something about the first message.
+    # Continue the conversation in the same thread by sending
+    # `thread_id` from the second message onwards.
+    response2 = langbase.pipes.run(
+        name="summary-agent",
+        stream=True,
+        thread_id=response1["thread_id"],
+        messages=[{"role": "user", "content": "Tell me the name of my company?"}],
+    )
+
+    runner2 = get_runner(response2)
+
+    # Use text_generator() to stream content
+    for content in runner2.text_generator():
+        print(content, end="", flush=True)
+
+    print("\n\nStream ended!")  # Add a newline after first response
+
+
+if __name__ == "__main__":
+    main()

examples/pipes/pipe.run.stream.llmkey.py

@@ -0,0 +1,38 @@
+"""
+Example demonstrating how to run a pipe with an LLM API key in streaming mode.
+"""
+
+import os
+import sys
+
+from dotenv import load_dotenv
+
+from langbase import Langbase, get_runner
+
+
+def main():
+    load_dotenv()
+
+    langbase = Langbase(api_key=os.getenv("LANGBASE_API_KEY"))
+
+    user_msg = "Who is an AI Engineer?"
+
+    # Get readable stream
+    response = langbase.pipes.run(
+        messages=[{"role": "user", "content": user_msg}],
+        stream=True,
+        raw_response=True,
+        name="summary-agent",
+        llm_key=os.getenv("LLM_KEY"),  # Your LLM API key
+    )
+
+    # Convert the stream to a stream runner.
+    runner = get_runner(response)
+
+    # Use text_generator() to stream content
+    for content in runner.text_generator():
+        print(content, end="", flush=True)
+
+
+if __name__ == "__main__":
+    main()

examples/pipes/pipe.structured.outputs.py

@@ -0,0 +1,90 @@
+"""
+Example demonstrating how to use structured outputs with a pipe.
+"""
+
+import json
+import os
+from typing import List
+
+from dotenv import load_dotenv
+from pydantic import BaseModel, Field
+
+from langbase import Langbase
+
+
+# Define the Strucutred Output JSON schema with Pydantic
+class Step(BaseModel):
+    explanation: str
+    output: str
+
+
+class MathReasoning(BaseModel):
+    steps: List[Step]
+    final_answer: str = Field(alias="final_answer")
+
+
+def create_math_tutor_pipe(langbase: Langbase):
+    json_schema = MathReasoning.model_json_schema()
+
+    pipe = langbase.pipes.create(
+        name="math-tutor",
+        model="openai:gpt-4o",
+        upsert=True,
+        messages=[
+            {
+                "role": "system",
+                "content": "You are a helpful math tutor. Guide the user through the solution step by step.",
+            },
+        ],
+        json=True,
+        response_format={
+            "type": "json_schema",
+            "json_schema": {
+                "name": "math_reasoning",
+                "schema": json_schema,
+            },
+        },
+    )
+
+    print("✅ Math Tutor pipe created:", json.dumps(pipe, indent=2))
+
+
+def run_math_tutor_pipe(langbase: Langbase, question: str):
+    response = langbase.pipes.run(
+        name="math-tutor",
+        messages=[{"role": "user", "content": question}],
+        stream=False,
+    )
+
+    # Parse and validate the response using Pydantic
+    solution = MathReasoning.model_validate_json(response["completion"])
+
+    print("✅ Structured Output Response:")
+    print("=" * 50)
+
+    for i, step in enumerate(solution.steps, 1):
+        print(f"Step {i}:")
+        print(f"  Explanation: {step.explanation}")
+        print(f"  Output: {step.output}")
+        print()
+
+    print(f"Final Answer: {solution.final_answer}")
+    print("=" * 50)
+
+
+def main():
+    load_dotenv()
+
+    if not os.getenv("LANGBASE_API_KEY"):
+        print("❌ Missing LANGBASE_API_KEY in environment variables.")
+        exit(1)
+
+    langbase = Langbase(api_key=os.getenv("LANGBASE_API_KEY"))
+
+    # Run this only once to create the pipe. Uncomment if it's your first time setting it up.
+    create_math_tutor_pipe(langbase)
+    run_math_tutor_pipe(langbase, "How can I solve 8x + 22 = -23?")
+
+
+if __name__ == "__main__":
+    main()