Python SDK & LangChain Tutorial

Prerequisites

Install the required Python packages in your virtual environment:

shellterminal

pip install openai langchain-openai langchain-core pydantic

1. Async Streaming Chat Application

Streaming responses reduces perceived latency by printing tokens as they arrive. Using AsyncOpenAI, you can process token streams asynchronously:

pythonstreaming_chat.py

import asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url=os.environ.get("INFINITY_BASE_URL", "https://infinityrouter.qd.je/v1"),
    api_key=os.environ.get("INFINITY_API_KEY"),
)

async def stream_response(prompt: str):
    print(f"User: {prompt}\nAssistant: ", end="", flush=True)
    
    stream = await client.chat.completions.create(
        model="claude-sonnet-5",
        messages=[
            {"role": "system", "content": "You are a helpful coding assistant."},
            {"role": "user", "content": prompt},
        ],
        stream=True,
    )

    async for chunk in stream:
        delta = chunk.choices[0].delta.content or ""
        print(delta, end="", flush=True)
    print("\n")

async def main():
    await stream_response("Write a fast binary search function in Python with docstrings.")

if __name__ == "__main__":
    asyncio.run(main())

2. Structured Output with Pydantic

Enforce rigid output schemas for extracting structured data from unstructured text using JSON schema definitions:

pythonstructured_output.py

import json
import os
from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI(
    base_url=os.environ.get("INFINITY_BASE_URL", "https://infinityrouter.qd.je/v1"),
    api_key=os.environ.get("INFINITY_API_KEY"),
)

class ArticleSummary(BaseModel):
    title: str = Field(description="The primary title of the article")
    key_points: list[str] = Field(description="Top 3 key takeaways")
    sentiment: str = Field(description="Positive, Neutral, or Negative")

schema = ArticleSummary.model_json_schema()

response = client.chat.completions.create(
    model="claude-sonnet-5",
    messages=[
        {"role": "system", "content": f"Extract structured data matching this JSON Schema: {json.dumps(schema)}"},
        {"role": "user", "content": "InfinityRouter announced 70% cheaper Claude models with automated upstream failover."},
    ],
    response_format={"type": "json_object"},
)

parsed_data = ArticleSummary.model_validate_json(response.choices[0].message.content)
print(f"Title: {parsed_data.title}")
print(f"Key points: {parsed_data.key_points}")
print(f"Sentiment: {parsed_data.sentiment}")

3. LangChain Chat & Chains

InfinityRouter integrates directly into LangChain using ChatOpenAI by overriding openai_api_base:

pythonlangchain_demo.py

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(
    model="claude-sonnet-5",
    openai_api_base=os.environ.get("INFINITY_BASE_URL", "https://infinityrouter.qd.je/v1"),
    openai_api_key=os.environ.get("INFINITY_API_KEY"),
    temperature=0.2,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert system architect specializing in low latency infrastructure."),
    ("user", "What are the primary tradeoffs of active-active database replication?"),
])

chain = prompt | llm | StrOutputParser()

response = chain.invoke({})
print(response)

Production Tip: Always set an account balance alert or rate limits in your API Keys settings to prevent rogue loops in autonomous agents from draining balances.