Metadata-Version: 2.4
Name: langasync
Version: 0.2.2
Summary: Use LLM batch APIs with your LangChain chains — 50% cost savings, zero code changes
License: Apache-2.0
Project-URL: Homepage, https://github.com/langasync/langasync
Project-URL: Repository, https://github.com/langasync/langasync
Project-URL: Issues, https://github.com/langasync/langasync/issues
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: License :: OSI Approved :: Apache Software License
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic<3.0,>=2.0
Requires-Dist: pydantic-settings<3.0,>=2.0
Requires-Dist: langchain-core<2.0,>=1.0
Requires-Dist: cloudpickle<4.0,>=3.0
Requires-Dist: httpx<1.0,>=0.25.0
Requires-Dist: langchain-anthropic<1.4,>=1.0
Requires-Dist: aioboto3<16.0,>=13.0
Requires-Dist: langchain-aws<2.0,>=1.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Requires-Dist: pytest-dotenv>=0.5; extra == "dev"
Requires-Dist: pytest-httpx>=0.30; extra == "dev"
Requires-Dist: python-dotenv>=1.0; extra == "dev"
Requires-Dist: black>=23.0; extra == "dev"
Requires-Dist: ruff>=0.1; extra == "dev"
Requires-Dist: langchain-openai<2.0,>=1.0; extra == "dev"
Requires-Dist: openai<3.0,>=2.0; extra == "dev"
Requires-Dist: anthropic<1.0,>=0.20; extra == "dev"
Requires-Dist: google-genai<2.0,>=1.0; extra == "dev"
Requires-Dist: langchain-google-genai<3.0,>=2.0; extra == "dev"
Requires-Dist: langchain-aws<2.0,>=1.0; extra == "dev"
Requires-Dist: freezegun>=1.0; extra == "dev"
Requires-Dist: mypy>=1.0; extra == "dev"
Requires-Dist: types-aioboto3[bedrock,s3]>=13.0; extra == "dev"
Dynamic: license-file

<p align="center">
  <a href="https://github.com/langasync/langasync">
    <img alt="langasync" src="https://raw.githubusercontent.com/langasync/langasync/master/langasync.png" width="400">
  </a>
</p>

<p align="center">
  <strong>50% cost savings on LLM APIs with zero code changes.</strong>
</p>

<p align="center">
  <a href="https://pypi.org/project/langasync/"><img alt="PyPI" src="https://img.shields.io/pypi/v/langasync?color=blue"></a>
  <a href="https://pypi.org/project/langasync/"><img alt="Python" src="https://img.shields.io/pypi/pyversions/langasync"></a>
  <a href="https://github.com/langasync/langasync/blob/master/LICENSE"><img alt="License" src="https://img.shields.io/badge/license-Apache--2.0-blue"></a>
  <a href="https://github.com/langasync/langasync/actions/workflows/test.yml"><img alt="Tests" src="https://github.com/langasync/langasync/actions/workflows/test.yml/badge.svg"></a>
</p>

<p align="center">
  <a href="#quick-start">Quick Start</a> •
  <a href="#features">Features</a> •
  <a href="#documentation">Documentation</a> •
  <a href="#supported-providers">Providers</a> •
  <a href="#contributing">Contributing</a>
</p>

**langasync** lets you use provider batch APIs (OpenAI, Anthropic, Google Gemini, AWS Bedrock) with your existing LangChain chains. Wrap your chain, submit inputs, get results at half the cost.

```python
from langasync import batch_chain

# Your existing chain — no changes needed
chain = prompt | model | parser

# Wrap for batch processing
batch_wrapper = batch_chain(chain)

# Submit and retrieve results
job = await batch_wrapper.submit(inputs)
results = await job.get_results()
```

## Why langasync?

Provider batch APIs offer **50% cost savings** on tokens for workloads that can tolerate 24-hour turnaround. But they require completely different code patterns — file uploads, polling, result parsing.

langasync abstracts all of that:

| Without langasync | With langasync |
|-------------------|----------------|
| Rewrite chains as batch requests | Same chain, just wrapped |
| Manage file uploads (OpenAI) | Automatic |
| Build custom polling logic | Built-in `BatchPoller` |
| Parse provider-specific responses | Unified `BatchItem` |
| Handle partial failures manually | Automatic success/error separation |

## Quick Start

### Installation

```bash
pip install langasync
```

### Set your API key

```bash
export OPENAI_API_KEY=sk-...
# or for Anthropic:
export ANTHROPIC_API_KEY=sk-ant-...
# or for Google Gemini:
export GOOGLE_API_KEY=AIza...
```

For AWS Bedrock, see the [Bedrock Setup Guide](docs/bedrock-setup.md) — it requires an S3 bucket, IAM role, and model access configuration.

### Basic Usage

```python
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langasync import batch_chain, BatchPoller, BatchStatus

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "Explain {topic} in one paragraph.")
])
model = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | model

batch_wrapper = batch_chain(chain)

async def main():
    # Submit — returns immediately
    job = await batch_wrapper.submit([
        {"topic": "quantum computing"},
        {"topic": "machine learning"},
        {"topic": "blockchain"},
    ])
    print(f"Batch submitted: {job.job_id}")

    # Poll until complete — batch APIs typically take minutes to hours
    poller = BatchPoller()
    async for result in poller.wait_all():
        if result.status_info.status == BatchStatus.COMPLETED:
            for r in result.results:
                print(r.content)

asyncio.run(main())
```

## Features

### Drop-in Batch Support

Wrap any LangChain chain with `batch_chain()`. Prompts, models, parsers — all work automatically.

```python
chain = prompt | model | parser
batch_wrapper = batch_chain(chain)
```

### Structured Output

Full support for Pydantic output parsers and schemas:

```python
from pydantic import BaseModel
from langchain_core.output_parsers import PydanticOutputParser

class Analysis(BaseModel):
    sentiment: str
    confidence: float

parser = PydanticOutputParser(pydantic_object=Analysis)
chain = prompt | model | parser
```

### Tool Calling

`.bind_tools()` works out of the box:

```python
model = ChatOpenAI().bind_tools([my_tool])
chain = prompt | model
```

### Multimodal (Images & PDFs)

Pass images and documents as part of your batch inputs:

```python
from langchain_core.messages import HumanMessage, SystemMessage

batch_wrapper = batch_chain(model, settings)

await batch_wrapper.submit([
    [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=[
            {"type": "text", "text": "Describe this image."},
            {"type": "image", "url": "https://example.com/photo.jpg"},
        ]),
    ],
])
```

### Job Persistence

Batch jobs can take up to 24 hours. Job metadata persists automatically — resume after process restart:

```python
from langasync import BatchPoller

# Later, in a new process
poller = BatchPoller()

async for result in poller.wait_all():
    print(f"Job {result.job_id}: {result.status_info.status}")
```

> **Note:** langasync persists job metadata (IDs, status) but not results. Save your results when you receive them (OpenAI and Anthropic delete after ~30 days, Gemini after 48 hours, Bedrock persists in your S3 bucket).

### Partial Failure Handling

Get successful results even when some requests fail:

```python
result = await job.get_results()

for r in result.results:
    if r.success:
        print(r.content)
    else:
        print(f"Failed: {r.error}")
```

## Supported Providers

| Provider | Status | Batch API | Savings |
|----------|--------|-----------|---------|
| **OpenAI** | ✅ Supported | [Batch API](https://platform.openai.com/docs/guides/batch) | 50% |
| **Anthropic** | ✅ Supported | [Message Batches](https://docs.anthropic.com/en/docs/build-with-claude/batch-processing) | 50% |
| **Google Gemini** | ✅ Supported | [Batch API](https://ai.google.dev/gemini-api/docs/batch) | 50% |
| **AWS Bedrock** (Claude) | ✅ Supported | [Batch Inference](https://docs.aws.amazon.com/bedrock/latest/userguide/batch-inference.html) ([Setup Guide](docs/bedrock-setup.md)) | 50% |
| Azure OpenAI | 🔜 Planned | — | — |

## Documentation

### API Reference

| Function / Class | Description |
|------------------|-------------|
| `batch_chain(chain)` | Wrap a LangChain chain for batch processing |
| `BatchPoller()` | Poll pending jobs and retrieve results |
| `LangasyncSettings` | Configuration via env vars or constructor |
| `BatchJobService` | Service layer — `create()`, `get()`, `list()` batch jobs |
| `BatchJobHandle` | Returned by `submit()` / `create()` — provides `get_results()`, `cancel()` |
| `BatchStatus` | Enum: `PENDING`, `IN_PROGRESS`, `COMPLETED`, `FAILED`, `CANCELLED`, `EXPIRED` |

### Configuration

langasync reads configuration from environment variables or a `.env` file automatically:

```bash
# Provider API keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=AIza...



# AWS Bedrock (see docs/bedrock-setup.md for full setup guide)
AWS_ACCESS_KEY_ID=AKIA...        
AWS_SECRET_ACCESS_KEY=...
AWS_REGION=us-east-1
BEDROCK_S3_BUCKET=my-batch-bucket
BEDROCK_ROLE_ARN=arn:aws:iam::123456789012:role/BedrockBatchRole

# Optional overrides
OPENAI_BASE_URL=https://api.openai.com/v1
ANTHROPIC_BASE_URL=https://api.anthropic.com
GOOGLE_BASE_URL=https://generativelanguage.googleapis.com/v1beta
LANGASYNC_BATCH_POLL_INTERVAL=60.0
LANGASYNC_BASE_STORAGE_PATH=./langasync_jobs
```

Or configure programmatically:

```python
from langasync import LangasyncSettings, batch_chain

settings = LangasyncSettings(
    openai_api_key="sk-...",
    batch_poll_interval=30.0,
    base_storage_path="./my_jobs",
)
batch_wrapper = batch_chain(chain, settings)
```

### Examples

See [examples/](https://github.com/langasync/langasync/tree/master/examples) for complete working examples:

```bash
# Submit a batch job
python examples/openai_example.py run

# Fetch results (can run later, after restart)
python examples/openai_example.py fetch
```

## Development

### Setup

```bash
git clone https://github.com/langasync/langasync.git
cd langasync
pip install -e ".[dev]"
pre-commit install
```

### Running Tests

```bash
# Unit tests (mocked, fast)
pytest tests

# Integration tests (requires API keys)
pytest tests/integration -o "addopts="
```

## Contributing

Contributions are welcome!

1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Make your changes
4. Run tests (`pytest`)
5. Submit a pull request

## Community

- [GitHub Issues](https://github.com/langasync/langasync/issues) — Bug reports and feature requests
- [GitHub Discussions](https://github.com/orgs/langasync/discussions) — Questions and ideas

## License

langasync is licensed under [Apache 2.0](LICENSE). Free to use for any purpose — personal projects, commercial products, production workloads.
