Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 126 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -40,14 +40,16 @@ The `generate_summary` function will now return a string with the summary of the

## How does it work?

This library uses the OpenAI SDK to interact with LLMs. Your function returns a string that becomes the prompt, and the decorator handles calling the LLM and parsing the response.
This library uses the OpenAI SDK to interact with LLMs. Your function can return either a string (which becomes the prompt) or a list of message dictionaries (for full conversation control). The decorator handles calling the LLM and parsing the response.

The key benefits of this approach:

- **Works with any OpenAI SDK-compatible provider**: Use OpenAI, OpenRouter, or any provider with OpenAI-compatible APIs
- **Full Python control**: Build prompts using Python (no template syntax to learn)
- **Type-safe structured outputs**: Use Pydantic models for validated responses
- **Async support**: Built-in async/await support for concurrent operations
- **Conversation history**: Pass message lists for multi-turn conversations
- **Multimodal support**: Include images, audio, and video via base64 encoding
- **Simple and focused**: Does one thing well - turn functions into LLM calls

## Features
Expand Down Expand Up @@ -166,6 +168,129 @@ result = custom_prompt(
)
```

### Conversation History

Instead of returning a string, you can return a list of message dictionaries to have full control over the conversation:

```python
@backend(client, model="gpt-4o-mini")
def chat_with_history(user_message: str, conversation_history: list) -> list:
"""Chat with conversation context."""
messages = [
{"role": "system", "content": "You are a helpful assistant."},
]

# Add previous conversation
messages.extend(conversation_history)

# Add new user message
messages.append({"role": "user", "content": user_message})

return messages

# Use it with conversation history
history = [
{"role": "user", "content": "What's your name?"},
{"role": "assistant", "content": "I'm Claude, an AI assistant."},
]

response = chat_with_history("What can you help me with?", history)
print(response)
```

Note: When you return a message list, the `system` parameter in the decorator is ignored.

### Multimodal Content (Images, Audio, Video)

You can include images, audio, or video by passing them as base64-encoded content in your messages:

```python
import base64

@backend(client, model="gpt-4o-mini")
def analyze_image(image_path: str, question: str) -> list:
"""Analyze an image with a question."""
# Read and encode image
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")

return [
{
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}"
},
},
],
}
]

result = analyze_image("photo.jpg", "What's in this image?")
print(result)
```

You can also mix multiple media types:

```python
@backend(client, model="gpt-4o-mini")
def analyze_multiple_media(image1_path: str, image2_path: str) -> list:
"""Compare two images."""
# Encode images
with open(image1_path, "rb") as f:
img1 = base64.b64encode(f.read()).decode("utf-8")
with open(image2_path, "rb") as f:
img2 = base64.b64encode(f.read()).decode("utf-8")

return [
{
"role": "user",
"content": [
{"type": "text", "text": "Compare these images:"},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{img1}"},
},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{img2}"},
},
],
}
]

result = analyze_multiple_media("image1.jpg", "image2.jpg")
```

For audio content:

```python
@backend(client, model="gpt-4o-mini")
def transcribe_audio(audio_path: str) -> list:
"""Transcribe audio content."""
with open(audio_path, "rb") as f:
audio_data = base64.b64encode(f.read()).decode("utf-8")

return [
{
"role": "user",
"content": [
{"type": "text", "text": "Transcribe this audio:"},
{
"type": "input_audio",
"input_audio": {
"data": audio_data,
"format": "wav" # or "mp3", "flac", etc.
},
},
],
}
]
```

### Using OpenRouter

OpenRouter provides access to hundreds of models through an OpenAI-compatible API:
Expand Down
103 changes: 103 additions & 0 deletions demo.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
import marimo

__generated_with = "0.17.7"
app = marimo.App()


@app.cell
def _():
import marimo as mo
return (mo,)


@app.cell(hide_code=True)
def _(mo):
mo.md(r"""
## Structured Output
""")
return


@app.cell
def _():
from smartfunc import backend
from openai import OpenAI
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv(".env")

client = OpenAI()

class Summary(BaseModel):
summary: str
pros: list[str]
cons: list[str]
return Summary, backend, client


@app.cell
def _(Summary, backend, client):
@backend(client, model="gpt-4o-mini", response_format=Summary)
def analyze_pokemon(name: str) -> str:
return f"Describe the following pokemon: {name}"
return (analyze_pokemon,)


@app.cell
def _(analyze_pokemon):
result = analyze_pokemon("pikachu")
print(result.summary)
print(result.pros)
print(result.cons)
return


@app.cell(hide_code=True)
def _(mo):
mo.md(r"""
## Images

You can also return conversations, which let's you re-use OpenAI's SDK for messages.
""")
return


@app.cell
def _():
url = "https://c02.purpledshub.com/uploads/sites/41/2023/01/How-to-see-the-Wolf-Moon-in-2023--4bb6bb7.jpg?w=940&webp=1"
return (url,)


@app.cell
def _(backend, client):
@backend(client, model="gpt-4o-mini")
def desc_image(url: str) -> str:
return [
{
"role": "user",
"content": [
{"type": "text", "text": "Describe the following image:"},
{
"type": "image_url",
"image_url": {"url": url},
},
],
}
]
return (desc_image,)


@app.cell
def _(desc_image, url):
desc_image(url)
return


@app.cell
def _():
return


if __name__ == "__main__":
app.run()
73 changes: 46 additions & 27 deletions smartfunc/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from functools import wraps
from typing import Any, Callable, Optional, Type, Union
from typing import Any, Callable, Optional, Type, Union, List, Dict
from pydantic import BaseModel
from openai import OpenAI, AsyncOpenAI

Expand Down Expand Up @@ -47,13 +47,17 @@ class backend:
"""Synchronous backend decorator for LLM-powered functions.

This class provides a decorator that transforms a function into an LLM-powered
endpoint. The function should return a string that will be used as the prompt,
and the decorator handles calling the LLM and parsing the response.
endpoint. The function can return either:
- A string that will be used as the user prompt
- A list of message dictionaries for full conversation control

The decorator handles calling the LLM and parsing the response.

Features:
- Works with any OpenAI SDK-compatible provider (OpenAI, OpenRouter, etc.)
- Optional structured output validation using Pydantic models
- Full control over prompt generation using Python
- Support for multimodal content (images, audio, video via base64)

Example:
from openai import OpenAI
Expand Down Expand Up @@ -100,21 +104,26 @@ def __init__(
def __call__(self, func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
# Call the function to get the prompt
prompt = func(*args, **kwargs)

if not isinstance(prompt, str):
# Call the function to get the prompt or messages
result = func(*args, **kwargs)

# Handle different return types
if isinstance(result, str):
# String: build messages with optional system prompt
messages = []
if self.system:
messages.append({"role": "system", "content": self.system})
messages.append({"role": "user", "content": result})
elif isinstance(result, list):
# List of messages: use directly
# System prompt is ignored if messages are provided
messages = result
else:
raise ValueError(
f"Function {func.__name__} must return a string prompt, "
f"got {type(prompt).__name__}"
f"Function {func.__name__} must return either a string prompt "
f"or a list of message dictionaries, got {type(result).__name__}"
)

# Build messages array
messages = []
if self.system:
messages.append({"role": "system", "content": self.system})
messages.append({"role": "user", "content": prompt})

# Prepare API call kwargs
call_kwargs = {
"model": self.model,
Expand Down Expand Up @@ -170,10 +179,15 @@ class async_backend:
Use this when you need non-blocking LLM operations, typically in async web
applications or for concurrent processing.

The function can return either:
- A string that will be used as the user prompt
- A list of message dictionaries for full conversation control

Features:
- Async/await support for non-blocking operations
- Works with any OpenAI SDK-compatible provider
- Optional structured output validation using Pydantic models
- Support for multimodal content (images, audio, video via base64)

Example:
from openai import AsyncOpenAI
Expand Down Expand Up @@ -219,21 +233,26 @@ def __init__(
def __call__(self, func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args, **kwargs):
# Call the function to get the prompt
prompt = func(*args, **kwargs)

if not isinstance(prompt, str):
# Call the function to get the prompt or messages
result = func(*args, **kwargs)

# Handle different return types
if isinstance(result, str):
# String: build messages with optional system prompt
messages = []
if self.system:
messages.append({"role": "system", "content": self.system})
messages.append({"role": "user", "content": result})
elif isinstance(result, list):
# List of messages: use directly
# System prompt is ignored if messages are provided
messages = result
else:
raise ValueError(
f"Function {func.__name__} must return a string prompt, "
f"got {type(prompt).__name__}"
f"Function {func.__name__} must return either a string prompt "
f"or a list of message dictionaries, got {type(result).__name__}"
)

# Build messages array
messages = []
if self.system:
messages.append({"role": "system", "content": self.system})
messages.append({"role": "user", "content": prompt})

# Prepare API call kwargs
call_kwargs = {
"model": self.model,
Expand Down
4 changes: 2 additions & 2 deletions tests/test_basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,14 +82,14 @@ def generate(prompt: str) -> str:


def test_function_must_return_string(mock_client_factory):
"""Test that function must return a string."""
"""Test that function must return a string or list."""
client = mock_client_factory()

@backend(client, model="gpt-4o-mini")
def bad_function() -> str:
return 123 # Not a string!

with pytest.raises(ValueError, match="must return a string prompt"):
with pytest.raises(ValueError, match="must return either a string prompt or a list"):
bad_function()


Expand Down
Loading