diff --git a/README.md b/README.md index ae16d29..25bb480 100644 --- a/README.md +++ b/README.md @@ -43,12 +43,22 @@ We see agents as the intersection of an LLM, storage, and orchestration. [Pixelt - **Observability**: Complete traceability with automatic logging of messages, tool calls, and performance metrics - **Agentic Extensions**: Add reasoning, reflection, memory, knowledge, and team workflows. -## Connect blueprints to Cursor, Windsurf, Cline: +## Use with Cursor, Windsurf, Cline -- **[Anthropic](https://github.com/pixeltable/pixelagent/blob/main/blueprints/single_provider/anthropic/README.md)** -- **[OpenAI](https://github.com/pixeltable/pixelagent/blob/main/blueprints/single_provider/openai/README.md)** -- **[AWS Bedrock](https://github.com/pixeltable/pixelagent/blob/main/blueprints/single_provider/bedrock/README.md)** -- **[Multi-provider](https://github.com/pixeltable/pixelagent#readme)** - `pip install pixelagent` gives you all four providers; see the Quick Start above. +Install the package and point your editor at it: + +```bash +pip install pixelagent +``` + +All four providers ship in the package, so an agent is the Quick Start above: + +```python +from pixelagent.openai import Agent # or .anthropic, .bedrock, .gemini + +agent = Agent(name="my_agent", system_prompt="You are a helpful assistant.") +print(agent.chat("Hello")) +``` ## Plug-and-Play Extensions @@ -95,6 +105,8 @@ from pixelagent.anthropic import Agent import yfinance as yf # Define a tool as a UDF +# In a module, e.g. tools.py -- pixeltable rejects a @pxt.udf defined in a +# script's global namespace, because it must be importable by name. @pxt.udf def stock_price(ticker: str) -> dict: """Get stock information for a ticker symbol""" @@ -140,91 +152,102 @@ conversational_agent = Agent( # ReAct pattern for step-by-step reasoning and planning import re from datetime import datetime -from pixelagent.openai import Agent + import pixeltable as pxt -# Define a tool -@pxt.udf -def stock_info(ticker: str) -> dict: - """Get stock information for analysis""" - import yfinance as yf - stock = yf.Ticker(ticker) - return stock.info +# Tools live in a module, not beside the agent: pixeltable needs a UDF to be +# importable by name so a stored computed column can find it again. +from react_tools import stock_info + +from pixelagent.openai import Agent -# ReAct system prompt with structured reasoning pattern REACT_PROMPT = """ Today is {date} IMPORTANT: You have {max_steps} maximum steps. You are on step {step}. -Follow this EXACT step-by-step reasoning and action pattern: - 1. THOUGHT: Think about what information you need to answer the question. -2. ACTION: Either use a tool OR write "FINAL" if you're ready to give your final answer. +2. ACTION: Either use a tool OR write "FINAL" if you are ready to answer. Available tools: {tools} -Always structure your response with these exact headings: - THOUGHT: [your reasoning] ACTION: [tool_name] OR simply write "FINAL" """ -# Helper function to extract sections from responses + def extract_section(text, section_name): pattern = rf'{section_name}:?\s*(.*?)(?=\n\s*(?:THOUGHT|ACTION):|$)' match = re.search(pattern, text, re.DOTALL | re.IGNORECASE) return match.group(1).strip() if match else "" -# Execute ReAct planning loop + +# One agent, built once. The system prompt is per-turn data, so a step that +# needs a different prompt overrides it on the call rather than rebuilding the +# agent -- which is what this example used to do, and which now raises +# SchemaConflict if the model or toolset differs. +agent = Agent( + name="financial_planner", + system_prompt="You are a financial analyst.", + tools=pxt.tools(stock_info), + reset=True, +) + + def run_react_loop(question, max_steps=5): - step = 1 - while step <= max_steps: - # Dynamic system prompt with current step - react_system_prompt = REACT_PROMPT.format( - date=datetime.now().strftime("%Y-%m-%d"), - tools=["stock_info"], - step=step, - max_steps=max_steps, - ) - - # Agent with updated system prompt - agent = Agent( - name="financial_planner", - system_prompt=react_system_prompt, - reset=False, # Maintain memory between steps + for step in range(1, max_steps + 1): + response = agent.chat( + question, + system_prompt=REACT_PROMPT.format( + date=datetime.now().strftime("%Y-%m-%d"), + tools=["stock_info"], + step=step, + max_steps=max_steps, + ), ) - - # Get agent's response for current step - response = agent.chat(question) - - # Extract action to determine next step action = extract_section(response, "ACTION") - - # Check if agent is ready for final answer if "FINAL" in action.upper(): break - - # Call tool if needed if "stock_info" in action.lower(): - tool_agent = Agent( - name="financial_planner", - tools=pxt.tools(stock_info) - ) - tool_agent.tool_call(question) - - step += 1 - - # Generate final recommendation - return Agent(name="financial_planner").chat(question) - -# Run the planning loop + agent.tool_call(question) + + return agent.chat(question) + + recommendation = run_react_loop("Create an investment recommendation for AAPL") ``` Check out our [tutorials](examples/) for more examples including reflection loops, planning patterns, and multi-provider implementations. +## Migrating from 0.1.x + +0.2.0 runs on Pixeltable 0.7.6. On 0.7.3+ every 0.1.x agent was broken: a +text-only `chat()` raised `expected non-None value`, and the Anthropic agent +could not be constructed at all. Upgrade with `pip install -U pixelagent`. + +Four things changed for callers: + +- **Python 3.11+ is required.** Pixeltable dropped 3.10 in 0.7.2. +- **Rebuilding an agent under the same name with a different `model` or + `tools` now raises `SchemaConflict`.** Previously the new argument was + silently discarded and the agent kept using the old model while reporting + the new one. Pass `reset=True` to rebuild, or use a different name. +- **`system_prompt`, `model_kwargs`, `max_tokens` and `n_latest_messages` are + per-turn data**, so `chat()` takes them as overrides: + `agent.chat(msg, system_prompt="...")`. Varying a prompt no longer needs a + second agent. +- **`chat()` gained `conversation_id`** (default `"default"`), so one agent + can hold several separate threads. + +`Agent(...)`, `.chat()`, `.tool_call()` and the `.memory` / `.agent` +tables are otherwise unchanged. + +Two bugs are fixed that needed no API change: a failed turn no longer leaves an +unanswered message in memory, and `chat()` no longer re-queries the row it just +inserted. + + ## Tutorials and Examples - **Basics**: Check out [Getting Started](examples/getting-started/pixelagent_basics_tutorial.py) for a step-by-step introduction to core concepts @@ -232,4 +255,4 @@ Check out our [tutorials](examples/) for more examples including reflection loop - **Specialized Directories**: Browse our example directories for deeper implementations of specific techniques -Ready to start building? Dive into the blueprints, tweak them to your needs, and let Pixeltable handle the AI data infrastructure while you focus on innovation! +Ready to start building? `pip install pixelagent`, work through the examples, and let Pixeltable handle the AI data infrastructure while you focus on your agent. diff --git a/blueprints/single_provider/anthropic/README.md b/blueprints/single_provider/anthropic/README.md deleted file mode 100644 index 227c3bd..0000000 --- a/blueprints/single_provider/anthropic/README.md +++ /dev/null @@ -1,322 +0,0 @@ -# Building an Anthropic Agent in Pixeltable: A Step-by-Step Guide - -This tutorial walks you through building `agent.py`, a persistent conversational agent with memory using Pixeltable's automated data orchestration and storage—powered by Anthropic's Claude model. We'll build the agent step by step, explaining each component as we go. By the end, you'll have a complete implementation that can handle both chat and tool-calling capabilities. - -The final code is available in `agent.py`, but we recommend following the tutorial to understand how each piece fits together. - -## Prerequisites -- Install required packages: - ```bash - pip install pixeltable anthropic - ``` -- Set up your Anthropic API key: - ```bash - export ANTHROPIC_API_KEY='your-api-key' - ``` - -## Step 1: Understanding the Building Blocks - -The agent is built on three core components that inherit from the shared `BaseAgent` class: - -1. **Memory Management**: Using Pixeltable tables to store conversation history with optional message limits -2. **Chat Pipeline**: A series of computed columns that process user input and generate responses -3. **Tool Execution**: Optional support for executing functions/tools during conversations - -Key classes and functions: - -- `Agent`: Main class that inherits from `BaseAgent` and implements Anthropic-specific logic -- `create_messages()`: UDF that formats conversation history for Claude, with support for images -- Pixeltable tables (inherited from `BaseAgent`): - - `memory`: Stores all conversation history - - `agent`: Manages chat interactions - - `tools`: Handles tool execution (if enabled) - -Here's the basic setup with imports and the Agent class initialization: - -```python -import base64 -import io -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf - -try: - from pixeltable.functions.anthropic import invoke_tools, messages -except ImportError: - raise ImportError("anthropic not found; run `pip install anthropic`") - -@pxt.udf -def create_messages(memory_context: list[dict], current_message: str, image: Optional[PIL.Image.Image] = None) -> list[dict]: - """Helper UDF to format conversation history and current message for Claude, with optional image support""" - messages = memory_context.copy() - - # For text-only messages - if not image: - messages.append({"role": "user", "content": current_message}) - return messages - - # For messages with images - # Creates content blocks with text and image in base64 format - content_blocks = [ - {"type": "text", "text": current_message}, - {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg"}} - ] - messages.append({"role": "user", "content": content_blocks}) - return messages - -class Agent: - def __init__( - self, - name: str, - system_prompt: str, - model: str = "claude-3-5-sonnet-latest", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages # None for unlimited history - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} -``` - -## Step 2: Building the Memory and Chat Pipeline - -The memory system consists of two main parts inherited from `BaseAgent`: - -1. **Table Setup** (`_setup_tables()`): - - Creates the `memory` table for persistent conversation history - - Creates the `agent` table for managing chat completions - - Optionally creates the `tools` table for tool execution - - Uses Pixeltable's automated data orchestration - -2. **Chat Pipeline** (`_setup_chat_pipeline()`): - - Uses computed columns to create a 4-step processing pipeline: - 1. Retrieves recent messages based on `n_latest_messages` limit (supports unlimited history when set to None) - 2. Formats messages with system prompt for Claude - 3. Gets completion from Anthropic API - 4. Extracts the final response text - -Here's how we implement the tables and pipeline: - -```python -def _setup_tables(self): - """Initialize the required Pixeltable tables for the agent""" - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for Claude - "image": PIL.Image.Image, # Optional image for multimodal input - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for Claude - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - -def _setup_chat_pipeline(self): - """Configure the chat completion pipeline using Pixeltable computed columns""" - # Get recent messages from memory - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - """Get recent messages from memory, respecting n_latest_messages limit if set""" - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Add computed columns to process chat completion - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), - if_exists="ignore", - ) - - # Create messages for Claude - self.agent.add_computed_column( - messages=create_messages( - self.agent.memory_context, self.agent.user_message - ), - if_exists="ignore", - ) - - # Get Anthropic's API response - self.agent.add_computed_column( - api_response=messages( - messages=self.agent.messages, - model=self.model, - system=self.system_prompt, - **self.chat_kwargs, - ), - if_exists="ignore", - ) - - # Parse Claude's response - self.agent.add_computed_column( - agent_response=self.agent.api_response.content[0].text, - if_exists="ignore", - ) -``` - -## Step 3: Adding Tool-Calling Support - -Tool calling is implemented as a three-stage process using Pixeltable's computed columns: - -1. **Initial Response**: - - Sends the user's request to Claude with available tools - - Claude decides if and which tools to call - -2. **Tool Execution**: - - `invoke_tools()` executes the requested tools - - Results are stored in the tools table - -3. **Final Response**: - - Tool results are sent back to Claude - - Claude generates a final response incorporating tool outputs - -Here's the implementation: - -```python -def _setup_tools_pipeline(self): - """Configure the tool call handshake pipeline""" - # Get initial response from Claude with tool calls - self.tools_table.add_computed_column( - initial_response=messages( - model=self.model, - system=self.system_prompt, - messages=[{"role": "user", "content": self.tools_table.tool_prompt}], - tools=self.tools, - **self.tool_kwargs, - ), - if_exists="ignore", - ) - - # Execute the tools - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - - # Pass the tool results back to Claude - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - - # Get final API response with tool results - self.tools_table.add_computed_column( - final_response=messages( - model=self.model, - system=self.system_prompt, - messages=[{"role": "user", "content": self.tools_table.tool_response_prompt}], - **self.tool_kwargs, - ), - if_exists="ignore", - ) - - # Parse Claude's response - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response.content[0].text, - if_exists="ignore", - ) -``` - -## Using the Agent - -To use the agent in your code: - -1. **Initialize the Agent**: -```python -agent = Agent( - name="my_agent", - system_prompt="You are a helpful assistant", - model="claude-3-5-sonnet-latest", # Latest Claude model - n_latest_messages=10, # None for unlimited history -) -``` - -2. **Chat with the Agent**: -```python -# Simple text chat -response = agent.chat("Hello, how are you?") - -# Chat with an image -from PIL import Image -img = Image.open("path/to/image.jpg") -response = agent.chat("What's in this image?", image=img) - -# Chat with tool execution -response = agent.tool_call("What's the weather in New York?") -``` - -3. **Access History**: -```python -# View recent conversations -memory = pxt.get_table("my_agent.memory") -print(memory.collect()) -``` - -## Key Features - -1. **Shared Base Implementation**: - - Inherits from `BaseAgent` class for common functionality - - Reduces code duplication with OpenAI agent - -2. **Flexible Memory Management**: - - Optional message limit with `n_latest_messages` - - Support for unlimited history when set to None - - Support for multimodal conversations with image handling - -3. **Automated Data Management**: - - Persistent storage using Pixeltable - - Automatic dependency tracking - - Efficient query optimization - -For more details and advanced usage, refer to the complete implementation in `agent.py`. - -## Testing the Agent - -You can test this implementation using the provided `test.py` script, which demonstrates both chat and tool-calling capabilities of the agent. \ No newline at end of file diff --git a/blueprints/single_provider/anthropic/agent.py b/blueprints/single_provider/anthropic/agent.py deleted file mode 100644 index 7e28b62..0000000 --- a/blueprints/single_provider/anthropic/agent.py +++ /dev/null @@ -1,418 +0,0 @@ -""" -Tutorial: Building Your Own Anthropic-powered Agent with Pixeltable - -This example demonstrates how to create a conversational AI agent using OpenAI's LLM model -and Pixeltable for persistent memory, storage, orchestration, and tool execution. The agent can maintain conversation -history and execute tools while keeping track of all interactions in a structured database. -""" - -import base64 -import io -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf - -try: - from pixeltable.functions.anthropic import invoke_tools, messages -except ImportError: - raise ImportError("anthropic not found; run `pip install anthropic`") - - -# Helper UDF to format conversation history and current message for LLM -@pxt.udf -def create_messages( - memory_context: list[dict], - current_message: str, - image: Optional[PIL.Image.Image] = None, -) -> list[dict]: - - # Create a copy to avoid modifying the original - messages = memory_context.copy() - - # For text-only messages - if not image: - messages.append({"role": "user", "content": current_message}) - return messages - - # Convert image to base64 - bytes_arr = io.BytesIO() - image.save(bytes_arr, format="JPEG") - b64_bytes = base64.b64encode(bytes_arr.getvalue()) - b64_encoded_image = b64_bytes.decode("utf-8") - - # Create content blocks with text and image - content_blocks = [ - {"type": "text", "text": current_message}, - { - "type": "image", - "source": { - "type": "base64", - "media_type": "image/jpeg", - "data": b64_encoded_image, - }, - }, - ] - - messages.append({"role": "user", "content": content_blocks}) - - return messages - - -class Agent: - """ - An AI agent powered by Anthropic's LLM model with persistent memory and tool execution capabilities. - - The agent maintains three key tables in Pixeltable: - 1. memory: Stores all conversation history with timestamps - 2. agent: Manages chat interactions and responses - 3. tools: (Optional) Handles tool execution and responses - - Key Features: - - Persistent conversation memory with optional message limit - - Tool execution support - - Structured data storage and orchestration using Pixeltable - """ - - def __init__( - self, - name: str, - system_prompt: str, - model: str = "claude-3-5-sonnet-latest", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - """ - Initialize the agent with the specified configuration. - - Args: - name: Unique name for the agent (used for table names) - system_prompt: System prompt that guides LLM's behavior - model: LLM model to use (defaults to claude-3-5-sonnet-latest) - n_latest_messages: Number of recent messages to include in context (None for unlimited) - tools: Optional tools configuration for function calling - reset: If True, deletes existing agent data - chat_kwargs: Additional kwargs for chat completion - tool_kwargs: Additional kwargs for tool execution - """ - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} - - # Set up or reset the agent's database - if reset: - pxt.drop_dir(self.directory, if_not_exists="ignore", force=True) - - # Create agent directory if it doesn't exist - pxt.create_dir(self.directory, if_exists="ignore") - - # Set up tables - self._setup_tables() - - # Get references to the created tables - self.memory = pxt.get_table(f"{self.directory}.memory") - self.agent = pxt.get_table(f"{self.directory}.agent") - self.tools_table = ( - pxt.get_table(f"{self.directory}.tools") if self.tools else None - ) - - def _setup_tables(self): - """ - Initialize the required Pixeltable tables for the agent. - Creates three tables: - 1. memory: Stores conversation history - 2. agent: Manages chat completions - 3. tools: (Optional) Handles tool execution - """ - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for LLM - "image": pxt.Image, # Optional image attachment - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for LLM - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - - def _setup_chat_pipeline(self): - """ - Configure the chat completion pipeline using Pixeltable computed columns. - This sets up a series of sequential computations that: - 1. Retrieve recent conversation history - 2. Format messages for LLM - 3. Get LLM's response - 4. Extract the response text - """ - - # Get recent messages from memory - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - """Get recent messages from memory, respecting n_latest_messages limit if set""" - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Add computed columns to process chat completion - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), if_exists="ignore" - ) - - # Create messages for LLM - self.agent.add_computed_column( - messages=create_messages( - self.agent.memory_context, self.agent.user_message, self.agent.image - ), - if_exists="ignore", - ) - - # Get Antropics API response - self.agent.add_computed_column( - api_response=messages( - messages=self.agent.messages, - model=self.model, - system=self.system_prompt, - **self.chat_kwargs, - ), - if_exists="ignore", - ) - - # Parse LLM's response - self.agent.add_computed_column( - agent_response=self.agent.api_response.content[0].text, if_exists="ignore" - ) - - def _setup_tools_pipeline(self): - """ - Configure the tool call handshake pipeline using Pixeltable computed columns. - This sets up a series of transformations that: - 1. Get initial response from LLM with tool calls - 2. Execute the tools - 3. Get final response from LLM with tool results - """ - # Get initial response from LLM with tool calls - self.tools_table.add_computed_column( - initial_response=messages( - model=self.model, - system=self.system_prompt, - messages=[{"role": "user", "content": self.tools_table.tool_prompt}], - tools=self.tools, - **self.tool_kwargs, - ), - if_exists="ignore", - ) - - # Execute the tools - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - - # Pass the tool results back to LLM for final response - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - - # Get final api response from Anthropic with tool results - self.tools_table.add_computed_column( - final_response=messages( - model=self.model, - system=self.system_prompt, - messages=[ - {"role": "user", "content": self.tools_table.tool_response_prompt} - ], - **self.tool_kwargs, - ), - if_exists="ignore", - ) - - # Parse LLM's response - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response.content[0].text, - if_exists="ignore", - ) - - def chat(self, message: str, image: Optional[PIL.Image.Image] = None) -> str: - """ - Send a message to the agent and get its response. - - This method: - 1. Stores the user message in memory - 2. Triggers the chat completion pipeline - 3. Stores the assistant's response in memory - 4. Returns the response - - Args: - message: The user's message - image: Optional image attachment - - Returns: - The agent's response - """ - now = datetime.now() - - # Generate unique IDs for the message pair - user_message_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": message, - "timestamp": now, - } - ] - ) - - # Store user message in agent table (which triggers the chat pipeline) - self.agent.insert( - [ - { - "message_id": user_message_id, - "user_message": message, - "timestamp": now, - "system_prompt": self.system_prompt, - "image": image, - } - ] - ) - - # Get LLM's response from agent table - result = ( - self.agent.select(self.agent.agent_response) - .where(self.agent.message_id == user_message_id) - .collect() - ) - response = result["agent_response"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": response, - "timestamp": now, - } - ] - ) - return response - - def tool_call(self, prompt: str) -> str: - """ - Execute a tool call with the given prompt. - - This method: - 1. Stores the user prompt in memory - 2. Triggers the tool call handshake pipeline - 3. Stores the tool's response in memory - 4. Returns the response - - Args: - prompt: The user's prompt - - Returns: - The tool's response - """ - if not self.tools: - return "No tools configured for this agent." - - now = datetime.now() - user_message_id = str(uuid4()) - tool_invoke_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": prompt, - "timestamp": now, - } - ] - ) - - # Store user prompt in tools table (which triggers the tool call handshake pipeline) - self.tools_table.insert( - [ - { - "tool_invoke_id": tool_invoke_id, - "tool_prompt": prompt, - "timestamp": now, - } - ] - ) - - # Get tool answer from tools table - result = ( - self.tools_table.select(self.tools_table.tool_answer) - .where(self.tools_table.tool_invoke_id == tool_invoke_id) - .collect() - ) - tool_answer = result["tool_answer"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": tool_answer, - "timestamp": now, - } - ] - ) - return tool_answer diff --git a/blueprints/single_provider/anthropic/test.py b/blueprints/single_provider/anthropic/test.py deleted file mode 100644 index c13e4ee..0000000 --- a/blueprints/single_provider/anthropic/test.py +++ /dev/null @@ -1,26 +0,0 @@ -import pixeltable as pxt -from agent import Agent - - -@pxt.udf -def weather(city: str) -> str: - """ - Returns the weather in a given city. - """ - return f"The weather in {city} is sunny." - - -# Create an agent -agent = Agent( - name="anthropic_agent", - system_prompt="You’re my assistant.", - tools=pxt.tools(weather), - reset=True, -) - -# Persistant chat and memory -print(agent.chat("Hi, how are you?")) -print(agent.chat("What was my last question?")) - -# Tool call -print(agent.tool_call("Get weather in San Francisco")) diff --git a/blueprints/single_provider/bedrock/README.md b/blueprints/single_provider/bedrock/README.md deleted file mode 100644 index ebfe94c..0000000 --- a/blueprints/single_provider/bedrock/README.md +++ /dev/null @@ -1,197 +0,0 @@ -# AWS Bedrock Agent Blueprint - -This blueprint demonstrates how to create a conversational AI agent powered by AWS Bedrock models using Pixeltable for persistent memory, storage, orchestration, and tool execution. - -## Features - -- **Persistent Memory**: Maintains conversation history in a structured database -- **Unlimited Memory Support**: Option to maintain complete conversation history without limits -- **Tool Execution**: Supports function calling with AWS Bedrock models -- **Image Support**: Can process and respond to images -- **Configurable Context Window**: Control how many previous messages are included in the context -- **Modular Architecture**: Built on a flexible BaseAgent design for maintainability and extensibility - -## Prerequisites - -- Python 3.8+ -- Pixeltable (`pip install pixeltable`) -- AWS Boto3 (`pip install boto3`) -- AWS credentials configured (via AWS CLI, environment variables, or IAM role) - -## AWS Credentials Setup - -Before using this blueprint, ensure you have: - -1. An AWS account with access to AWS Bedrock -2. Proper IAM permissions to use Bedrock models -3. AWS credentials configured using one of these methods: - - AWS CLI: Run `aws configure` - - Environment variables: Set `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and `AWS_REGION` - - IAM role (if running on AWS infrastructure) - -## Quick Start - -```python -from blueprints.single_provider.bedrock.agent import Agent - -# Create a simple chat agent -agent = Agent( - name="my_bedrock_agent", - system_prompt="You are a helpful assistant.", - model="amazon.nova-pro-v1:0", # You can use other Bedrock models like "anthropic.claude-3-sonnet-20240229-v1:0" - n_latest_messages=10, # Number of recent messages to include in context (set to None for unlimited) - reset=True # Start with a fresh conversation history -) - -# Chat with the agent -response = agent.chat("Hello, who are you?") -print(response) - -# Chat with an image -from PIL import Image -img = Image.open("path/to/image.jpg") -response = agent.chat("What's in this image?", image=img) -print(response) -``` - -## Tool Execution Example - -```python -import pixeltable as pxt - -# Define tools -weather_tools = pxt.tools([ - { - "type": "function", - "function": { - "name": "get_weather", - "description": "Get the current weather for a location", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "The city and state, e.g. San Francisco, CA" - } - }, - "required": ["location"] - } - } - } -]) - -# Define tool implementation -def get_weather(location): - # In a real application, this would call a weather API - return f"It's sunny and 72°F in {location}" - -# Create agent with tools -agent = Agent( - name="weather_assistant", - system_prompt="You are a helpful weather assistant.", - model="amazon.nova-pro-v1:0", - tools=weather_tools, - reset=True -) - -# Register tool implementation -weather_tools.register_tool("get_weather", get_weather) - -# Use tool calling -response = agent.tool_call("What's the weather like in Seattle?") -print(response) -``` - -## Available Bedrock Models - -This blueprint works with various AWS Bedrock models, including: - -- `amazon.nova-pro-v1:0` (default) -- `anthropic.claude-3-sonnet-20240229-v1:0` -- `anthropic.claude-3-opus-20240229-v1:0` -- `anthropic.claude-3-haiku-20240307-v1:0` -- `meta.llama3-70b-instruct-v1:0` -- `meta.llama3-8b-instruct-v1:0` - -Note that different models have different capabilities and pricing. Refer to the [AWS Bedrock documentation](https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html) for more details. - -## Advanced Configuration - -You can pass additional parameters to the Bedrock API using the `chat_kwargs` and `tool_kwargs` parameters: - -```python -agent = Agent( - name="advanced_agent", - system_prompt="You are a helpful assistant.", - model="amazon.nova-pro-v1:0", - chat_kwargs={ - "temperature": 0.7, - "max_tokens": 1000 - }, - tool_kwargs={ - "temperature": 0.2 # Lower temperature for more deterministic tool calls - } -) -``` - -## Unlimited Memory Support - -You can create an agent with unlimited conversation history by setting `n_latest_messages=None`: - -```python -# Create an agent with unlimited memory -agent = Agent( - name="memory_agent", - system_prompt="You are a helpful assistant with perfect memory.", - model="amazon.nova-pro-v1:0", - n_latest_messages=None, # No limit on conversation history - reset=True -) - -# The agent will remember all previous interactions -response1 = agent.chat("My name is Alice.") -response2 = agent.chat("What's my name?") # Agent will remember -response3 = agent.chat("Do you remember my name?") # Agent will still remember -``` - -This is particularly useful for applications that require long-term memory or context awareness. - -## Architecture - -This blueprint uses a modular architecture with: - -1. **BaseAgent**: An abstract base class that handles common functionality like: - - Table setup and management - - Memory persistence - - Core chat and tool call implementations - -2. **Agent**: The Bedrock-specific implementation that inherits from BaseAgent and implements: - - Bedrock-specific message formatting - - Bedrock API integration - - Tool calling for Bedrock models - -This architecture makes the code more maintainable and extensible. - -## How It Works - -The agent uses Pixeltable to create and manage three tables: - -1. **memory**: Stores all conversation history with timestamps -2. **agent**: Manages chat interactions and responses -3. **tools**: (Optional) Handles tool execution and responses - -When you send a message to the agent, it: - -1. Stores your message in the memory table -2. Triggers a pipeline that retrieves recent conversation history -3. Formats the messages for the Bedrock API (with proper content structure) -4. Gets a response from the Bedrock model -5. Stores the response in memory -6. Returns the response to you - -Tool execution follows a similar pattern but uses a specialized pipeline: -1. The user's prompt is sent to Bedrock with available tools -2. Bedrock decides which tools to call and with what parameters -3. The tools are executed and their results are returned -4. The results are sent back to Bedrock for a final response -5. The response is stored in memory and returned to the user diff --git a/blueprints/single_provider/bedrock/__init__.py b/blueprints/single_provider/bedrock/__init__.py deleted file mode 100644 index a6df24a..0000000 --- a/blueprints/single_provider/bedrock/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .agent import Agent - -__all__ = ["Agent"] diff --git a/blueprints/single_provider/bedrock/agent.py b/blueprints/single_provider/bedrock/agent.py deleted file mode 100644 index f855a39..0000000 --- a/blueprints/single_provider/bedrock/agent.py +++ /dev/null @@ -1,455 +0,0 @@ -""" -Tutorial: Building Your Own AWS Bedrock-powered Agent with Pixeltable - -This example demonstrates how to create a conversational AI agent using AWS Bedrock LLM models -and Pixeltable for persistent memory, storage, orchestration, and tool execution. The agent can maintain conversation -history and execute tools while keeping track of all interactions in a structured database. -""" - -from abc import ABC, abstractmethod -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf -from PIL import Image - -from utils import create_messages - -try: - from pixeltable.functions.bedrock import converse, invoke_tools -except ImportError: - raise ImportError("boto3 not found; run `pip install boto3`") - - -class BaseAgent(ABC): - """ - An Base agent powered by LLM model with persistent memory and tool execution capabilities. - - This base agent gets inherited by the Bedrock Agent class. - - The agent maintains three key tables in Pixeltable: - 1. memory: Stores all conversation history with timestamps - 2. agent: Manages chat interactions and responses - 3. tools: (Optional) Handles tool execution and responses - - Key Features: - - Persistent conversation memory with optional message limit - - Tool execution support - - Structured data storage and orchestration using Pixeltable - """ - - def __init__( - self, - name: str, - system_prompt: str, - model: str, - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - """ - Initialize the agent with the specified configuration. - - Args: - name: Unique name for the agent (used for table names) - system_prompt: System prompt that guides LLM's behavior - model: LLM model to use - n_latest_messages: Number of recent messages to include in context (None for unlimited) - tools: Optional tools configuration for function calling - reset: If True, deletes existing agent data - chat_kwargs: Additional kwargs for chat completion - tool_kwargs: Additional kwargs for tool execution - """ - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} - - # Set up or reset the agent's database - if reset: - pxt.drop_dir(self.directory, if_not_exists="ignore", force=True) - - # Create agent directory if it doesn't exist - pxt.create_dir(self.directory, if_exists="ignore") - - # Set up tables - self._setup_tables() - - # Get references to the created tables - self.memory = pxt.get_table(f"{self.directory}.memory") - self.agent = pxt.get_table(f"{self.directory}.agent") - self.tools_table = ( - pxt.get_table(f"{self.directory}.tools") if self.tools else None - ) - - def _setup_tables(self): - """ - Initialize the required Pixeltable tables for the agent. - Creates three tables: - 1. memory: Stores conversation history - 2. agent: Manages chat completions - 3. tools: (Optional) Handles tool execution - """ - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for Claude - "image": pxt.Image, # Optional image attachment - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for Claude - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - # Set up tools pipeline - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - - @abstractmethod - def _setup_chat_pipeline(self): - """To be implemented by subclasses""" - raise NotImplementedError - - @abstractmethod - def _setup_tools_pipeline(self): - """To be implemented by subclasses""" - raise NotImplementedError - - def chat(self, message: str, image: Optional[Image.Image] = None) -> str: - """ - Send a message to the agent and get its response. - - This method: - 1. Stores the user message in memory - 2. Triggers the chat completion pipeline - 3. Stores the assistant's response in memory - 4. Returns the response - - Args: - message: The user's message - image: Optional image attachment - - Returns: - The agent's response - """ - now = datetime.now() - - # Generate unique IDs for the message pair - user_message_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": message, - "timestamp": now, - } - ] - ) - - # Store user message in agent table (which triggers the chat pipeline) - self.agent.insert( - [ - { - "message_id": user_message_id, - "user_message": message, - "timestamp": now, - "system_prompt": self.system_prompt, - "image": image, - } - ] - ) - - # Get LLM's response from agent table - result = ( - self.agent.select(self.agent.agent_response) - .where(self.agent.message_id == user_message_id) - .collect() - ) - response = result["agent_response"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": response, - "timestamp": now, - } - ] - ) - return response - - def tool_call(self, prompt: str) -> str: - """ - Execute a tool call with the given prompt. - - This method: - 1. Stores the user prompt in memory - 2. Triggers the tool call handshake pipeline - 3. Stores the tool's response in memory - 4. Returns the response - - Args: - prompt: The user's prompt - - Returns: - The tool's response - """ - if not self.tools: - return "No tools configured for this agent." - - # Use separate timestamps for user and assistant messages - user_timestamp = datetime.now() - user_message_id = str(uuid4()) - tool_invoke_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": prompt, - "timestamp": user_timestamp, - } - ] - ) - - # Store user prompt in tools table (which triggers the tool call handshake pipeline) - self.tools_table.insert( - [ - { - "tool_invoke_id": tool_invoke_id, - "tool_prompt": prompt, - "timestamp": user_timestamp, - } - ] - ) - - # Get tool answer from tools table - result = ( - self.tools_table.select(self.tools_table.tool_answer) - .where(self.tools_table.tool_invoke_id == tool_invoke_id) - .collect() - ) - tool_answer = result["tool_answer"][0] - - # Store LLM's response in memory with a slightly later timestamp - assistant_timestamp = datetime.now() - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": tool_answer, - "timestamp": assistant_timestamp, - } - ] - ) - return tool_answer - - -class Agent(BaseAgent): - """ - AWS Bedrock-specific implementation of the BaseAgent. - - This agent uses AWS Bedrock's Claude API for generating responses and handling tools. - It inherits common functionality from BaseAgent including: - - Table setup and management - - Memory persistence - - Base chat and tool call implementations - - The agent supports both limited and unlimited conversation history through - the n_latest_messages parameter for regular chat, while tool calls use only - the current message without conversation history. - """ - - def __init__( - self, - name: str, - system_prompt: str, - model: str = "amazon.nova-pro-v1:0", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - # Initialize the base agent with all common parameters - super().__init__( - name=name, - system_prompt=system_prompt, - model=model, - n_latest_messages=n_latest_messages, # None for unlimited history - tools=tools, - reset=reset, - chat_kwargs=chat_kwargs, - tool_kwargs=tool_kwargs, - ) - - def _setup_chat_pipeline(self): - """ - Configure the chat completion pipeline using Pixeltable's computed columns. - This method implements the abstract method from BaseAgent. - - The pipeline consists of 4 steps: - 1. Retrieve recent messages from memory - 2. Format messages for Bedrock Claude - 3. Get completion from Bedrock - 4. Extract the response text - - Note: The pipeline automatically handles memory limits based on n_latest_messages. - When set to None, it maintains unlimited conversation history. - """ - - # Step 1: Define a query to get recent messages - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - """ - Get recent messages from memory, respecting n_latest_messages limit if set. - Messages are ordered by timestamp (newest first). - Returns all messages if n_latest_messages is None. - """ - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Step 2: Add computed columns to process the conversation - # First, get the conversation history - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), if_exists="ignore" - ) - - # Format messages for Bedrock Claude - self.agent.add_computed_column( - messages=create_messages( - self.agent.memory_context, - self.agent.user_message, - self.agent.image, - ), - if_exists="ignore", - ) - - # Get Bedrock Claude's API response - self.agent.add_computed_column( - response=converse( - messages=self.agent.messages, - model_id=self.model, - system=[{"text": self.system_prompt}], - **self.chat_kwargs, - ), - if_exists="ignore", - ) - - # Extract the final response text from Bedrock Claude's specific response format - self.agent.add_computed_column( - agent_response=self.agent.response.output.message.content[0].text, - if_exists="ignore" - ) - - def _setup_tools_pipeline(self): - """ - Configure the tool execution pipeline using Pixeltable's computed columns. - This method implements the abstract method from BaseAgent. - - The pipeline has 4 stages: - 1. Get initial response from Bedrock Claude with potential tool calls - 2. Execute any requested tools - 3. Format tool results for follow-up - 4. Get final response incorporating tool outputs - - Note: For tool calls, we only use the current message without conversation history - to ensure tool execution is based solely on the current request. - """ - # Stage 1: Get initial response with potential tool calls - # Note: We only use the current tool prompt without memory context - self.tools_table.add_computed_column( - initial_response=converse( - model_id=self.model, - system=[{"text": self.system_prompt}], - messages=[{"role": "user", "content": [{"text": self.tools_table.tool_prompt}]}], - tool_config=self.tools, # Pass available tools to Bedrock Claude - **self.tool_kwargs, - ), - if_exists="ignore", - ) - - # Stage 2: Execute any tools that Bedrock Claude requested - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - - # Stage 3: Format tool results for follow-up - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - - # Stage 4: Get final response incorporating tool results - # Again, we only use the current tool response without memory context - self.tools_table.add_computed_column( - final_response=converse( - model_id=self.model, - system=[{"text": self.system_prompt}], - messages=[ - {"role": "user", "content": [{"text": self.tools_table.tool_response_prompt}]} - ], - **self.tool_kwargs, - ), - if_exists="ignore", - ) - - # Extract the final response text from Bedrock Claude's format - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response.output.message.content[0].text, - if_exists="ignore", - ) diff --git a/blueprints/single_provider/bedrock/test.py b/blueprints/single_provider/bedrock/test.py deleted file mode 100644 index d44401b..0000000 --- a/blueprints/single_provider/bedrock/test.py +++ /dev/null @@ -1,92 +0,0 @@ -""" -Test script for the AWS Bedrock Agent Blueprint. - -This script demonstrates how to create and use a Bedrock-powered agent -with both basic chat functionality and tool execution. - -Prerequisites: -- AWS credentials configured (via AWS CLI, environment variables, or IAM role) -- Access to AWS Bedrock models -""" - -import pixeltable as pxt -from blueprints.single_provider.bedrock.agent import Agent - - -@pxt.udf -def stock_price(ticker: str) -> float: - """Get the current stock price for a given ticker symbol.""" - # This is a mock implementation for testing - prices = { - "AAPL": 175.34, - "MSFT": 325.89, - "GOOGL": 142.56, - "AMZN": 178.23, - "NVDA": 131.17, - } - return prices.get(ticker.upper(), 0.0) - - -def main(): - # Create a Bedrock agent with memory - agent = Agent( - name="bedrock_test", - system_prompt="You are a helpful assistant that can answer questions and use tools.", - model="amazon.nova-pro-v1:0", # Use the Amazon Nova Pro model - n_latest_messages=None, # Unlimited memory to ensure all messages are included - tools=pxt.tools(stock_price), # Register the stock_price tool - reset=True, # Reset the agent's memory for testing - ) - - print("\n=== Testing Conversational Memory ===\n") - - # First conversation turn - user_message = "Hello, my name is Alice." - print(f"User: {user_message}") - response = agent.chat(user_message) - print(f"Agent: {response}\n") - - # Second conversation turn - the agent should remember the user's name - user_message = "What's my name?" - print(f"User: {user_message}") - response = agent.chat(user_message) - print(f"Agent: {response}\n") - - print("\n=== Testing Tool Calling (No Memory) ===\n") - - # Tool call - should not use memory from previous conversation - user_message = "What is the stock price of NVDA today?" - print(f"User: {user_message}") - response = agent.tool_call(user_message) - print(f"Agent: {response}\n") - - # Another tool call - should not remember previous tool call - user_message = "What about AAPL?" - print(f"User: {user_message}") - response = agent.tool_call(user_message) - print(f"Agent: {response}\n") - - print("\n=== Testing Memory After Tool Calls ===\n") - - # Regular chat after tool calls - should still remember the user's name - user_message = "Do you still remember my name?" - print(f"User: {user_message}") - response = agent.chat(user_message) - print(f"Agent: {response}\n") - - # Check if the memory contains all the messages - print("\n=== Checking Memory Contents ===\n") - memory_contents = agent.memory.select( - agent.memory.role, - agent.memory.content - ).order_by(agent.memory.timestamp, asc=True).collect() - - print("Memory contains the following messages:") - for i in range(len(memory_contents)): - role = memory_contents["role"][i] - content = memory_contents["content"][i] - print(f"{i+1}. {role}: {content[:50]}...") - - -if __name__ == "__main__": - main() diff --git a/blueprints/single_provider/bedrock/utils.py b/blueprints/single_provider/bedrock/utils.py deleted file mode 100644 index 4a04efb..0000000 --- a/blueprints/single_provider/bedrock/utils.py +++ /dev/null @@ -1,79 +0,0 @@ -import base64 -import io -from typing import Optional - -import PIL -import pixeltable as pxt - - -@pxt.udf -def create_messages( - memory_context: list[dict], - current_message: str, - image: Optional[PIL.Image.Image] = None, -) -> list[dict]: - """ - Create a formatted message list for Bedrock Claude models. - - This function formats the conversation history and current message - into the structure expected by Bedrock Claude models. - - Args: - memory_context: List of previous messages from memory - current_message: The current user message - image: Optional image to include with the message - - Returns: - List of formatted messages for Bedrock Claude - """ - # Create a copy to avoid modifying the original and format for Bedrock - messages = [] - - # Get messages in oldest-first order - reversed_memory = list(reversed(memory_context)) - - # Ensure the conversation starts with a user message - # If the first message is from the assistant, skip it - start_idx = 0 - if reversed_memory and reversed_memory[0]["role"] == "assistant": - start_idx = 1 - - # Format previous messages for Bedrock - for msg in reversed_memory[start_idx:]: - # Convert string content to the required list format - if isinstance(msg["content"], str): - messages.append({ - "role": msg["role"], - "content": [{"text": msg["content"]}] - }) - else: - # If it's already in the correct format, keep it as is - messages.append(msg) - - # For text-only messages - if not image: - messages.append({"role": "user", "content": [{"text": current_message}]}) - return messages - - # Convert image to base64 - bytes_arr = io.BytesIO() - image.save(bytes_arr, format="JPEG") - b64_bytes = base64.b64encode(bytes_arr.getvalue()) - b64_encoded_image = b64_bytes.decode("utf-8") - - # Create content blocks with text and image - content_blocks = [ - {"type": "text", "text": current_message}, - { - "type": "image", - "source": { - "type": "base64", - "media_type": "image/jpeg", - "data": b64_encoded_image, - }, - }, - ] - - messages.append({"role": "user", "content": content_blocks}) - - return messages diff --git a/blueprints/single_provider/gemini/README.md b/blueprints/single_provider/gemini/README.md deleted file mode 100644 index 1b1b3be..0000000 --- a/blueprints/single_provider/gemini/README.md +++ /dev/null @@ -1,359 +0,0 @@ -# Building a Google Gemini Agent in Pixeltable: A Step-by-Step Guide - -This tutorial walks you through building `agent.py`, a persistent conversational agent with memory using Pixeltable's automated data orchestration and storage—powered by Google Gemini's models. We'll build the agent step by step, explaining each component as we go. By the end, you'll have a complete implementation that can handle both chat and tool-calling capabilities. - -The final code is available in `agent.py`, but we recommend following the tutorial to understand how each piece fits together. - -## Prerequisites -- Install required packages: - ```bash - pip install pixeltable google-genai - ``` -- Set up your Google AI API key: - ```bash - export GOOGLE_API_KEY='your-api-key' - ``` - -## Step 1: Understanding the Building Blocks - -The agent is built on three core components: - -1. **Memory Management**: Using Pixeltable tables to store conversation history and enable persistent memory across sessions -2. **Chat Pipeline**: A series of computed columns that process user input and generate responses -3. **Tool Execution**: Optional support for executing functions/tools during conversations - -Key classes and functions: - -- `Agent`: Main class that orchestrates all components -- `create_messages()`: UDF that formats conversation history for the LLM, with support for images -- Pixeltable tables: - - `memory`: Stores all conversation history - - `agent`: Manages chat interactions - - `tools`: Handles tool execution (if enabled) - -The code shows the basic setup with imports and the Agent class initialization: - -```python -import base64 -import io -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf - -try: - from pixeltable.functions.gemini import generate_content, invoke_tools -except ImportError: - raise ImportError("google.genai not found; run `pip install google-genai`") - -@pxt.udf -def create_messages( - system_prompt: str, memory_context: list[dict], current_message: str, - image: Optional[PIL.Image.Image] = None -) -> list[dict]: - messages = [{"role": "system", "content": system_prompt}] - messages.extend(memory_context.copy()) - - # For text-only messages - if not image: - messages.append({"role": "user", "content": current_message}) - return messages - - # For messages with images - # Encode Image - bytes_arr = io.BytesIO() - image.save(bytes_arr, format="jpeg") - b64_bytes = base64.b64encode(bytes_arr.getvalue()) - b64_encoded_image = b64_bytes.decode("utf-8") - - # Create content blocks with text and image - content_blocks = [ - {"type": "text", "text": current_message}, - { - "type": "image_url", - "image_url": {"url": f"data:image/jpeg;base64,{b64_encoded_image}"}, - }, - ] - - messages.append({"role": "user", "content": content_blocks}) - return messages - -class Agent: - def __init__( - self, - name: str, - system_prompt: str, - model: str = "gemini-2.0-flash", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} -``` - -## Step 2: Building the Memory and Chat Pipeline - -The memory system consists of two main parts: - -1. **Table Setup** (`_setup_tables()`): - - Creates the `memory` table to store all conversations with timestamps - - Creates the `agent` table to manage chat completions - - Optionally creates the `tools` table if tool support is enabled - - All tables use Pixeltable's automated data orchestration - -2. **Chat Pipeline** (`_setup_chat_pipeline()`): - - Uses computed columns to create a 4-step processing pipeline: - 1. Retrieves recent messages based on `n_latest_messages` limit - 2. Formats messages with system prompt for the LLM - 3. Gets completion from Google Gemini API - 4. Extracts the final response text - -The implementation shows how Pixeltable's computed columns automatically handle dependencies: - -```python -def _setup_tables(self): - """ - Initialize the required Pixeltable tables for the agent. - Creates three tables: - 1. memory: Stores conversation history - 2. agent: Manages chat completions - 3. tools: (Optional) Handles tool execution - """ - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for Gemini - "image": PIL.Image.Image, # Optional image for multimodal input - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for Gemini - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - # Set up tools pipeline - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - -def _setup_chat_pipeline(self): - """ - Configure the chat completion pipeline using Pixeltable computed columns - 1. Get recent messages from memory - 2. Create messages for Gemini - 3. Get Gemini's response - 4. Extract the response text - """ - # Get recent messages from memory, respecting n_latest_messages limit if set - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Add computed columns to agent table - # Get recent messages from memory - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), - if_exists="ignore", - ) - # Create messages for Gemini - self.agent.add_computed_column( - prompt=create_messages( - self.agent.system_prompt, - self.agent.memory_context, - self.agent.user_message, - ), - if_exists="ignore", - ) - # Get api response - self.agent.add_computed_column( - response=generate_content( - messages=self.agent.prompt, model=self.model, **self.chat_kwargs - ), - if_exists="ignore", - ) - # Extract the response text - self.agent.add_computed_column( - agent_response=self.agent.response['candidates'][0]['content']['parts'][0]['text'], - if_exists="ignore", - ) -``` - -## Step 3: Adding Tool-Calling Support - -Tool calling is implemented as a three-stage process: - -1. **Initial Response**: - - Sends the user's request to the LLM with available tools - - LLM decides if and which tools to call - -2. **Tool Execution**: - - `invoke_tools()` executes the requested tools - - Results are stored in the tools table - -3. **Final Response**: - - Tool results are sent back to the LLM - - LLM generates a final response incorporating tool outputs - -The pipeline is implemented using computed columns that automatically handle the tool calling handshake: - -```python -def _setup_tools_pipeline(self): - """ - Configure the tool call handshake pipeline using Pixeltable computed columns - 1. Get initial response from Gemini with tool calls - 2. Execute the tools - 3. Pass the tool results back to Gemini for final response - """ - # Get initial response from Gemini with tool calls - self.tools_table.add_computed_column( - initial_response=generate_content( - model=self.model, - messages=[{"role": "user", "content": self.tools_table.tool_prompt}], - tools=self.tools, - **self.tool_kwargs - ), - if_exists="ignore", - ) - - # Execute the tools - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - - # Pass the tool results back to Gemini for final response - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - - # Get final API response from Gemini with tool results - self.tools_table.add_computed_column( - final_response=generate_content( - model=self.model, - messages=[{"role": "user", "content": self.tools_table.tool_response_prompt}], - **self.chat_kwargs - ), - if_exists="ignore", - ) - - # Parse Gemini's response - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response['candidates'][0]['content']['parts'][0]['text'], - if_exists="ignore", - ) -``` - -## Using the Agent - -To use the agent in your code: - -1. **Initialize the Agent**: -```python -agent = Agent( - name="my_agent", - system_prompt="You are a helpful assistant", - model="gemini-2.0-flash", # or other Gemini models - n_latest_messages=10 # None for unlimited history -) -``` - -2. **Chat with the Agent**: -```python -# Simple text chat -response = agent.chat("Hello, how are you?") - -# Chat with an image -from PIL import Image -img = Image.open("path/to/image.jpg") -response = agent.chat("What's in this image?", image=img) - -# Chat with tool execution -response = agent.tool_call("What's the weather in New York?") -``` - -3. **Access History**: -```python -# View recent conversations -memory = pxt.get_table("my_agent.memory") -print(memory.collect()) -``` - -For more details and advanced usage, refer to the complete implementation in `agent.py`. - -## Testing the Agent - -You can test this implementation using the provided `test.py` script, which demonstrates both chat and tool-calling capabilities of the agent. - -## Key Features - -The implementation includes several key features: - -1. **Shared Base Implementation** - - Inherits common Pydantic fields from BaseAgent - - Consistent interface across LLM providers - - Abstract methods enforce proper implementation - - Reduced code duplication - -2. **Flexible Memory Management** - - Stores all conversation history in Pixeltable - - Supports both limited and unlimited memory (n_latest_messages=None) - - Unique message IDs for reliable tracking - - Support for multimodal conversations with image handling - -3. **Robust Tool Execution** - - Clean handshake between LLM and tools - - Structured tool response handling - - Automatic tool result incorporation - -4. **Data Orchestration** - - Automated pipeline management - - Persistent storage - - Efficient query optimization - -For more details and advanced usage, refer to the complete implementation in `agent.py`. diff --git a/blueprints/single_provider/gemini/agent.py b/blueprints/single_provider/gemini/agent.py deleted file mode 100644 index f68256e..0000000 --- a/blueprints/single_provider/gemini/agent.py +++ /dev/null @@ -1,393 +0,0 @@ -""" -Tutorial: Building Your Own Google Gemini-powered Agent with Pixeltable - -This example demonstrates how to create a conversational AI agent using Google Gemini's LLM model -and Pixeltable for persistent memory, storage, orchestration, and tool execution. The agent can maintain conversation -history and execute tools while keeping track of all interactions in a structured database. -""" - -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf - -try: - from pixeltable.functions.gemini import generate_content, invoke_tools -except ImportError: - raise ImportError("google.genai not found; run `pip install google-genai`") - - -@pxt.udf -def create_content( - memory_context: list[dict], - current_message: str, -) -> str: - - # Build the conversation context as a text string without system prompt - context = "" - - # Add memory context - for msg in memory_context: - context += f"{msg['role'].title()}: {msg['content']}\n" - - # Add current message - context += f"User: {current_message}\n" - context += "Assistant: " - - return context - - -class Agent: - """ - A Google Gemini agent powered by their LLM models with persistent memory and tool execution capabilities. - - The agent maintains three key tables in Pixeltable: - 1. memory: Stores all conversation history with timestamps - 2. agent: Manages chat interactions and responses - 3. tools: (Optional) Handles tool execution and responses - - Key Features: - - Persistent conversation memory with optional message limit - - Tool execution support - - Structured data storage and orchestration using Pixeltable - """ - - def __init__( - self, - name: str, - system_prompt: str, - model: str = "gemini-2.0-flash", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - """ - Initialize the agent with the specified configuration. - - Args: - name: Unique name for the agent (used for table names) - system_prompt: System prompt that guides LLM's behavior - model: LLM model to use - n_latest_messages: Number of recent messages to include in context (None for unlimited) - tools: Optional tools configuration for function calling - reset: If True, deletes existing agent data - chat_kwargs: Additional kwargs for chat completion - tool_kwargs: Additional kwargs for tool execution - """ - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} - - # Set up or reset the agent's database - if reset: - pxt.drop_dir(self.directory, if_not_exists="ignore", force=True) - - # Create agent directory if it doesn't exist - pxt.create_dir(self.directory, if_exists="ignore") - - # Set up tables - self._setup_tables() - - # Get references to the created tables - self.memory = pxt.get_table(f"{self.directory}.memory") - self.agent = pxt.get_table(f"{self.directory}.agent") - self.tools_table = ( - pxt.get_table(f"{self.directory}.tools") if self.tools else None - ) - - def _setup_tables(self): - """ - Initialize the required Pixeltable tables for the agent. - Creates three tables: - 1. memory: Stores conversation history - 2. agent: Manages chat completions - 3. tools: (Optional) Handles tool execution - """ - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for LLM - "image": pxt.Image, # Optional image attachment - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for Gemini - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - # Set up tools pipeline - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - - def _setup_chat_pipeline(self): - """ - Configure the chat completion pipeline using Pixeltable computed columns - 1. Get recent messages from memory - 2. Create messages for Gemini - 3. Get Gemini's response - 4. Extract the response text - """ - - # Get recent messages from memory, respecting n_latest_messages limit if set - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Add computed columns to agent table - # Get recent messages from memory - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), - if_exists="ignore", - ) - # Create content for Gemini without system prompt - self.agent.add_computed_column( - prompt=create_content( - self.agent.memory_context, - self.agent.user_message, - ), - if_exists="ignore", - ) - - # Prepare config with system instruction - chat_config = self.chat_kwargs.copy() if self.chat_kwargs else {} - chat_config['system_instruction'] = self.system_prompt - - # Get api response - self.agent.add_computed_column( - response=generate_content( - contents=self.agent.prompt, model=self.model, config=chat_config - ), - if_exists="ignore", - ) - # Extract the response text - self.agent.add_computed_column( - agent_response=self.agent.response['candidates'][0]['content']['parts'][0]['text'], - if_exists="ignore", - ) - - def _setup_tools_pipeline(self): - """ - Configure the tool call handshake pipeline using Pixeltable computed columns - 1. Get initial response from Gemini with tool calls - 2. Execute the tools - 3. Pass the tool results back to Gemini for final response - """ - # Prepare config with system instruction for tools - tool_config = self.tool_kwargs.copy() if self.tool_kwargs else {} - tool_config['system_instruction'] = self.system_prompt - - # Get initial response from Gemini with tool calls - self.tools_table.add_computed_column( - initial_response=generate_content( - contents=self.tools_table.tool_prompt, - model=self.model, - tools=self.tools, - config=tool_config, - ), - if_exists="ignore", - ) - # Execute the tools - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - # Pass the tool results back to Gemini for final response - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - # Get final response from Gemini with tool results - self.tools_table.add_computed_column( - final_response=generate_content( - contents=self.tools_table.tool_response_prompt, - model=self.model, - config=tool_config, - ), - if_exists="ignore", - ) - # Extract the final response text - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response['candidates'][0]['content']['parts'][0]['text'], - if_exists="ignore", - ) - - def chat(self, message: str, image: Optional[PIL.Image.Image] = None) -> str: - """ - Send a message to the agent and get its response. - - This method: - 1. Stores the user message in memory - 2. Triggers the chat completion pipeline - 3. Stores the assistant's response in memory - 4. Returns the response - - Args: - message: The user's message - image: Optional image attachment - - Returns: - The agent's response - """ - now = datetime.now() - - # Generate unique IDs for the message pair - user_message_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": message, - "timestamp": now, - } - ] - ) - - # Store user message in agent table (which triggers the chat pipeline) - self.agent.insert( - [ - { - "message_id": user_message_id, - "user_message": message, - "timestamp": now, - "system_prompt": self.system_prompt, - "image": image, - } - ] - ) - - # Get LLM's response from agent table - result = ( - self.agent.select(self.agent.agent_response) - .where(self.agent.message_id == user_message_id) - .collect() - ) - response = result["agent_response"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": response, - "timestamp": now, - } - ] - ) - return response - - def tool_call(self, prompt: str) -> str: - """ - Execute a tool call with the given prompt. - - This method: - 1. Stores the user prompt in memory - 2. Triggers the tool call handshake pipeline - 3. Stores the tool's response in memory - 4. Returns the response - - Args: - prompt: The user's prompt - - Returns: - The tool's response - """ - if not self.tools: - return "No tools configured for this agent." - - now = datetime.now() - user_message_id = str(uuid4()) - tool_invoke_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": prompt, - "timestamp": now, - } - ] - ) - - # Store user prompt in tools table (which triggers the tool call handshake pipeline) - self.tools_table.insert( - [ - { - "tool_invoke_id": tool_invoke_id, - "tool_prompt": prompt, - "timestamp": now, - } - ] - ) - - # Get tool answer from tools table - result = ( - self.tools_table.select(self.tools_table.tool_answer) - .where(self.tools_table.tool_invoke_id == tool_invoke_id) - .collect() - ) - tool_answer = result["tool_answer"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": tool_answer, - "timestamp": now, - } - ] - ) - return tool_answer diff --git a/blueprints/single_provider/gemini/requirements.txt b/blueprints/single_provider/gemini/requirements.txt deleted file mode 100644 index 1a374ac..0000000 --- a/blueprints/single_provider/gemini/requirements.txt +++ /dev/null @@ -1,2 +0,0 @@ -pixeltable -google-genai \ No newline at end of file diff --git a/blueprints/single_provider/gemini/test.py b/blueprints/single_provider/gemini/test.py deleted file mode 100644 index 69ef0a8..0000000 --- a/blueprints/single_provider/gemini/test.py +++ /dev/null @@ -1,58 +0,0 @@ -from dotenv import load_dotenv -import pixeltable as pxt -from agent import Agent - -# Load environment variables from .env file -load_dotenv() - -@pxt.udf -def weather(city: str) -> str: - """ - Returns the weather in a given city. - """ - return f"The weather in {city} is sunny." - - -# Create an agent -agent = Agent( - name="gemini_agent", - system_prompt="You're my assistant.", - tools=pxt.tools(weather), - reset=True, -) - -# Persistant chat and memory -print("=== FIRST CHAT ===") -response1 = agent.chat("Hi, how are you?") -print(f"Response: {response1}") - -# Check what's in memory after first chat -print("\n=== MEMORY CONTENTS AFTER FIRST CHAT ===") -memory_contents = agent.memory.select(agent.memory.role, agent.memory.content, agent.memory.timestamp).order_by(agent.memory.timestamp).collect() -for row in memory_contents: - print(f"Role: {row['role']}, Content: {row['content']}, Timestamp: {row['timestamp']}") - -print("\n=== SECOND CHAT ===") -response2 = agent.chat("What was my last question?") -print(f"Response: {response2}") - -# Check memory context that should be retrieved -print("\n=== MEMORY CONTENTS AFTER SECOND CHAT ===") -memory_contents = agent.memory.select(agent.memory.role, agent.memory.content, agent.memory.timestamp).order_by(agent.memory.timestamp).collect() -for row in memory_contents: - print(f"Role: {row['role']}, Content: {row['content']}, Timestamp: {row['timestamp']}") - -# Check what was actually passed to the agent table -print("\n=== AGENT TABLE CONTENTS ===") -agent_contents = agent.agent.select(agent.agent.user_message, agent.agent.memory_context, agent.agent.prompt, agent.agent.agent_response).collect() -for i, row in enumerate(agent_contents): - print(f"Request {i+1}:") - print(f" User Message: {row['user_message']}") - print(f" Memory Context: {row['memory_context']}") - print(f" Prompt: {row['prompt'][:200]}...") # First 200 chars - print(f" Response: {row['agent_response']}") - print() - -print("\n=== TOOL CALL ===") -tool_response = agent.tool_call("Get weather in San Francisco") -print(f"Tool Response: {tool_response}") diff --git a/blueprints/single_provider/openai/README.md b/blueprints/single_provider/openai/README.md deleted file mode 100644 index 28981f1..0000000 --- a/blueprints/single_provider/openai/README.md +++ /dev/null @@ -1,359 +0,0 @@ -# Building an OpenAI Agent in Pixeltable: A Step-by-Step Guide - -This tutorial walks you through building `agent.py`, a persistent conversational agent with memory using Pixeltable's automated data orchestration and storage—powered by OpenAI's GPT models. We'll build the agent step by step, explaining each component as we go. By the end, you'll have a complete implementation that can handle both chat and tool-calling capabilities. - -The final code is available in `agent.py`, but we recommend following the tutorial to understand how each piece fits together. - -## Prerequisites -- Install required packages: - ```bash - pip install pixeltable openai - ``` -- Set up your OpenAI API key: - ```bash - export OPENAI_API_KEY='your-api-key' - ``` - -## Step 1: Understanding the Building Blocks - -The agent is built on three core components: - -1. **Memory Management**: Using Pixeltable tables to store conversation history and enable persistent memory across sessions -2. **Chat Pipeline**: A series of computed columns that process user input and generate responses -3. **Tool Execution**: Optional support for executing functions/tools during conversations - -Key classes and functions: - -- `Agent`: Main class that orchestrates all components -- `create_messages()`: UDF that formats conversation history for the LLM, with support for images -- Pixeltable tables: - - `memory`: Stores all conversation history - - `agent`: Manages chat interactions - - `tools`: Handles tool execution (if enabled) - -The code shows the basic setup with imports and the Agent class initialization: - -```python -import base64 -import io -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf - -try: - from pixeltable.functions.openai import chat_completions, invoke_tools -except ImportError: - raise ImportError("openai not found; run `pip install openai`") - -@pxt.udf -def create_messages( - system_prompt: str, memory_context: list[dict], current_message: str, - image: Optional[PIL.Image.Image] = None -) -> list[dict]: - messages = [{"role": "system", "content": system_prompt}] - messages.extend(memory_context.copy()) - - # For text-only messages - if not image: - messages.append({"role": "user", "content": current_message}) - return messages - - # For messages with images - # Encode Image - bytes_arr = io.BytesIO() - image.save(bytes_arr, format="jpeg") - b64_bytes = base64.b64encode(bytes_arr.getvalue()) - b64_encoded_image = b64_bytes.decode("utf-8") - - # Create content blocks with text and image - content_blocks = [ - {"type": "text", "text": current_message}, - { - "type": "image_url", - "image_url": {"url": f"data:image/jpeg;base64,{b64_encoded_image}"}, - }, - ] - - messages.append({"role": "user", "content": content_blocks}) - return messages - -class Agent: - def __init__( - self, - name: str, - system_prompt: str, - model: str = "gpt-4o-mini", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} -``` - -## Step 2: Building the Memory and Chat Pipeline - -The memory system consists of two main parts: - -1. **Table Setup** (`_setup_tables()`): - - Creates the `memory` table to store all conversations with timestamps - - Creates the `agent` table to manage chat completions - - Optionally creates the `tools` table if tool support is enabled - - All tables use Pixeltable's automated data orchestration - -2. **Chat Pipeline** (`_setup_chat_pipeline()`): - - Uses computed columns to create a 4-step processing pipeline: - 1. Retrieves recent messages based on `n_latest_messages` limit - 2. Formats messages with system prompt for the LLM - 3. Gets completion from OpenAI API - 4. Extracts the final response text - -The implementation shows how Pixeltable's computed columns automatically handle dependencies: - -```python -def _setup_tables(self): - """ - Initialize the required Pixeltable tables for the agent. - Creates three tables: - 1. memory: Stores conversation history - 2. agent: Manages chat completions - 3. tools: (Optional) Handles tool execution - """ - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for GPT - "image": PIL.Image.Image, # Optional image for multimodal input - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for Claude - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - # Set up tools pipeline - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - -def _setup_chat_pipeline(self): - """ - Configure the chat completion pipeline using Pixeltable computed columns - 1. Get recent messages from memory - 2. Create messages for GPT - 3. Get GPT's response - 4. Extract the response text - """ - # Get recent messages from memory, respecting n_latest_messages limit if set - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Add computed columns to agent table - # Get recent messages from memory - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), - if_exists="ignore", - ) - # Create messages for Openai - self.agent.add_computed_column( - prompt=create_messages( - self.agent.system_prompt, - self.agent.memory_context, - self.agent.user_message, - ), - if_exists="ignore", - ) - # Get api response - self.agent.add_computed_column( - response=chat_completions( - messages=self.agent.prompt, model=self.model, **self.chat_kwargs - ), - if_exists="ignore", - ) - # Extract the response text - self.agent.add_computed_column( - agent_response=self.agent.response.choices[0].message.content, - if_exists="ignore", - ) -``` - -## Step 3: Adding Tool-Calling Support - -Tool calling is implemented as a three-stage process: - -1. **Initial Response**: - - Sends the user's request to the LLM with available tools - - LLM decides if and which tools to call - -2. **Tool Execution**: - - `invoke_tools()` executes the requested tools - - Results are stored in the tools table - -3. **Final Response**: - - Tool results are sent back to the LLM - - LLM generates a final response incorporating tool outputs - -The pipeline is implemented using computed columns that automatically handle the tool calling handshake: - -```python -def _setup_tools_pipeline(self): - """ - Configure the tool call handshake pipeline using Pixeltable computed columns - 1. Get initial response from GPT with tool calls - 2. Execute the tools - 3. Pass the tool results back to GPT for final response - """ - # Get initial response from GPT with tool calls - self.tools_table.add_computed_column( - initial_response=chat_completions( - model=self.model, - messages=[{"role": "user", "content": self.tools_table.tool_prompt}], - tools=self.tools, - **self.tool_kwargs - ), - if_exists="ignore", - ) - - # Execute the tools - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - - # Pass the tool results back to GPT for final response - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - - # Get final API response from OpenAI with tool results - self.tools_table.add_computed_column( - final_response=chat_completions( - model=self.model, - messages=[{"role": "user", "content": self.tools_table.tool_response_prompt}], - **self.chat_kwargs - ), - if_exists="ignore", - ) - - # Parse GPT's response - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response.choices[0].message.content, - if_exists="ignore", - ) -``` - -## Using the Agent - -To use the agent in your code: - -1. **Initialize the Agent**: -```python -agent = Agent( - name="my_agent", - system_prompt="You are a helpful assistant", - model="gpt-4o-mini", # or other OpenAI models - n_latest_messages=10 # None for unlimited history -) -``` - -2. **Chat with the Agent**: -```python -# Simple text chat -response = agent.chat("Hello, how are you?") - -# Chat with an image -from PIL import Image -img = Image.open("path/to/image.jpg") -response = agent.chat("What's in this image?", image=img) - -# Chat with tool execution -response = agent.tool_call("What's the weather in New York?") -``` - -3. **Access History**: -```python -# View recent conversations -memory = pxt.get_table("my_agent.memory") -print(memory.collect()) -``` - -For more details and advanced usage, refer to the complete implementation in `agent.py`. - -## Testing the Agent - -You can test this implementation using the provided `test.py` script, which demonstrates both chat and tool-calling capabilities of the agent. - -## Key Features - -The implementation includes several key features: - -1. **Shared Base Implementation** - - Inherits common Pydantic fields from BaseAgent - - Consistent interface across LLM providers - - Abstract methods enforce proper implementation - - Reduced code duplication - -2. **Flexible Memory Management** - - Stores all conversation history in Pixeltable - - Supports both limited and unlimited memory (n_latest_messages=None) - - Unique message IDs for reliable tracking - - Support for multimodal conversations with image handling - -3. **Robust Tool Execution** - - Clean handshake between LLM and tools - - Structured tool response handling - - Automatic tool result incorporation - -4. **Data Orchestration** - - Automated pipeline management - - Persistent storage - - Efficient query optimization - -For more details and advanced usage, refer to the complete implementation in `agent.py`. \ No newline at end of file diff --git a/blueprints/single_provider/openai/agent.py b/blueprints/single_provider/openai/agent.py deleted file mode 100644 index 300dfb3..0000000 --- a/blueprints/single_provider/openai/agent.py +++ /dev/null @@ -1,403 +0,0 @@ -""" -Tutorial: Building Your Own OpenAI-powered Agent with Pixeltable - -This example demonstrates how to create a conversational AI agent using OpenAI's LLM model -and Pixeltable for persistent memory, storage, orchestration, and tool execution. The agent can maintain conversation -history and execute tools while keeping track of all interactions in a structured database. -""" - -import base64 -import io -from datetime import datetime -from typing import Optional -from uuid import uuid4 - -import PIL -import pixeltable as pxt -import pixeltable.functions as pxtf - -try: - from pixeltable.functions.openai import chat_completions, invoke_tools -except ImportError: - raise ImportError("openai not found; run `pip install openai`") - - -@pxt.udf -def create_messages( - system_prompt: str, - memory_context: list[dict], - current_message: str, - image: Optional[PIL.Image.Image] = None, -) -> list[dict]: - - messages = [{"role": "system", "content": system_prompt}] - messages.extend(memory_context.copy()) - - if not image: - messages.append({"role": "user", "content": current_message}) - return messages - - # Encode Image - bytes_arr = io.BytesIO() - image.save(bytes_arr, format="jpeg") - b64_bytes = base64.b64encode(bytes_arr.getvalue()) - b64_encoded_image = b64_bytes.decode("utf-8") - - # Create content blocks with text and image - content_blocks = [ - {"type": "text", "text": current_message}, - { - "type": "image_url", - "image_url": {"url": f"data:image/jpeg;base64,{b64_encoded_image}"}, - }, - ] - - messages.append({"role": "user", "content": content_blocks}) - return messages - - -class Agent: - """ - An OpenAI agent powered by their LLM models with persistent memory and tool execution capabilities. - - The agent maintains three key tables in Pixeltable: - 1. memory: Stores all conversation history with timestamps - 2. agent: Manages chat interactions and responses - 3. tools: (Optional) Handles tool execution and responses - - Key Features: - - Persistent conversation memory with optional message limit - - Tool execution support - - Structured data storage and orchestration using Pixeltable - """ - - def __init__( - self, - name: str, - system_prompt: str, - model: str = "gpt-4o-mini", - n_latest_messages: Optional[int] = 10, - tools: Optional[pxt.tools] = None, - reset: bool = False, - chat_kwargs: Optional[dict] = None, - tool_kwargs: Optional[dict] = None, - ): - """ - Initialize the agent with the specified configuration. - - Args: - name: Unique name for the agent (used for table names) - system_prompt: System prompt that guides LLM's behavior - model: LLM model to use - n_latest_messages: Number of recent messages to include in context (None for unlimited) - tools: Optional tools configuration for function calling - reset: If True, deletes existing agent data - chat_kwargs: Additional kwargs for chat completion - tool_kwargs: Additional kwargs for tool execution - """ - self.directory = name - self.system_prompt = system_prompt - self.model = model - self.n_latest_messages = n_latest_messages - self.tools = tools - self.chat_kwargs = chat_kwargs or {} - self.tool_kwargs = tool_kwargs or {} - - # Set up or reset the agent's database - if reset: - pxt.drop_dir(self.directory, if_not_exists="ignore", force=True) - - # Create agent directory if it doesn't exist - pxt.create_dir(self.directory, if_exists="ignore") - - # Set up tables - self._setup_tables() - - # Get references to the created tables - self.memory = pxt.get_table(f"{self.directory}.memory") - self.agent = pxt.get_table(f"{self.directory}.agent") - self.tools_table = ( - pxt.get_table(f"{self.directory}.tools") if self.tools else None - ) - - def _setup_tables(self): - """ - Initialize the required Pixeltable tables for the agent. - Creates three tables: - 1. memory: Stores conversation history - 2. agent: Manages chat completions - 3. tools: (Optional) Handles tool execution - """ - # Create memory table for conversation history - self.memory = pxt.create_table( - f"{self.directory}.memory", - { - "message_id": pxt.String, # Unique ID for each message - "role": pxt.String, # 'user' or 'assistant' - "content": pxt.String, # Message content - "timestamp": pxt.Timestamp, # When the message was received - }, - if_exists="ignore", - ) - - # Create agent table for managing chat interactions - self.agent = pxt.create_table( - f"{self.directory}.agent", - { - "message_id": pxt.String, # Unique ID for each message - "user_message": pxt.String, # User's message content - "timestamp": pxt.Timestamp, # When the message was received - "system_prompt": pxt.String, # System prompt for LLM - "image": pxt.Image, # Optional image attachment - }, - if_exists="ignore", - ) - - # Create tools table if tools are configured - if self.tools: - self.tools_table = pxt.create_table( - f"{self.directory}.tools", - { - "tool_invoke_id": pxt.String, # Unique ID for each tool invocation - "tool_prompt": pxt.String, # Tool prompt for Claude - "timestamp": pxt.Timestamp, # When the tool was invoked - }, - if_exists="ignore", - ) - # Set up tools pipeline - self._setup_tools_pipeline() - - # Set up chat pipeline - self._setup_chat_pipeline() - - def _setup_chat_pipeline(self): - """ - Configure the chat completion pipeline using Pixeltable computed columns - 1. Get recent messages from memory - 2. Create messages for GPT - 3. Get GPT's response - 4. Extract the response text - """ - - # Get recent messages from memory, respecting n_latest_messages limit if set - @pxt.query - def get_recent_memory(current_timestamp: pxt.Timestamp) -> list[dict]: - query = ( - self.memory.where(self.memory.timestamp < current_timestamp) - .order_by(self.memory.timestamp, asc=False) - .select(role=self.memory.role, content=self.memory.content) - ) - if self.n_latest_messages is not None: - query = query.limit(self.n_latest_messages) - return query - - # Add computed columns to agent table - # Get recent messages from memory - self.agent.add_computed_column( - memory_context=get_recent_memory(self.agent.timestamp), - if_exists="ignore", - ) - # Create messages for Openai - self.agent.add_computed_column( - prompt=create_messages( - self.agent.system_prompt, - self.agent.memory_context, - self.agent.user_message, - ), - if_exists="ignore", - ) - # Get api response - self.agent.add_computed_column( - response=chat_completions( - messages=self.agent.prompt, model=self.model, **self.chat_kwargs - ), - if_exists="ignore", - ) - # Extract the response text - self.agent.add_computed_column( - agent_response=self.agent.response.choices[0].message.content, - if_exists="ignore", - ) - - def _setup_tools_pipeline(self): - """ - Configure the tool call handshake pipeline using Pixeltable computed columns - 1. Get initial response from GPT with tool calls - 2. Execute the tools - 3. Pass the tool results back to GPT for final response - """ - # Get initial response from GPT with tool calls - self.tools_table.add_computed_column( - initial_response=chat_completions( - model=self.model, - messages=[{"role": "user", "content": self.tools_table.tool_prompt}], - tools=self.tools, - **self.tool_kwargs, - ), - if_exists="ignore", - ) - # Execute the tools - self.tools_table.add_computed_column( - tool_output=invoke_tools(self.tools, self.tools_table.initial_response), - if_exists="ignore", - ) - # Pass the tool results back to GPT for final response - self.tools_table.add_computed_column( - tool_response_prompt=pxtf.string.format( - "{0}: {1}", self.tools_table.tool_prompt, self.tools_table.tool_output - ), - if_exists="ignore", - ) - # Get final response from GPT with tool results - self.tools_table.add_computed_column( - final_response=chat_completions( - model=self.model, - messages=[ - {"role": "user", "content": self.tools_table.tool_response_prompt}, - ], - **self.tool_kwargs, - ), - if_exists="ignore", - ) - # Extract the final response text - self.tools_table.add_computed_column( - tool_answer=self.tools_table.final_response.choices[0].message.content, - if_exists="ignore", - ) - - def chat(self, message: str, image: Optional[PIL.Image.Image] = None) -> str: - """ - Send a message to the agent and get its response. - - This method: - 1. Stores the user message in memory - 2. Triggers the chat completion pipeline - 3. Stores the assistant's response in memory - 4. Returns the response - - Args: - message: The user's message - image: Optional image attachment - - Returns: - The agent's response - """ - now = datetime.now() - - # Generate unique IDs for the message pair - user_message_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": message, - "timestamp": now, - } - ] - ) - - # Store user message in agent table (which triggers the chat pipeline) - self.agent.insert( - [ - { - "message_id": user_message_id, - "user_message": message, - "timestamp": now, - "system_prompt": self.system_prompt, - "image": image, - } - ] - ) - - # Get LLM's response from agent table - result = ( - self.agent.select(self.agent.agent_response) - .where(self.agent.message_id == user_message_id) - .collect() - ) - response = result["agent_response"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": response, - "timestamp": now, - } - ] - ) - return response - - def tool_call(self, prompt: str) -> str: - """ - Execute a tool call with the given prompt. - - This method: - 1. Stores the user prompt in memory - 2. Triggers the tool call handshake pipeline - 3. Stores the tool's response in memory - 4. Returns the response - - Args: - prompt: The user's prompt - - Returns: - The tool's response - """ - if not self.tools: - return "No tools configured for this agent." - - now = datetime.now() - user_message_id = str(uuid4()) - tool_invoke_id = str(uuid4()) - assistant_message_id = str(uuid4()) - - # Store user message in memory - self.memory.insert( - [ - { - "message_id": user_message_id, - "role": "user", - "content": prompt, - "timestamp": now, - } - ] - ) - - # Store user prompt in tools table (which triggers the tool call handshake pipeline) - self.tools_table.insert( - [ - { - "tool_invoke_id": tool_invoke_id, - "tool_prompt": prompt, - "timestamp": now, - } - ] - ) - - # Get tool answer from tools table - result = ( - self.tools_table.select(self.tools_table.tool_answer) - .where(self.tools_table.tool_invoke_id == tool_invoke_id) - .collect() - ) - tool_answer = result["tool_answer"][0] - - # Store LLM's response in memory - self.memory.insert( - [ - { - "message_id": assistant_message_id, - "role": "assistant", - "content": tool_answer, - "timestamp": now, - } - ] - ) - return tool_answer diff --git a/blueprints/single_provider/openai/test.py b/blueprints/single_provider/openai/test.py deleted file mode 100644 index b52c129..0000000 --- a/blueprints/single_provider/openai/test.py +++ /dev/null @@ -1,26 +0,0 @@ -import pixeltable as pxt -from agent import Agent - - -@pxt.udf -def weather(city: str) -> str: - """ - Returns the weather in a given city. - """ - return f"The weather in {city} is sunny." - - -# Create an agent -agent = Agent( - name="openai_agent", - system_prompt="You’re my assistant.", - tools=pxt.tools(weather), - reset=True, -) - -# Persistant chat and memory -print(agent.chat("Hi, how are you?")) -print(agent.chat("What was my last question?")) - -# Tool call -print(agent.tool_call("Get weather in San Francisco")) diff --git a/pyproject.toml b/pyproject.toml index c84e946..d16a3f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api" [project] name = "pixelagent" -version = "0.1.6" +version = "0.2.0" description = "A Pixeltable-native agent framework for OpenAI, Anthropic, AWS Bedrock and Google Gemini" license = "Apache-2.0" readme = "README.md" diff --git a/uv.lock b/uv.lock index 8cf7a54..68c2eb3 100644 --- a/uv.lock +++ b/uv.lock @@ -1914,7 +1914,7 @@ wheels = [ [[package]] name = "pixelagent" -version = "0.1.6" +version = "0.2.0" source = { editable = "." } dependencies = [ { name = "pixeltable" },