{ "cells": [ { "cell_type": "raw", "metadata": { "vscode": { "languageId": "raw" } }, "source": [ "---\n", "title: \"Agentic Patterns\"\n", "icon: \"notebook\"\n", "description: \"[Open in Kaggle](https://kaggle.com/kernels/welcome?src=https://github.com/pixeltable/pixeltable/blob/release/docs/release/howto/cookbooks/agents/agentic-patterns.ipynb) | [Open in Colab](https://colab.research.google.com/github/pixeltable/pixeltable/blob/release/docs/release/howto/cookbooks/agents/agentic-patterns.ipynb) | [View on GitHub](https://github.com/pixeltable/pixeltable/blob/release/docs/release/howto/cookbooks/agents/agentic-patterns.ipynb)\"\n", "---" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Two popular taxonomies describe the building blocks of agentic AI systems:\n", "\n", "- **Cognitive / reasoning-oriented** (Taxonomy 1): Reflection, Tool Use, ReAct, Planning, Multi-Agent — asks *\"how does the agent think?\"*\n", "- **Architectural / system-design-oriented** (Taxonomy 2): Prompt Chaining, Routing, Parallelization, Tool Use, Evaluator-Optimizer, Orchestrator-Worker — asks *\"how do you wire LLM calls together?\"*\n", "\n", "(See [OpenAI's Practical Guide to Building Agents](https://cdn.openai.com/business-guides-and-resources/a-practical-guide-to-building-agents.pdf), [Anthropic's multi-agent research system](https://www.anthropic.com/engineering/multi-agent-research-system), and [Pydantic AI's multi-agent delegation](https://ai.pydantic.dev/multi-agent-applications/#agent-delegation).)\n", "\n", "Mapping them against each other reveals:\n", "\n", "| Taxonomy 1 (cognitive) | Taxonomy 2 (architectural) | Relationship |\n", "|---|---|---|\n", "| Tool Use | Tool Use | Direct overlap |\n", "| Reflection | Evaluator-Optimizer | Close cousins |\n", "| Multi-Agent | Orchestrator-Worker | Close cousins |\n", "| ReAct | *(cross-cutting)* | Reasoning strategy applicable within any pattern |\n", "| Planning | *(cross-cutting)* | Reasoning strategy applicable within any pattern |\n", "| — | Prompt Chaining | Unique architectural wiring pattern |\n", "| — | Routing | Unique architectural wiring pattern |\n", "| — | Parallelization | Unique architectural wiring pattern |\n", "\n", "The cleanest framing: **six architectural patterns** that describe how you structure LLM calls, plus **two cross-cutting reasoning strategies** (ReAct and Planning) that can be layered inside any of them.\n", "\n", "This cookbook implements all eight in Pixeltable, where your agent *is* a table:\n", "\n", "| Concept | Imperative frameworks | Pixeltable |\n", "|---|---|---|\n", "| Pipeline step | Function call in a loop | Computed column |\n", "| Parallel execution | `asyncio.gather` | Independent computed columns (automatic) |\n", "| Persistence / observability | Separate logging layer | Built-in — every intermediate result is stored and queryable |\n", "| Caching | Manual memoization | Automatic — same input is never recomputed |\n", "| Reusable sub-agent | Agent class with `.run()` | `pxt.udf(table, return_value=...)` |" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Setup" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install -qU pixeltable openai" ] }, { "cell_type": "code", "execution_count": 143, "metadata": {}, "outputs": [], "source": [ "import getpass\n", "import os\n", "\n", "if 'OPENAI_API_KEY' not in os.environ:\n", " os.environ['OPENAI_API_KEY'] = getpass.getpass('OpenAI API Key: ')" ] }, { "cell_type": "code", "execution_count": 144, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created directory 'agentic_patterns'.\n" ] }, { "data": { "text/plain": [ "" ] }, "execution_count": 144, "metadata": {}, "output_type": "execute_result" } ], "source": [ "import pixeltable as pxt\n", "from pixeltable.functions import openai\n", "\n", "pxt.drop_dir('agentic_patterns', force=True)\n", "pxt.create_dir('agentic_patterns')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Pattern 1: Prompt Chaining\n", "\n", "Break a complex task into sequential steps, where each step's output feeds the next.\n", "\n", "**Imperative approach:** a chain of function calls or an explicit pipeline object.\n", "**Pixeltable approach:** each step is a computed column. The engine resolves dependencies automatically.\n", "\n", "```\n", "input → step 1 (outline) → step 2 (draft) → step 3 (polish) → output\n", "```" ] }, { "cell_type": "code", "execution_count": 145, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'chain'.\n" ] } ], "source": [ "# Create a table with a single input column\n", "chain = pxt.create_table('agentic_patterns/chain', {'topic': pxt.String})" ] }, { "cell_type": "code", "execution_count": 146, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 146, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 1: generate an outline\n", "chain.add_computed_column(\n", " outline_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Create a 3-point outline for a short article about: '\n", " + chain.topic,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "chain.add_computed_column(\n", " outline=chain.outline_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 147, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 147, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 2: write a draft from the outline\n", "chain.add_computed_column(\n", " draft_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Write a short article (2-3 paragraphs) based on this outline:\\n\\n'\n", " + chain.outline,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "chain.add_computed_column(\n", " draft=chain.draft_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 148, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 148, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 3: polish the draft\n", "chain.add_computed_column(\n", " polish_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Edit this article for clarity and conciseness. '\n", " 'Return only the improved text:\\n\\n' + chain.draft,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "chain.add_computed_column(\n", " final_article=chain.polish_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 149, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inserted 1 row with 0 errors in 14.58 s (0.07 rows/s)\n" ] }, { "data": { "text/html": [ "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
topicoutlinedraftfinal_article
the benefits of declarative AI pipelines**Outline for Article: Benefits of Declarative AI Pipelines**\n", "\n", "1. **Enhanced Clarity and Maintainability**\n", "   - Definition of declarative AI pipelines and how they differ from imperative approaches.\n", "   - The role of high-level abstraction in providing clear, succinct representations of AI workflows.\n", "   - Benefits of ease of understanding, enabling collaboration among teams, and simplifying maintenance and updates.\n", "\n", "2. **Increased Efficiency and Productivity**\n", "   - Automated management of res ...... ture.\n", "   - Examples of how declarative pipelines streamline repetitive tasks and optimize resource allocation.\n", "\n", "3. **Scalability and Flexibility**\n", "   - How declarative AI pipelines facilitate scaling to handle large datasets and complex models effectively.\n", "   - The ability to easily adapt and modify pipelines in response to changing project requirements or data inputs.\n", "   - Use cases highlighting successful implementations in dynamic environments that require quick iterations and deployment.**The Benefits of Declarative AI Pipelines**\n", "\n", "Declarative AI pipelines represent a significant shift from traditional imperative approaches in machine learning, emphasizing a high-level abstraction that simplifies the creation and management of AI workflows. Unlike imperative programming, which focuses on detailing the step-by-step procedures, declarative pipelines allow users to define what the desired outcome is without getting bogged down in how to achieve it. This clarity enhances mainta ...... mlessly, allowing organizations to handle vast amounts of data without sacrificing performance. Furthermore, the ability to swiftly adapt and modify these pipelines in response to evolving project needs or data inputs makes them ideal for dynamic environments. Successful use cases across various industries showcase how declarative pipelines have empowered organizations to iterate quickly and deploy solutions with confidence, further solidifying their place at the forefront of AI development.**The Benefits of Declarative AI Pipelines**\n", "\n", "Declarative AI pipelines mark a significant shift from traditional imperative approaches in machine learning, offering a high-level abstraction that simplifies the creation and management of AI workflows. Unlike imperative programming, which details step-by-step procedures, declarative pipelines let users specify the desired outcomes without worrying about the implementation. This clarity enhances maintainability and fosters collaboration, as all ......  and complex models, allowing organizations to process vast amounts of data without compromising performance. Additionally, the ability to quickly adapt and modify these pipelines in response to changing project needs or data inputs makes them well-suited for dynamic environments. Successful case studies across various industries demonstrate how declarative pipelines empower organizations to iterate rapidly and deploy solutions confidently, solidifying their role in advancing AI development.
" ], "text/plain": [ " topic \\\n", "0 the benefits of declarative AI pipelines \n", "\n", " outline \\\n", "0 **Outline for Article: Benefits of Declarative... \n", "\n", " draft \\\n", "0 **The Benefits of Declarative AI Pipelines**\\n... \n", "\n", " final_article \n", "0 **The Benefits of Declarative AI Pipelines**\\n... " ] }, "execution_count": 149, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Insert a topic — all three steps execute automatically\n", "chain.insert([{'topic': 'the benefits of declarative AI pipelines'}])\n", "\n", "chain.select(\n", " chain.topic, chain.outline, chain.draft, chain.final_article\n", ").collect()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Every intermediate result (`outline`, `draft`, `final_article`) is persisted in the table. Inserting another topic reuses the same pipeline — no code changes needed. If the same topic is inserted again, cached results are returned instantly." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Pattern 2: Routing\n", "\n", "Classify an input and route it to a specialized handler. This is the agent equivalent of a switch/case statement.\n", "\n", "**Imperative approach:** a triage agent that performs handoffs to specialized agents.\n", "**Pixeltable approach:** one computed column classifies; a UDF selects the prompt; a second LLM call generates the response.\n", "\n", "```\n", "input → classify intent → select specialized prompt → generate response\n", "```" ] }, { "cell_type": "code", "execution_count": 150, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'router'.\n" ] } ], "source": [ "router = pxt.create_table(\n", " 'agentic_patterns/router', {'query': pxt.String}\n", ")" ] }, { "cell_type": "code", "execution_count": 151, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 151, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 1: classify the query intent\n", "router.add_computed_column(\n", " classify_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Classify this customer query into exactly one category: '\n", " 'technical, billing, or general. Reply with the single word only.\\n\\n'\n", " 'Query: ' + router.query,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "router.add_computed_column(\n", " intent=router.classify_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 152, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 152, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 2: route to a specialized system prompt based on the classification\n", "@pxt.udf\n", "def route_prompt(intent: str, query: str) -> list[dict]:\n", " \"\"\"Select a system prompt based on the classified intent.\"\"\"\n", " system_prompts = {\n", " 'technical': 'You are a senior technical support engineer. '\n", " 'Provide precise, step-by-step troubleshooting guidance.',\n", " 'billing': 'You are a billing specialist. '\n", " 'Be empathetic and clear about charges, refunds, and payment options.',\n", " 'general': 'You are a friendly customer service representative. '\n", " 'Answer helpfully and concisely.',\n", " }\n", " # Default to general if classification is unexpected\n", " system = system_prompts.get(\n", " intent.strip().lower(), system_prompts['general']\n", " )\n", " return [\n", " {'role': 'system', 'content': system},\n", " {'role': 'user', 'content': query},\n", " ]\n", "\n", "\n", "router.add_computed_column(\n", " routed_messages=route_prompt(router.intent, router.query)\n", ")" ] }, { "cell_type": "code", "execution_count": 153, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 153, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 3: generate the specialized response\n", "router.add_computed_column(\n", " response_raw=openai.chat_completions(\n", " messages=router.routed_messages, model='gpt-4o-mini'\n", " )\n", ")\n", "router.add_computed_column(\n", " response=router.response_raw.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 154, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inserted 3 rows with 0 errors in 6.93 s (0.43 rows/s)\n" ] }, { "data": { "text/html": [ "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
queryintentresponse
I was charged twice for my subscription last monthbillingI’m really sorry to hear that you've been charged twice for your subscription. I understand how frustrating this can be, and I'm here to help resolve the issue.\n", "\n", "Could you please provide me with your account details or the specific dates of the charges? This will help me look into the matter more effectively. Once I have that information, I can check if it was an error and discuss potential refunds if applicable. \n", "\n", "Additionally, we can review your payment options to ensure everything is set up correctly for the future. Thank you for your patience as we work this out!
What programming languages do you support?technicalAs a senior technical support engineer, I can provide assistance across a wide range of programming languages, including but not limited to:\n", "\n", "1. **Python**\n", "2. **JavaScript**\n", "3. **Java**\n", "4. **C#**\n", "5. **C++**\n", "6. **Ruby**\n", "7. **PHP**\n", "8. **Swift**\n", "9. **Go**\n", "10. **Rust**\n", "11. **Kotlin**\n", "12. **TypeScript**\n", "13. **SQL**\n", "14. **HTML/CSS** (for web development)\n", "\n", "If you have specific questions or issues regarding any of these programming languages, please provide details, and I will guide you through troubleshooting steps or answering your questions accordingly.
My API calls are returning 429 errors since this morningtechnicalA 429 error indicates "Too Many Requests," meaning your application is exceeding the rate limits set by the API provider. Here is a step-by-step troubleshooting guide to help you resolve this issue:\n", "\n", "### Step 1: Verify API Documentation\n", "1. **Check Rate Limits**: Review the API documentation to find the rate limits for your account and endpoints you are using. Determine if you are indeed exceeding these limits.\n", "\n", "### Step 2: Inspect Your API Call Patterns\n", "1. **Review Calls**: Analyze your API  ...... h details of the error and your current implementation.\n", "\n", "### Step 8: Scale Up (if applicable)\n", "1. **Review Plan**: If you're consistently hitting your rate limits and it’s affecting your application, consider upgrading to a higher plan if available.\n", "\n", "### Conclusion\n", "Follow these steps systematically to narrow down the cause of the 429 errors and address the issue. It’s essential to ensure that your application adheres to the API limits to maintain functionality and avoid service interruptions.
" ], "text/plain": [ " query intent \\\n", "0 I was charged twice for my subscription last m... billing \n", "1 What programming languages do you support? technical \n", "2 My API calls are returning 429 errors since th... technical \n", "\n", " response \n", "0 I’m really sorry to hear that you've been char... \n", "1 As a senior technical support engineer, I can ... \n", "2 A 429 error indicates \"Too Many Requests,\" mea... " ] }, "execution_count": 154, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Insert queries spanning different intents\n", "router.insert(\n", " [\n", " {\n", " 'query': 'My API calls are returning 429 errors since this morning'\n", " },\n", " {'query': 'I was charged twice for my subscription last month'},\n", " {'query': 'What programming languages do you support?'},\n", " ]\n", ")\n", "\n", "router.select(router.query, router.intent, router.response).collect()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Each query was classified and then handled by a specialized system prompt. The `intent` column is inspectable for every row, making it easy to audit routing decisions." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Pattern 3: Parallelization\n", "\n", "Run multiple independent LLM calls on the same input simultaneously, then combine the results.\n", "\n", "**Imperative approach:** `asyncio.gather` or thread pools.\n", "**Pixeltable approach:** add independent computed columns. The engine parallelizes them automatically because they share no dependencies.\n", "\n", "```\n", " ┌→ sentiment ─┐\n", "input ──┼→ entities ──┼→ merge → combined output\n", " └→ summary ─┘\n", "```" ] }, { "cell_type": "code", "execution_count": 155, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'parallel'.\n" ] } ], "source": [ "parallel = pxt.create_table(\n", " 'agentic_patterns/parallel', {'text': pxt.String}\n", ")" ] }, { "cell_type": "code", "execution_count": 156, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 156, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Three independent LLM calls — Pixeltable runs them in parallel automatically\n", "parallel.add_computed_column(\n", " sentiment_raw=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Analyze the sentiment of this text. '\n", " 'Reply with: positive, negative, or neutral.\\n\\n'\n", " + parallel.text,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "parallel.add_computed_column(\n", " sentiment=parallel.sentiment_raw.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")\n", "\n", "parallel.add_computed_column(\n", " entities_raw=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Extract all named entities (people, companies, locations) '\n", " 'from this text. Return a comma-separated list.\\n\\n'\n", " + parallel.text,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "parallel.add_computed_column(\n", " entities=parallel.entities_raw.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")\n", "\n", "parallel.add_computed_column(\n", " summary_raw=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Summarize this text in one sentence.\\n\\n'\n", " + parallel.text,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "parallel.add_computed_column(\n", " summary=parallel.summary_raw.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 157, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 157, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Merge the parallel results into a single structured report\n", "@pxt.udf\n", "def merge_analysis(sentiment: str, entities: str, summary: str) -> dict:\n", " \"\"\"Combine parallel analysis results into one report.\"\"\"\n", " return {\n", " 'sentiment': sentiment.strip(),\n", " 'entities': entities.strip(),\n", " 'summary': summary.strip(),\n", " }\n", "\n", "\n", "parallel.add_computed_column(\n", " report=merge_analysis(\n", " parallel.sentiment, parallel.entities, parallel.summary\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "parallel.insert(\n", " [\n", " {\n", " 'text': 'Apple announced record quarterly revenue of $124 billion, '\n", " 'driven by strong iPhone sales in Europe and Asia. CEO Tim Cook '\n", " \"expressed optimism about the company's AI initiatives, while \"\n", " 'some analysts remain cautious about increased R&D spending.'\n", " }\n", " ]\n", ")\n", "\n", "parallel.select(\n", " parallel.text, parallel.sentiment, parallel.entities, parallel.summary\n", ").collect()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The three LLM calls (`sentiment`, `entities`, `summary`) have no dependency on each other, so Pixeltable dispatches them concurrently. The `merge_analysis` UDF waits for all three before combining the results. No async code required." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Pattern 4: Tool Use\n", "\n", "Give an LLM access to external functions it can call to gather information or take action.\n", "\n", "**Imperative approach:** `@function_tool` decorator, tool loop that re-prompts until the LLM stops requesting tools.\n", "**Pixeltable approach:** `pxt.tools()` bundles UDFs into tool definitions; `invoke_tools()` executes the LLM's choices — both as computed columns.\n", "\n", "```\n", "input → LLM (with tools) → invoke_tools() → results\n", "```\n", "\n", "For a deeper walkthrough including MCP servers, see [Use tool calling with LLMs](https://docs.pixeltable.com/howto/cookbooks/agents/llm-tool-calling)." ] }, { "cell_type": "code", "execution_count": 159, "metadata": {}, "outputs": [], "source": [ "# Define tool functions as UDFs\n", "@pxt.udf\n", "def get_weather(city: str) -> str:\n", " \"\"\"Get the current weather for a city.\"\"\"\n", " weather_data = {\n", " 'new york': 'Sunny, 72F',\n", " 'london': 'Cloudy, 58F',\n", " 'tokyo': 'Rainy, 65F',\n", " 'paris': 'Partly cloudy, 68F',\n", " }\n", " return weather_data.get(\n", " city.lower(), f'Weather data not available for {city}'\n", " )\n", "\n", "\n", "@pxt.udf\n", "def get_stock_price(symbol: str) -> str:\n", " \"\"\"Get the current stock price for a ticker symbol.\"\"\"\n", " prices = {'AAPL': '$178.50', 'GOOGL': '$141.25', 'MSFT': '$378.90'}\n", " return prices.get(symbol.upper(), f'Price not available for {symbol}')\n", "\n", "\n", "# Bundle into a Tools object\n", "tools = pxt.tools(get_weather, get_stock_price)" ] }, { "cell_type": "code", "execution_count": 160, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'tool_agent'.\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 160, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Create the tool-calling pipeline\n", "tool_agent = pxt.create_table(\n", " 'agentic_patterns/tool_agent', {'query': pxt.String}\n", ")\n", "\n", "# LLM decides which tool(s) to call\n", "tool_agent.add_computed_column(\n", " response=openai.chat_completions(\n", " messages=[{'role': 'user', 'content': tool_agent.query}],\n", " model='gpt-4o-mini',\n", " tools=tools,\n", " )\n", ")\n", "\n", "# Execute the tool calls automatically\n", "tool_agent.add_computed_column(\n", " tool_output=openai.invoke_tools(tools, tool_agent.response)\n", ")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "tool_agent.insert(\n", " [\n", " {'query': \"What's the weather in Tokyo?\"},\n", " {'query': \"What's Apple's stock price?\"},\n", " {\n", " 'query': \"What's the weather in Paris and Microsoft's stock price?\"\n", " },\n", " ]\n", ")\n", "\n", "for row in tool_agent.select(\n", " tool_agent.query, tool_agent.tool_output\n", ").collect():\n", " print(f'Query: {row[\"query\"]}')\n", " for tool_name, results in (row['tool_output'] or {}).items():\n", " if results:\n", " print(f' -> {tool_name}: {results}')\n", " print()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The LLM chose which tools to invoke (including multiple tools for the last query). `invoke_tools()` executed them and stored results. The full LLM response is also persisted in the `response` column for debugging." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Pattern 5: Evaluator-Optimizer\n", "\n", "One LLM generates output, a second LLM evaluates it, and the results are used to decide whether to refine. This is the architectural cousin of the *Reflection* pattern from Taxonomy 1 — an agent critiques its own output and iteratively improves it.\n", "\n", "**Imperative approach:** a while-loop that re-prompts until a quality threshold is met (see [Pixelagent's reflection example](https://github.com/pixeltable/pixelagent/tree/main/examples/reflection)).\n", "**Pixeltable approach:** chained computed columns — generate, evaluate, then conditionally refine. The evaluation score is stored alongside the content for analysis.\n", "\n", "```\n", "input → generate → evaluate (score + feedback) → refine if needed → output\n", "```" ] }, { "cell_type": "code", "execution_count": 162, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'evaluator'.\n" ] } ], "source": [ "evaluator = pxt.create_table(\n", " 'agentic_patterns/evaluator', {'product_brief': pxt.String}\n", ")" ] }, { "cell_type": "code", "execution_count": 163, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 163, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 1: generate initial marketing copy\n", "evaluator.add_computed_column(\n", " gen_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Write a short marketing tagline (one sentence) for this product:\\n\\n'\n", " + evaluator.product_brief,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "evaluator.add_computed_column(\n", " first_draft=evaluator.gen_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 164, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 164, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 2: evaluate the draft with an LLM-as-judge\n", "evaluator.add_computed_column(\n", " eval_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Rate this marketing tagline on a scale of 1-10 for clarity, '\n", " 'creativity, and persuasiveness. Then provide one sentence of feedback '\n", " 'for improvement.\\n\\n'\n", " 'Tagline: ' + evaluator.first_draft + '\\n\\n'\n", " 'Reply in this exact format:\\n'\n", " 'Score: \\nFeedback: ',\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "evaluator.add_computed_column(\n", " evaluation=evaluator.eval_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 165, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 165, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 3: refine using the feedback\n", "evaluator.add_computed_column(\n", " refine_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Improve this marketing tagline based on the feedback below. '\n", " 'Return only the improved tagline.\\n\\n'\n", " 'Original: ' + evaluator.first_draft + '\\n\\n'\n", " 'Feedback: ' + evaluator.evaluation,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "evaluator.add_computed_column(\n", " refined=evaluator.refine_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 166, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inserted 2 rows with 0 errors in 2.95 s (0.68 rows/s)\n" ] }, { "data": { "text/html": [ "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
product_brieffirst_draftevaluationrefined
A noise-canceling headphone designed for open-plan offices, with 30-hour battery life and a built-in microphone for calls."Stay focused and connected in open offices with our 30-hour noise-canceling headphones—your ultimate work companion!"Score: 8  \n", "Feedback: Consider simplifying the phrasing to enhance clarity and impact, such as "Connect and concentrate anywhere with our 30-hour noise-canceling headphones!""Concentrate and connect anywhere with our 30-hour noise-canceling headphones!"
An AI-powered code review tool that catches bugs, suggests improvements, and learns your team's coding style over time."Elevate your code quality with our AI-driven review tool that catches bugs, enhances style, and evolves with your team!"Score: 8  \n", "Feedback: To enhance clarity, consider simplifying the phrasing to make it more concise and impactful."Boost your code quality with our AI review tool that catches bugs, improves style, and grows with your team!"
" ], "text/plain": [ " product_brief \\\n", "0 A noise-canceling headphone designed for open-... \n", "1 An AI-powered code review tool that catches bu... \n", "\n", " first_draft \\\n", "0 \"Stay focused and connected in open offices wi... \n", "1 \"Elevate your code quality with our AI-driven ... \n", "\n", " evaluation \\\n", "0 Score: 8 \\nFeedback: Consider simplifying the... \n", "1 Score: 8 \\nFeedback: To enhance clarity, cons... \n", "\n", " refined \n", "0 \"Concentrate and connect anywhere with our 30-... \n", "1 \"Boost your code quality with our AI review to... " ] }, "execution_count": 166, "metadata": {}, "output_type": "execute_result" } ], "source": [ "evaluator.insert(\n", " [\n", " {\n", " 'product_brief': 'A noise-canceling headphone designed for open-plan offices, '\n", " 'with 30-hour battery life and a built-in microphone for calls.'\n", " },\n", " {\n", " 'product_brief': 'An AI-powered code review tool that catches bugs, suggests '\n", " \"improvements, and learns your team's coding style over time.\"\n", " },\n", " ]\n", ")\n", "\n", "evaluator.select(\n", " evaluator.product_brief,\n", " evaluator.first_draft,\n", " evaluator.evaluation,\n", " evaluator.refined,\n", ").collect()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Both the first draft and the refined version are stored side-by-side with the evaluation. This makes it straightforward to compare outputs, audit the judge's reasoning, or filter rows where the score fell below a threshold." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Pattern 6: Orchestrator-Worker\n", "\n", "A central agent decomposes a task, delegates sub-tasks to specialized worker agents, and synthesizes the results. This is the architectural cousin of the *Multi-Agent* pattern from Taxonomy 1, and the same structure Anthropic uses in their [multi-agent research system](https://www.anthropic.com/engineering/multi-agent-research-system) — a lead agent coordinates parallel subagents, each with their own context and tools.\n", "\n", "**Imperative approach:** an orchestrator agent class that spawns worker agent instances and collects their outputs.\n", "**Pixeltable approach:** each worker is a table with computed columns, wrapped as a callable function via `pxt.udf(table, return_value=...)`. The orchestrator table calls these functions as computed columns.\n", "\n", "```\n", "input → decompose → worker A (summarizer) ─┐\n", " → worker B (fact-checker) ─┼→ synthesize → output\n", "```\n", "\n", "For more on table UDFs, see [Use a table pipeline as a reusable function](https://docs.pixeltable.com/howto/cookbooks/agents/pattern-table-as-udf)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Build worker agents as tables" ] }, { "cell_type": "code", "execution_count": 167, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'summarizer'.\n", "Added 0 column values with 0 errors in 0.10 s\n", "Added 0 column values with 0 errors in 0.06 s\n" ] } ], "source": [ "# Worker A: summarizer\n", "summarizer_tbl = pxt.create_table(\n", " 'agentic_patterns/summarizer', {'text': pxt.String}\n", ")\n", "summarizer_tbl.add_computed_column(\n", " response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Summarize this text in 2-3 sentences:\\n\\n'\n", " + summarizer_tbl.text,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "summarizer_tbl.add_computed_column(\n", " summary=summarizer_tbl.response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")\n", "\n", "# Wrap as a callable function\n", "summarize = pxt.udf(summarizer_tbl, return_value=summarizer_tbl.summary)" ] }, { "cell_type": "code", "execution_count": 168, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'checker'.\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.02 s\n" ] } ], "source": [ "# Worker B: fact-checker\n", "checker_tbl = pxt.create_table(\n", " 'agentic_patterns/checker', {'claim': pxt.String}\n", ")\n", "checker_tbl.add_computed_column(\n", " response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Assess whether this claim is plausible. '\n", " 'Reply with: PLAUSIBLE or DUBIOUS, followed by a one-sentence explanation.\\n\\n'\n", " 'Claim: ' + checker_tbl.claim,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "checker_tbl.add_computed_column(\n", " assessment=checker_tbl.response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")\n", "\n", "# Wrap as a callable function\n", "fact_check = pxt.udf(checker_tbl, return_value=checker_tbl.assessment)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Build the orchestrator" ] }, { "cell_type": "code", "execution_count": 169, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'orchestrator'.\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 169, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Orchestrator table: delegates to workers, then synthesizes\n", "orchestrator = pxt.create_table(\n", " 'agentic_patterns/orchestrator', {'article': pxt.String}\n", ")\n", "\n", "# Dispatch to worker A (summarizer) and worker B (fact-checker) in parallel\n", "orchestrator.add_computed_column(\n", " summary=summarize(text=orchestrator.article)\n", ")\n", "orchestrator.add_computed_column(\n", " fact_check_result=fact_check(claim=orchestrator.article)\n", ")" ] }, { "cell_type": "code", "execution_count": 170, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.02 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 170, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Synthesize worker outputs into a final briefing\n", "orchestrator.add_computed_column(\n", " synth_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Based on the summary and fact-check below, write a brief '\n", " 'editorial note (2-3 sentences) about this article.\\n\\n'\n", " 'Summary: ' + orchestrator.summary + '\\n\\n'\n", " 'Fact-check: ' + orchestrator.fact_check_result,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "orchestrator.add_computed_column(\n", " briefing=orchestrator.synth_response.choices[\n", " 0\n", " ].message.content.astype(pxt.String)\n", ")" ] }, { "cell_type": "code", "execution_count": 171, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Inserted 1 row with 0 errors in 4.69 s (0.21 rows/s)\n" ] }, { "data": { "text/html": [ "\n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", " \n", "
summaryfact_check_resultbriefing
A study in Nature reveals that global sea levels have risen by 4.5 mm annually over the past decade, nearly double the rate seen in the 1990s, mainly due to ice sheet loss in Greenland and Antarctica and the thermal expansion of ocean water. These findings indicate that coastal cities could experience serious flooding risks by 2050 if strong mitigation efforts are not implemented.PLAUSIBLE - The claim aligns with current scientific understanding of sea level rise trends due to ice melt and thermal expansion, and studies in reputable journals like Nature often report on these alarming changes.This article highlights alarming findings from a recent study in *Nature*, underscoring the urgent need for action to combat climate change as global sea levels rise at an unprecedented rate. With rising tides threatening coastal cities by 2050, the report serves as a crucial reminder of the pressing implications of ice sheet loss and ocean warming. It is imperative that we heed these warnings and implement robust mitigation strategies to safeguard vulnerable communities and ecosystems.
" ], "text/plain": [ " summary \\\n", "0 A study in Nature reveals that global sea leve... \n", "\n", " fact_check_result \\\n", "0 PLAUSIBLE - The claim aligns with current scie... \n", "\n", " briefing \n", "0 This article highlights alarming findings from... " ] }, "execution_count": 171, "metadata": {}, "output_type": "execute_result" } ], "source": [ "orchestrator.insert(\n", " [\n", " {\n", " 'article': 'A recent study published in Nature found that global sea levels '\n", " 'rose by 4.5 mm per year over the last decade, nearly double the rate observed '\n", " 'in the 1990s. Researchers attribute the acceleration primarily to ice sheet '\n", " 'loss in Greenland and Antarctica, compounded by thermal expansion of ocean '\n", " 'water. The findings suggest coastal cities may face significant flooding risks '\n", " 'by 2050 without aggressive mitigation strategies.'\n", " }\n", " ]\n", ")\n", "\n", "orchestrator.select(\n", " orchestrator.summary,\n", " orchestrator.fact_check_result,\n", " orchestrator.briefing,\n", ").collect()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The orchestrator table called two independent worker pipelines (`summarize` and `fact_check`), each backed by their own table with full intermediate-result persistence. The synthesis step consumed both outputs to produce the final briefing. Adding a new worker (e.g., a tone analyzer) requires only creating another table, wrapping it with `pxt.udf()`, and adding one more computed column to the orchestrator." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Strategy A: ReAct\n", "\n", "ReAct is not a wiring pattern — it is a **reasoning strategy** that can be applied inside any of the six patterns above. The agent alternates between reasoning about the next step and acting on it (typically via tools), observing the result before deciding what to do next.\n", "\n", "**Imperative approach:** a while-loop that parses the LLM's THOUGHT/ACTION output, calls tools, and feeds observations back (see [Pixelagent's ReAct example](https://github.com/pixeltable/pixelagent/tree/main/examples/planning)).\n", "**Pixeltable approach:** the reasoning loop lives in a UDF that inserts rows into a tool-calling table and reads back results. The table stores every thought-action-observation triple for full observability.\n", "\n", "```\n", "question → [THOUGHT → ACTION → OBSERVATION] × N → final answer\n", "```" ] }, { "cell_type": "code", "execution_count": 172, "metadata": {}, "outputs": [], "source": [ "import re\n", "\n", "# Define a tool for the ReAct agent\n", "\n", "\n", "@pxt.udf\n", "def lookup_population(country: str) -> str:\n", " \"\"\"Look up the approximate population of a country.\"\"\"\n", " populations = {\n", " 'united states': '331 million',\n", " 'china': '1.4 billion',\n", " 'india': '1.4 billion',\n", " 'germany': '84 million',\n", " 'brazil': '214 million',\n", " 'japan': '125 million',\n", " }\n", " return populations.get(\n", " country.lower(), f'Population data not available for {country}'\n", " )\n", "\n", "\n", "react_tools = pxt.tools(lookup_population)" ] }, { "cell_type": "code", "execution_count": 173, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'react_steps'.\n", "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.00 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 173, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Build a tool-calling table that the ReAct loop will insert into\n", "react_steps = pxt.create_table(\n", " 'agentic_patterns/react_steps',\n", " {'step': pxt.Int, 'prompt': pxt.String, 'system_prompt': pxt.String},\n", ")\n", "\n", "react_steps.add_computed_column(\n", " response=openai.chat_completions(\n", " messages=[\n", " {'role': 'system', 'content': react_steps.system_prompt},\n", " {'role': 'user', 'content': react_steps.prompt},\n", " ],\n", " model='gpt-4o-mini',\n", " tools=react_tools,\n", " )\n", ")\n", "react_steps.add_computed_column(\n", " answer=react_steps.response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")\n", "react_steps.add_computed_column(\n", " tool_output=openai.invoke_tools(react_tools, react_steps.response)\n", ")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# The ReAct loop: reason → act → observe, repeated until done\n", "REACT_SYSTEM = (\n", " \"You are a research assistant. Answer the user's question step by step.\\n\"\n", " 'Available tools: lookup_population\\n\\n'\n", " 'On each turn, respond in this exact format:\\n'\n", " 'THOUGHT: \\n'\n", " 'ACTION: \\n\\n'\n", " 'When ACTION is FINAL, include your final answer after it.\\n'\n", " 'Current step: {step} of {max_steps}.'\n", ")\n", "\n", "question = 'Which country has a larger population, Brazil or Germany?'\n", "max_steps = 4\n", "history = []\n", "\n", "for step in range(1, max_steps + 1):\n", " # Build prompt with accumulated observations\n", " prompt = question\n", " if history:\n", " prompt += '\\n\\nPrevious observations:\\n' + '\\n'.join(history)\n", "\n", " system = REACT_SYSTEM.format(step=step, max_steps=max_steps)\n", "\n", " react_steps.insert(\n", " [{'step': step, 'prompt': prompt, 'system_prompt': system}]\n", " )\n", "\n", " # Read back the result for this step\n", " row = (\n", " react_steps.where(react_steps.step == step)\n", " .select(react_steps.answer, react_steps.tool_output)\n", " .collect()\n", " )\n", " answer_text = row['answer'][0] or ''\n", " tool_out = row['tool_output'][0]\n", "\n", " # Record observation from tool output (if any)\n", " if tool_out:\n", " history.append(f'Step {step} tool result: {tool_out}')\n", "\n", " # Check if the agent decided to finalize\n", " if 'FINAL' in answer_text.upper():\n", " break\n", "\n", "print(f'Completed in {step} steps')\n", "for row in react_steps.select(\n", " react_steps.step, react_steps.answer, react_steps.tool_output\n", ").collect():\n", " print(f'Step {row[\"step\"]}:')\n", " if row['answer']:\n", " print(f' {row[\"answer\"][:200]}')\n", " for tool_name, results in (row['tool_output'] or {}).items():\n", " if results:\n", " print(f' -> {tool_name}: {results}')\n", " print()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Every thought, action, and observation is persisted as a row in the `react_steps` table. The loop itself is plain Python; the LLM calls and tool execution happen declaratively via computed columns. This makes the reasoning trace fully queryable after the fact — useful for debugging or evaluation." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Strategy B: Planning\n", "\n", "Planning is the second cross-cutting reasoning strategy. Instead of acting step-by-step (ReAct), the agent first generates a complete plan, then executes each step. This is especially effective for complex tasks where the structure of the solution can be determined upfront.\n", "\n", "**Imperative approach:** an LLM generates a plan as structured JSON, then a loop executes each step (see [Pixelagent's planning example](https://github.com/pixeltable/pixelagent/tree/main/examples/planning)).\n", "**Pixeltable approach:** a prompt-chaining pipeline where the first column generates the plan and a UDF parses it into executable steps. Each step then feeds into subsequent computed columns.\n", "\n", "```\n", "question → generate plan → execute step 1 → execute step 2 → ... → synthesize\n", "```" ] }, { "cell_type": "code", "execution_count": 175, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created table 'planner'.\n", "Added 0 column values with 0 errors in 0.00 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 175, "metadata": {}, "output_type": "execute_result" } ], "source": [ "import json as json_mod\n", "\n", "planner = pxt.create_table(\n", " 'agentic_patterns/planner', {'question': pxt.String}\n", ")\n", "\n", "# Step 1: generate a plan as structured JSON\n", "planner.add_computed_column(\n", " plan_response=openai.chat_completions(\n", " messages=[\n", " {\n", " 'role': 'user',\n", " 'content': 'Break this question into 2-3 research steps. '\n", " 'Return ONLY a JSON object like {\"steps\": [\"sub-question 1\", \"sub-question 2\"]}. '\n", " 'No other text.\\n\\n'\n", " 'Question: ' + planner.question,\n", " }\n", " ],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "planner.add_computed_column(\n", " plan_text=planner.plan_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": 176, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 176, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 2: parse the plan and execute each sub-question, then synthesize\n", "@pxt.udf\n", "def execute_plan(plan_json: str, original_question: str) -> list[dict]:\n", " \"\"\"Parse the plan JSON and return structured sub-questions.\"\"\"\n", " try:\n", " data = json_mod.loads(plan_json)\n", " # Handle both {\"steps\": [...]} and direct [...]\n", " steps = (\n", " data\n", " if isinstance(data, list)\n", " else data.get('steps', data.get('questions', []))\n", " )\n", " return [\n", " {'step': i + 1, 'sub_question': q}\n", " for i, q in enumerate(steps)\n", " ]\n", " except (json_mod.JSONDecodeError, TypeError):\n", " return [{'step': 1, 'sub_question': original_question}]\n", "\n", "\n", "planner.add_computed_column(\n", " plan_steps=execute_plan(planner.plan_text, planner.question)\n", ")" ] }, { "cell_type": "code", "execution_count": 177, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n", "Added 0 column values with 0 errors in 0.01 s\n" ] }, { "data": { "text/plain": [ "No rows affected." ] }, "execution_count": 177, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Step 3: execute the plan — answer each sub-question, then synthesize\n", "@pxt.udf\n", "def format_plan_for_execution(\n", " plan_steps: list[dict], original_question: str\n", ") -> str:\n", " \"\"\"Format the plan steps into a single execution prompt.\"\"\"\n", " step_list = '\\n'.join(\n", " f'{s[\"step\"]}. {s[\"sub_question\"]}' for s in plan_steps\n", " )\n", " return (\n", " f'Answer each of these research sub-questions briefly, '\n", " f'then provide a final synthesis that answers the original question.\\n\\n'\n", " f'Original question: {original_question}\\n\\n'\n", " f'Sub-questions:\\n{step_list}'\n", " )\n", "\n", "\n", "planner.add_computed_column(\n", " exec_prompt=format_plan_for_execution(\n", " planner.plan_steps, planner.question\n", " )\n", ")\n", "\n", "planner.add_computed_column(\n", " exec_response=openai.chat_completions(\n", " messages=[{'role': 'user', 'content': planner.exec_prompt}],\n", " model='gpt-4o-mini',\n", " )\n", ")\n", "planner.add_computed_column(\n", " final_answer=planner.exec_response.choices[0].message.content.astype(\n", " pxt.String\n", " )\n", ")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "planner.insert(\n", " [\n", " {\n", " 'question': 'What are the economic and environmental trade-offs of electric vehicles vs hydrogen fuel cells?'\n", " }\n", " ]\n", ")\n", "\n", "row = planner.select(\n", " planner.question, planner.plan_text, planner.final_answer\n", ").collect()\n", "print('Plan:', row['plan_text'][0])\n", "print()\n", "print('Answer:', row['final_answer'][0][:500])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The plan (stored in `plan_steps`) is fully inspectable. The execution step answers all sub-questions in a single LLM call, but this could also use parallelization (Pattern 3) to answer each sub-question independently and merge the results. Planning and ReAct compose naturally with any of the six architectural patterns." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Choosing a Pattern\n", "\n", "### Six architectural patterns\n", "\n", "| Use case | Pattern | Key Pixeltable feature |\n", "|---|---|---|\n", "| Multi-step content generation | **Prompt Chaining** | Sequential computed columns |\n", "| Intent-based request handling | **Routing** | Classification column + UDF routing |\n", "| Independent analyses on same input | **Parallelization** | Independent computed columns (auto-parallel) |\n", "| LLM needs external data or actions | **Tool Use** | `pxt.tools()` + `invoke_tools()` |\n", "| Quality assurance / self-improvement | **Evaluator-Optimizer** | LLM-as-judge + refinement columns |\n", "| Complex multi-agent workflows | **Orchestrator-Worker** | `pxt.udf(table, return_value=...)` |\n", "\n", "### Two cross-cutting reasoning strategies\n", "\n", "| Strategy | When to use | How it layers in |\n", "|---|---|---|\n", "| **ReAct** | The agent needs to reason step-by-step and call tools based on intermediate observations | Loop that inserts rows into a tool-calling table; every thought-action-observation is persisted |\n", "| **Planning** | The full structure of the task can be determined upfront before execution | First column generates a plan; downstream columns execute and synthesize |\n", "\n", "Patterns compose naturally. An orchestrator-worker system might use routing in the orchestrator, tool use within a worker, and ReAct reasoning inside the tool-calling loop. Because each pattern is just a set of computed columns on a table, combining them requires no special glue code." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## See Also\n", "\n", "**Pixeltable cookbooks:**\n", "\n", "- [Use tool calling with LLMs](https://docs.pixeltable.com/howto/cookbooks/agents/llm-tool-calling) — deep dive into `pxt.tools()`, `invoke_tools()`, and MCP server integration\n", "- [Build an agent with persistent memory](https://docs.pixeltable.com/howto/cookbooks/agents/pattern-agent-memory) — embedding indexes for semantic memory recall\n", "- [Build a RAG pipeline](https://docs.pixeltable.com/howto/cookbooks/agents/pattern-rag-pipeline) — document chunking, embedding, and retrieval-augmented generation\n", "- [Look up structured data with retrieval UDFs](https://docs.pixeltable.com/howto/cookbooks/agents/pattern-data-lookup) — `pxt.retrieval_udf()` for key-based lookups\n", "- [Use a table pipeline as a reusable function](https://docs.pixeltable.com/howto/cookbooks/agents/pattern-table-as-udf) — `pxt.udf(table)` explained in depth\n", "\n", "**Pixelagent examples** (imperative implementations of the same patterns):\n", "\n", "- [Reflection loop](https://github.com/pixeltable/pixelagent/tree/main/examples/reflection) — main agent + critic agent with iterative refinement\n", "- [ReAct / Planning](https://github.com/pixeltable/pixelagent/tree/main/examples/planning) — step-by-step reasoning with tool calls\n", "- [Tool calling](https://github.com/pixeltable/pixelagent/tree/main/examples/tool-calling) — OpenAI, Anthropic, and Bedrock tool integration\n", "- [Memory](https://github.com/pixeltable/pixelagent/tree/main/examples/memory) — persistent and semantic memory management\n", "\n", "**External references:**\n", "\n", "- [OpenAI's Practical Guide to Building Agents](https://cdn.openai.com/business-guides-and-resources/a-practical-guide-to-building-agents.pdf) — the six architectural patterns\n", "- [Anthropic: How we built our multi-agent research system](https://www.anthropic.com/engineering/multi-agent-research-system) — orchestrator-worker at scale\n", "- [Pydantic AI: Multi-agent applications](https://ai.pydantic.dev/multi-agent-applications/#agent-delegation) — agent delegation patterns" ] } ], "metadata": { "kernelspec": { "display_name": "pixeltable", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.11" } }, "nbformat": 4, "nbformat_minor": 2 }