Compare commits

...
31 Commits
Author SHA1 Message Date
Mervin Praison a3bfc6f795 Merge pull request #563 from MervinPraison/claude/issue-459-20250531_155633
fix: resolve MCP tools not working in Streamlit applications (issue #459)
2025-05-31 17:05:57 +01:00
Mervin Praison ebc130854d Merge pull request #562 from MervinPraison/claude/issue-5-20250531_154718
feat: add deployment scheduling functionality
2025-05-31 17:04:49 +01:00
Mervin Praison f40a5c1359 Rename schedule_config.yaml.example to src/praisonai/schedule_config.yaml.example 2025-05-31 17:04:06 +01:00
Mervin Praison 3284f7bbc6 Rename test_scheduler.py to src/praisonai/tests/unit/test_scheduler.py 2025-05-31 17:03:34 +01:00
claude[bot]andMervinPraison 87928611f7 fix: resolve MCP tools not working in Streamlit applications (issue #459)
This commit provides working examples and comprehensive documentation for using MCP tools with Streamlit, without modifying core MCP implementation.

Key fixes:
- Proper session state management to prevent agent re-initialization
- Use standard LLM format instead of provider/model format
- Comprehensive error handling and user feedback
- Working examples with troubleshooting guidance

Files added:
- examples/python/ui/mcp-streamlit-airbnb.py - Complete working Streamlit app
- examples/python/ui/mcp-streamlit-simple.py - Simple fixed version of original code
- docs/ui/streamlit/mcp-streamlit.mdx - Comprehensive documentation and troubleshooting

The root issue was improper usage patterns in Streamlit, not the core MCP implementation.

Co-authored-by: MervinPraison <[email protected]>

🤖 Generated with [Claude Code](https://claude.ai/code)
2025-05-31 15:59:21 +00:00
Mervin Praison 2727bf5992 Merge pull request #561 from MervinPraison/claude/issue-170-20250531_154607
fix: resolve crawling NoneType concatenation error in realtime tools
2025-05-31 16:58:03 +01:00
claude[bot]andMervinPraison 45d04163c3 feat: add deployment scheduling functionality
- Add DeploymentScheduler with provider-agnostic design
- Support interval-based scheduling (daily, hourly, */6h, 3600s)
- CLI integration with --schedule and --schedule-config options
- Backward compatible with existing --deploy behavior
- Thread-safe scheduler with retry mechanism
- Example configuration file for scheduled deployments
- Test suite for validation

Addresses issue #5: --deploy schedule runs

Co-authored-by: MervinPraison <[email protected]>
2025-05-31 15:53:57 +00:00
claude[bot]andMervinPraison 477d0d2e26 fix: resolve crawling NoneType concatenation error in realtime tools
- Fix "can only concatenate str (not "NoneType") to str" error in web crawling
- Update process_tavily_results and fallback_to_duckduckgo to use proper async patterns
- Add null checks for crawl_result.markdown to prevent NoneType errors
- Use async context manager with AsyncWebCrawler and await crawler.arun()
- Match the working async pattern used in chat.py

Fixes #170

Co-authored-by: MervinPraison <[email protected]>
2025-05-31 15:48:02 +00:00
Mervin Praison 77394f5e5f Merge pull request #560 from MervinPraison/develop
Develop
2025-05-31 16:44:03 +01:00
MervinPraison 29e533c18c chore: remove unsloth training dependency from pyproject.toml
- Cleared the train dependencies list by removing unsloth[colab-new]>=2024.11.7.
- Commented out the unsloth dependency to maintain a record without affecting functionality.
2025-05-31 16:43:46 +01:00
MervinPraison 48dbea0566 Merge branch 'main' of https://github.com/MervinPraison/PraisonAI into develop 2025-05-31 12:42:45 +01:00
Mervin Praison 969d6697e1 Merge pull request #559 from MervinPraison/claude/issue-232-20250531_060542
fix: add dependency check for train command to prevent confusing errors
2025-05-31 12:42:22 +01:00
Mervin Praisonandcoderabbitai[bot] 787d2508b1 Update src/praisonai/pyproject.toml
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
2025-05-31 12:41:18 +01:00
claude[bot]andMervinPraison ec42dc92e4 feat: add unsloth training dependencies to pyproject.toml
- Add train optional dependencies with unsloth[colab-new]>=2024.11.7
- Update Poetry extras to include unsloth for train command
- Enables pip install "praisonai[train]" for training dependencies
- Maintains backward compatibility

Co-authored-by: MervinPraison <[email protected]>
2025-05-31 10:45:40 +00:00
claude[bot]andMervinPraison 5b7bae2f43 fix: add dependency check for train command to prevent confusing errors
- Add TRAIN_AVAILABLE flag to track unsloth dependency availability
- Add dependency validation in both parse_args() and main() methods
- Show clear error message with installation instructions when dependencies missing
- Follow same pattern as other commands (chat, code, call, realtime) for consistency
- Fixes issue where train command would show "train not available" without helpful context

Resolves #232

Co-authored-by: MervinPraison <[email protected]>
2025-05-31 06:07:31 +00:00
MervinPraison 69a1efac47 fix(tests): update expected name in tool call parsing test for malformed JSON
- Changed the expected function name in the test for malformed JSON from 'hello_world' to 'unknown_function' to reflect correct error handling behaviour.
- Adjusted the Python path in the test file to point to the correct source directory for the praisonai-agents module.
2025-05-31 06:51:09 +01:00
Mervin Praison e6381139ec Merge pull request #558 from MervinPraison/develop
Develop
2025-05-30 18:10:59 +01:00
MervinPraison aca2a58971 chore: update PraisonAI version to 2.2.22 across Dockerfiles and documentation
- Updated the version of PraisonAI from 2.2.21 to 2.2.22 in all relevant Dockerfiles.
- Adjusted README and local development documentation to reflect the new version.
- Ensured consistency in versioning across all instances where PraisonAI is referenced.
2025-05-30 18:10:42 +01:00
MervinPraison 7a07ebd05e Merge branch 'main' of https://github.com/MervinPraison/PraisonAI into develop 2025-05-30 18:06:46 +01:00
Mervin Praison 418d6c810b Merge pull request #557 from MervinPraison/claude/issue-471-20250530_151237
fix(ollama): resolve tool-call issues with local Ollama models
2025-05-30 18:04:17 +01:00
Mervin Praison 8bcfdfe713 Update and rename test_ollama_fix.py to src/praisonai/tests/unit/test_ollama_fix.py 2025-05-30 18:01:53 +01:00
claude[bot]andMervinPraison 57f2fa0471 fix(ollama): resolve tool-call issues with local Ollama models
- Fix tool parameter logic in agent.py to handle empty tools list properly
- Add comprehensive Ollama provider detection via model name and env vars
- Implement centralized tool call argument parsing with error handling
- Add consistent JSON error handling for all providers (sync and async)
- Users no longer need to pass tools to both agents AND tasks

Fixes #471

Co-authored-by: MervinPraison <[email protected]>
2025-05-30 15:17:06 +00:00
MervinPraison 2fdffd2d8d playwright mcp 2025-05-30 15:56:30 +01:00
Mervin Praison 471a6a106c Update article_generation_agents.ipynb 2025-05-30 15:41:18 +01:00
Mervin Praison 7c51c3aad1 Update article_generation_agents.ipynb 2025-05-30 15:40:48 +01:00
Mervin Praison dfa261f5ef Merge pull request #556 from MervinPraison/develop
chore: update PraisonAI version to 2.2.21 across Dockerfiles and docu…
2025-05-30 15:36:44 +01:00
MervinPraison 9a97f0b80f chore: update PraisonAI version to 2.2.21 across Dockerfiles and documentation
- Updated the version of PraisonAI from 2.2.20 to 2.2.21 in all relevant Dockerfiles.
- Adjusted README and local development documentation to reflect the new version.
- Ensured consistency in versioning across all instances where PraisonAI is referenced.
2025-05-30 15:36:03 +01:00
Mervin Praison eef9a5f7ae Merge pull request #555 from MervinPraison/claude/issue-18-20250530_125354
feat: Add comprehensive secondary market research agent system
2025-05-30 14:24:29 +01:00
Mervin Praison 510e9b1797 Merge pull request #554 from MervinPraison/claude/issue-23-20250530_125338
Add CSV processing examples for issue #23
2025-05-30 14:22:23 +01:00
claude[bot]andMervinPraison 43da66d129 feat: Add comprehensive secondary market research agent system
- Multi-agent system with 6 specialized research agents
- Customizable parameters for company, geography, industry, and sections
- YAML configuration for easy setup and customization
- FastAPI integration with async job processing and REST API
- Professional report generation suitable for business decision-making
- Jupyter notebook with multiple implementation examples
- Complete documentation and usage guidelines

Addresses issue #18 - Secondary market research agent with customization

Co-authored-by: MervinPraison <[email protected]>
2025-05-30 13:00:21 +00:00
claude[bot]andMervinPraison e1a61959ea feat: Add comprehensive CSV processing examples for issue #23
- Add simple CSV URL processor for basic use cases
- Add comprehensive CSV processing with 4 different methods
- Add complete documentation with setup and troubleshooting
- Address GitHub issue #23: "How to pass in a CSV file"
- Support sequential processing of URLs and other CSV data
- Include validation, error handling, and multiple processing patterns

Co-authored-by: MervinPraison <[email protected]>
2025-05-30 12:58:39 +00:00
45 changed files with 4169 additions and 141 deletions

No files matched your search

+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
flask \
"praisonai>=2.2.20" \
"praisonai>=2.2.22" \
"praisonai[api]" \
gunicorn \
markdown
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.2.20" \
"praisonai>=2.2.22" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.2.20" \
"praisonai>=2.2.22" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.2.20" \
"praisonai>=2.2.22" \
"praisonai[ui]" \
"praisonai[crewai]"
+2 -2
View File
@@ -121,7 +121,7 @@ healthcheck:
## 📦 Package Versions
All Docker images use consistent, up-to-date versions:
- PraisonAI: `>=2.2.20`
- PraisonAI: `>=2.2.22`
- PraisonAI Agents: `>=0.0.92`
- Python: `3.11-slim`
@@ -218,7 +218,7 @@ docker-compose up -d
### Version Pinning
To use specific versions, update the Dockerfile:
```dockerfile
RUN pip install "praisonai==2.2.20" "praisonaiagents==0.0.92"
RUN pip install "praisonai==2.2.22" "praisonaiagents==0.0.92"
```
## 🌐 Production Deployment
+1 -1
View File
@@ -110,7 +110,7 @@ Loads environment variables from .env file or system and sets them.</p>
file.write(&#34;FROM python:3.11-slim\n&#34;)
file.write(&#34;WORKDIR /app\n&#34;)
file.write(&#34;COPY . .\n&#34;)
file.write(&#34;RUN pip install flask praisonai==2.2.20 gunicorn markdown\n&#34;)
file.write(&#34;RUN pip install flask praisonai==2.2.22 gunicorn markdown\n&#34;)
file.write(&#34;EXPOSE 8080\n&#34;)
file.write(&#39;CMD [&#34;gunicorn&#34;, &#34;-b&#34;, &#34;0.0.0.0:8080&#34;, &#34;api:app&#34;]\n&#39;)
+1 -1
View File
@@ -27,7 +27,7 @@ WORKDIR /app
COPY . .
RUN pip install flask praisonai==2.2.20 watchdog
RUN pip install flask praisonai==2.2.22 watchdog
EXPOSE 5555
+54
View File
@@ -0,0 +1,54 @@
---
title: 'Playwright'
description: 'Using Playwright MCP with PraisonAI'
---
# Playwright MCP
[Playwright](https://playwright.dev/) is a powerful browser automation tool that can be used with the Model Context Protocol (MCP) to enable agents to interact with web browsers.
## Installation
To use Playwright with MCP, you'll need to install the Playwright MCP package:
```bash
npm install @playwright/mcp
```
## Running the Playwright MCP Server
You need to run the Playwright MCP server in a separate terminal before your agent can use it:
```bash
npx @playwright/mcp@latest --port 8931
```
This will start a Playwright MCP server on port 8931 that your agent can connect to.
## Using Playwright MCP with PraisonAI
Once the Playwright MCP server is running, you can create an agent that uses it as a tool:
```python
from praisonaiagents import Agent, MCP
search_agent = Agent(
instructions="""You help search the web.""",
llm="gpt-4o-mini",
tools=MCP("http://localhost:8931/sse")
)
search_agent.start("Find about Praison AI")
```
The agent will be able to use Playwright to automate browser interactions and search the web.
## Advanced Configuration
You can customize the Playwright MCP server by passing additional options:
```bash
npx @playwright/mcp@latest --port 8931 --browser chromium
```
For more information on available options, refer to the [Playwright MCP documentation](https://github.com/microsoft/playwright/tree/main/packages/mcp).
+1
View File
@@ -251,6 +251,7 @@
"mcp/anthropic",
"mcp/gemini",
"mcp/mistral",
"mcp/playwright",
"mcp/whatsapp",
"mcp/perplexity",
"mcp/ollama-python",
+2 -2
View File
@@ -31,7 +31,7 @@ export OPENAI_API_KEY=xxxxxxxxx
```python
import os
from crawl4ai import WebCrawler
from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from pydantic import BaseModel, Field
@@ -41,7 +41,7 @@ class OpenAIModelFee(BaseModel):
output_fee: str = Field(..., description="Fee for output token ßfor the OpenAI model.")
url = 'https://openai.com/api/pricing/'
crawler = WebCrawler()
crawler = AsyncWebCrawler()
crawler.warmup()
result = crawler.run(
+1 -1
View File
@@ -120,7 +120,7 @@ from praisonaiagents.tools import scrape_page, extract_links, crawl, extract_tex
# Create search agent
agent = Agent(
name="WebCrawler",
name="AsyncWebCrawler",
role="Web Scraping Specialist",
goal="Extract and analyze web content efficiently.",
backstory="Expert in web scraping and content extraction.",
+1 -1
View File
@@ -155,7 +155,7 @@ To facilitate local development with live reload, you can use Docker. Follow the
COPY . .
RUN pip install flask praisonai==2.2.20 watchdog
RUN pip install flask praisonai==2.2.22 watchdog
EXPOSE 5555
+1 -1
View File
@@ -208,7 +208,7 @@ To facilitate local development with live reload, you can use Docker. Follow the
COPY . .
RUN pip install flask praisonai==2.2.20 watchdog
RUN pip install flask praisonai==2.2.22 watchdog
EXPOSE 5555
+253
View File
@@ -0,0 +1,253 @@
---
title: "MCP with Streamlit"
description: "How to properly integrate MCP (Model Context Protocol) tools with PraisonAI Agents in Streamlit applications"
---
# MCP with Streamlit
This guide explains how to properly integrate MCP (Model Context Protocol) tools with PraisonAI Agents in Streamlit applications, addressing common issues and providing working solutions.
## Common Issues
When integrating MCP tools with Streamlit, users often encounter these problems:
### Issue #1: Agent Re-initialization
**Problem**: Agent gets re-initialized on every Streamlit interaction, causing MCP tools to fail.
**Solution**: Use Streamlit's session state to initialize the agent only once.
### Issue #2: LLM Provider Format
**Problem**: Using provider/model format like `"ollama/llama3.2"` can cause tool calling issues in Streamlit environments.
**Solution**: Use standard model names like `"gpt-4o-mini"` instead of provider/model format.
### Issue #3: Missing Error Handling
**Problem**: No user feedback when MCP initialization fails.
**Solution**: Implement comprehensive error handling with user-friendly messages.
## Working Example
Here's a complete working example that demonstrates the correct approach:
```python
import streamlit as st
from praisonaiagents import Agent
from praisonaiagents.mcp import MCP
import traceback
st.title("🏠 AI Airbnb Assistant")
# Configuration in sidebar
with st.sidebar:
st.header("⚙️ Configuration")
# Use standard model names, not provider/model format
llm_model = st.selectbox(
"Choose LLM Model",
options=[
"gpt-4o-mini", # ✅ Correct format
"gpt-4o",
"claude-3-5-sonnet-20241022"
],
index=0
)
debug_mode = st.checkbox("Enable Debug Mode", value=False)
# Initialize session state (CRITICAL for Streamlit)
if "agent_initialized" not in st.session_state:
st.session_state.agent_initialized = False
st.session_state.agent = None
st.session_state.init_error = None
# Function to initialize agent with proper error handling
def initialize_agent():
try:
with st.spinner("🔄 Initializing AI agent..."):
# Create MCP tools
mcp_tools = MCP(
"npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt",
timeout=60,
debug=debug_mode
)
# Create agent - this should only happen ONCE
agent = Agent(
instructions="You are a helpful Airbnb assistant...",
llm=llm_model, # Use standard format
tools=mcp_tools,
verbose=debug_mode
)
return agent, None
except Exception as e:
error_msg = f"Failed to initialize agent: {str(e)}"
if debug_mode:
error_msg += f"\n\nFull traceback:\n{traceback.format_exc()}"
return None, error_msg
# Agent initialization (only runs once)
if not st.session_state.agent_initialized:
if st.button("🚀 Initialize AI Assistant", type="primary"):
agent, error = initialize_agent()
if agent:
st.session_state.agent = agent
st.session_state.agent_initialized = True
st.session_state.init_error = None
st.success("✅ AI Assistant initialized successfully!")
st.rerun()
else:
st.session_state.init_error = error
st.error(f"❌ Initialization failed: {error}")
# Main interface (only show if agent is ready)
if st.session_state.agent_initialized and st.session_state.agent:
query = st.text_input("🔍 What are you looking for?")
if st.button("Search") and query:
try:
with st.spinner("🔍 Searching..."):
# Use the SAME agent instance from session state
result = st.session_state.agent.start(query)
st.write(result)
except Exception as e:
st.error(f"❌ Search failed: {str(e)}")
```
## Best Practices
### 1. Session State Management
Always use Streamlit's session state to manage agent lifecycle:
```python
# ✅ Correct - Initialize once in session state
if "agent" not in st.session_state:
st.session_state.agent = Agent(...)
# ❌ Wrong - Re-initializes on every interaction
agent = Agent(...)
```
### 2. LLM Model Selection
Use standard model names, avoid provider/model format in Streamlit:
```python
# ✅ Correct - Standard model names
llm="gpt-4o-mini"
llm="claude-3-5-sonnet-20241022"
# ❌ Problematic in Streamlit - Provider/model format
llm="ollama/llama3.2"
llm="anthropic/claude-3-5-sonnet"
```
### 3. Error Handling
Implement comprehensive error handling:
```python
try:
# MCP initialization
mcp_tools = MCP("your-mcp-command")
agent = Agent(tools=mcp_tools, ...)
except Exception as e:
st.error(f"Initialization failed: {str(e)}")
# Provide troubleshooting tips
```
### 4. User Feedback
Provide clear feedback during initialization:
```python
with st.spinner("🔄 Initializing AI agent and MCP tools..."):
# Initialization code here
pass
if initialization_successful:
st.success("✅ AI Assistant initialized successfully!")
else:
st.error("❌ Initialization failed")
```
## Troubleshooting
### MCP Tools Not Found
If you get "MCP tool cannot be found" errors:
1. **Check MCP Server Command**: Ensure the MCP server command is correct and the server starts successfully
2. **Verify Dependencies**: Make sure all required dependencies (Node.js, npm, etc.) are installed
3. **Test Manually**: Try running the MCP server command manually first
4. **Enable Debug Mode**: Set `debug=True` in MCP constructor for detailed logs
### Tool Calling Issues
If tools aren't being called properly:
1. **Use Standard LLM Format**: Avoid provider/model format like `"ollama/llama3.2"`
2. **Check API Keys**: Ensure proper environment variables are set for your chosen LLM
3. **Session State**: Verify agent is properly stored in session state
4. **Timeout Settings**: Increase timeout if needed: `MCP(command, timeout=120)`
### Threading Conflicts
If you encounter threading-related errors:
1. **Single Agent Instance**: Use only one agent instance per session
2. **Proper Cleanup**: Let Streamlit handle cleanup naturally
3. **Avoid Manual Threading**: Don't create additional threads in Streamlit
## Complete Working Example
For a complete, production-ready example, see:
- [`examples/python/ui/mcp-streamlit-airbnb.py`](https://github.com/MervinPraison/PraisonAI/blob/main/examples/python/ui/mcp-streamlit-airbnb.py)
This example includes:
- ✅ Proper session state management
- ✅ Comprehensive error handling
- ✅ User-friendly interface
- ✅ Debug mode support
- ✅ Troubleshooting guidance
- ✅ Configuration options
## Environment Setup
Make sure your environment has the required dependencies:
```bash
# Install PraisonAI Agents
pip install praisonaiagents
# Install Streamlit
pip install streamlit
# For Airbnb MCP server example
npm install -g @openbnb/mcp-server-airbnb
# Set environment variables for your chosen LLM
export OPENAI_API_KEY="your-key"
# or
export ANTHROPIC_API_KEY="your-key"
```
## Running the Example
```bash
# Run the working example
streamlit run examples/python/ui/mcp-streamlit-airbnb.py
# Or create your own based on the patterns above
streamlit run your-mcp-app.py
```
## Summary
The key to successfully using MCP with Streamlit is:
1. **Proper session state management** - Initialize agent only once
2. **Standard LLM format** - Avoid provider/model format in Streamlit
3. **Comprehensive error handling** - Provide user feedback
4. **Correct usage patterns** - Follow Streamlit best practices
By following these guidelines, MCP tools will work reliably in your Streamlit applications without requiring any modifications to the core MCP implementation.
@@ -7,7 +7,7 @@
},
"source": [
"# Article Generation Agents\n",
"[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/MervinPraison/PraisonAI/blob/main/cookbooks/notebooks/article_generation_agents.ipynb)"
"[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/MervinPraison/PraisonAI/blob/main/examples/cookbooks/yaml/article_generation_agents.ipynb)"
]
},
{
@@ -609,4 +609,4 @@
},
"nbformat": 4,
"nbformat_minor": 0
}
}
@@ -51,7 +51,7 @@
},
"outputs": [],
"source": [
"from crawl4ai import WebCrawler\n",
"from crawl4ai import AsyncWebCrawler\n",
"from crawl4ai.extraction_strategy import LLMExtractionStrategy\n",
"from pydantic import BaseModel, Field\n",
"from praisonai_tools import BaseTool\n",
@@ -66,7 +66,7 @@
" description: str = \"Extracts model fees for input and output tokens from the given pricing page.\"\n",
"\n",
" def _run(self, url: str):\n",
" crawler = WebCrawler()\n",
" crawler = AsyncWebCrawler()\n",
" crawler.warmup()\n",
"\n",
" result = crawler.run(\n",
@@ -196,8 +196,8 @@
"output_type": "stream",
"text": [
"[LOG] 🚀 Initializing LocalSeleniumCrawlerStrategy\n",
"[LOG] 🌤️ Warming up the WebCrawler\n",
"[LOG] 🌞 WebCrawler is ready to crawl\n",
"[LOG] 🌤️ Warming up the AsyncWebCrawler\n",
"[LOG] 🌞 AsyncWebCrawler is ready to crawl\n",
"[LOG] 🚀 Crawling done for https://openai.com/api/pricing/, success: True, time taken: 0.86 seconds\n",
"[LOG] 🚀 Content extracted for https://openai.com/api/pricing/, success: True, time taken: 0.01 seconds\n",
"[LOG] 🔥 Extracting semantic blocks for https://openai.com/api/pricing/, Strategy: LLMExtractionStrategy\n",
@@ -245,8 +245,8 @@
"output_type": "stream",
"text": [
"[LOG] 🚀 Initializing LocalSeleniumCrawlerStrategy\n",
"[LOG] 🌤️ Warming up the WebCrawler\n",
"[LOG] 🌞 WebCrawler is ready to crawl\n",
"[LOG] 🌤️ Warming up the AsyncWebCrawler\n",
"[LOG] 🌞 AsyncWebCrawler is ready to crawl\n",
"[LOG] 🚀 Crawling done for https://www.anthropic.com/pricing, success: True, time taken: 2.93 seconds\n",
"[LOG] 🚀 Content extracted for https://www.anthropic.com/pricing, success: True, time taken: 0.06 seconds\n",
"[LOG] 🔥 Extracting semantic blocks for https://www.anthropic.com/pricing, Strategy: LLMExtractionStrategy\n",
@@ -323,8 +323,8 @@
"output_type": "stream",
"text": [
"[LOG] 🚀 Initializing LocalSeleniumCrawlerStrategy\n",
"[LOG] 🌤️ Warming up the WebCrawler\n",
"[LOG] 🌞 WebCrawler is ready to crawl\n",
"[LOG] 🌤️ Warming up the AsyncWebCrawler\n",
"[LOG] 🌞 AsyncWebCrawler is ready to crawl\n",
"[LOG] 🚀 Crawling done for https://cohere.com/pricing, success: True, time taken: 6.34 seconds\n",
"[LOG] 🚀 Content extracted for https://cohere.com/pricing, success: True, time taken: 0.14 seconds\n",
"[LOG] 🔥 Extracting semantic blocks for https://cohere.com/pricing, Strategy: LLMExtractionStrategy\n",
@@ -0,0 +1,640 @@
{
"cells": [
{
"cell_type": "markdown",
"id": "1",
"metadata": {},
"source": [
"# Secondary Market Research Agents\n",
"[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/MervinPraison/PraisonAI/blob/main/cookbooks/notebooks/secondary_market_research_agents.ipynb)\n",
"\n",
"This notebook demonstrates how to create a comprehensive secondary market research system using PraisonAI agents. The system generates detailed reports with customizable parameters for company, geography, and industry analysis."
]
},
{
"cell_type": "markdown",
"id": "2",
"metadata": {},
"source": [
"## Features\n",
"\n",
"- **Multi-agent Research System**: Specialized agents for different research areas\n",
"- **Customizable Parameters**: Company, geography, industry, and research sections\n",
"- **Comprehensive Reports**: Market overview, competitive analysis, financial performance, growth opportunities, and risk assessment\n",
"- **FastAPI Integration**: Ready for production deployment\n",
"- **Structured Output**: Professional report format suitable for business decision-making"
]
},
{
"cell_type": "markdown",
"id": "3",
"metadata": {},
"source": [
"## Installation"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4",
"metadata": {},
"outputs": [],
"source": [
"# Install required dependencies\n",
"%pip install praisonai[crewai] > /dev/null\n",
"%pip install fastapi uvicorn > /dev/null\n",
"%pip install requests > /dev/null"
]
},
{
"cell_type": "markdown",
"id": "5",
"metadata": {},
"source": [
"## Setup API Key"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "6",
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"from google.colab import userdata\n",
"\n",
"# Set up OpenAI API key (add to Google Colab secrets)\n",
"os.environ[\"OPENAI_API_KEY\"] = userdata.get('OPENAI_API_KEY') or \"ENTER OPENAI_API_KEY HERE\"\n",
"os.environ[\"OPENAI_MODEL_NAME\"] = \"gpt-4o-mini\""
]
},
{
"cell_type": "markdown",
"id": "7",
"metadata": {},
"source": [
"## Method 1: Using YAML Configuration"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "8",
"metadata": {},
"outputs": [],
"source": [
"# YAML configuration for market research agents\n",
"agent_yaml = \"\"\"\n",
"framework: \"crewai\"\n",
"topic: \"Secondary Market Research Analysis\"\n",
"\n",
"# Configuration variables\n",
"variables:\n",
" company: \"Tesla\"\n",
" geography: \"North America\" \n",
" industry: \"Electric Vehicles\"\n",
"\n",
"roles:\n",
" market_overview_specialist:\n",
" role: \"Market Overview Specialist\" \n",
" backstory: |\n",
" You are a seasoned market research analyst with 10+ years of experience \n",
" in analyzing market dynamics, trends, and growth patterns.\n",
" goal: |\n",
" Analyze the {industry} market in {geography} to provide comprehensive \n",
" market overview including size, trends, and growth drivers.\n",
" verbose: true\n",
" allow_delegation: false\n",
" tools:\n",
" - \"InternetSearchTool\"\n",
" tasks:\n",
" market_overview_research:\n",
" description: |\n",
" Conduct comprehensive market overview research for {industry} in {geography}:\n",
" 1. Market size and valuation analysis\n",
" 2. Key growth trends and patterns\n",
" 3. Primary market drivers\n",
" 4. Market segmentation\n",
" expected_output: |\n",
" A comprehensive market overview report with quantitative data and insights.\n",
"\n",
" competitive_intelligence_analyst:\n",
" role: \"Competitive Intelligence Analyst\"\n",
" backstory: |\n",
" You are a competitive intelligence expert specializing in analyzing \n",
" competitive landscapes and market positioning.\n",
" goal: |\n",
" Analyze {company}'s competitive landscape in the {industry} market.\n",
" verbose: true\n",
" allow_delegation: false\n",
" tools:\n",
" - \"InternetSearchTool\"\n",
" tasks:\n",
" competitive_analysis:\n",
" description: |\n",
" Conduct detailed competitive analysis for {company}:\n",
" 1. Identify top 5-7 competitors\n",
" 2. Market share analysis\n",
" 3. Competitive positioning\n",
" 4. Strengths and weaknesses\n",
" expected_output: |\n",
" A comprehensive competitive intelligence report.\n",
"\n",
" financial_performance_analyst:\n",
" role: \"Financial Performance Analyst\"\n",
" backstory: |\n",
" You are a financial analyst with expertise in corporate finance \n",
" and industry benchmarking.\n",
" goal: |\n",
" Analyze {company}'s financial performance and industry benchmarks.\n",
" verbose: true\n",
" allow_delegation: false\n",
" tools:\n",
" - \"InternetSearchTool\"\n",
" tasks:\n",
" financial_analysis:\n",
" description: |\n",
" Conduct financial performance analysis for {company}:\n",
" 1. Revenue trends and growth\n",
" 2. Profitability metrics\n",
" 3. Financial ratios\n",
" 4. Industry benchmarking\n",
" expected_output: |\n",
" A detailed financial performance analysis report.\n",
"\n",
" research_report_synthesizer:\n",
" role: \"Research Report Synthesizer\"\n",
" backstory: |\n",
" You are an expert business report writer specializing in \n",
" synthesizing complex research into actionable insights.\n",
" goal: |\n",
" Create a comprehensive secondary market research report for {company}.\n",
" verbose: true\n",
" allow_delegation: false\n",
" tasks:\n",
" report_synthesis:\n",
" description: |\n",
" Synthesize all research findings into a comprehensive report:\n",
" 1. Executive Summary\n",
" 2. Market Overview\n",
" 3. Competitive Analysis\n",
" 4. Financial Performance\n",
" 5. Conclusions and Recommendations\n",
" expected_output: |\n",
" A complete, professionally formatted secondary market research report.\n",
" context:\n",
" - \"market_overview_research\"\n",
" - \"competitive_analysis\" \n",
" - \"financial_analysis\"\n",
"\n",
"dependencies:\n",
" - market_overview_specialist\n",
" - competitive_intelligence_analyst\n",
" - financial_performance_analyst\n",
" - research_report_synthesizer\n",
"\"\"\""
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9",
"metadata": {},
"outputs": [],
"source": [
"# Import required libraries\n",
"from praisonai import PraisonAI\n",
"from praisonai_tools import BaseTool\n",
"from duckduckgo_search import DDGS\n",
"import requests\n",
"from bs4 import BeautifulSoup\n",
"\n",
"# Create internet search tool\n",
"class InternetSearchTool(BaseTool):\n",
" name: str = \"InternetSearchTool\"\n",
" description: str = \"Search Internet for relevant information\"\n",
"\n",
" def _run(self, query: str):\n",
" ddgs = DDGS()\n",
" results = ddgs.text(keywords=query, region='wt-wt', safesearch='moderate', max_results=5)\n",
" return results\n",
"\n",
"# Create PraisonAI instance\n",
"praisonai = PraisonAI(agent_yaml=agent_yaml, tools=[InternetSearchTool])\n",
"\n",
"# Run the market research\n",
"print(\"🔍 Starting Secondary Market Research for Tesla...\")\n",
"result = praisonai.run()\n",
"\n",
"print(\"\\n📋 Market Research Complete!\")\n",
"print(\"=\" * 60)\n",
"print(result)"
]
},
{
"cell_type": "markdown",
"id": "10",
"metadata": {},
"source": [
"## Method 2: Using Python API with Custom Parameters"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "11",
"metadata": {},
"outputs": [],
"source": [
"# Import the market research system\n",
"from praisonaiagents import Agent, Task, PraisonAIAgents, Tools\n",
"from typing import Dict, List, Any\n",
"import asyncio\n",
"from datetime import datetime\n",
"\n",
"class MarketResearchConfig:\n",
" \"\"\"Configuration for market research parameters\"\"\"\n",
" \n",
" def __init__(self, company: str, geography: str, industry: str, sections: List[str]):\n",
" self.company = company\n",
" self.geography = geography\n",
" self.industry = industry\n",
" self.sections = sections\n",
" self.timestamp = datetime.now().isoformat()\n",
"\n",
"def create_market_research_agents(config: MarketResearchConfig):\n",
" \"\"\"Create specialized market research agents\"\"\"\n",
" \n",
" agents = {}\n",
" \n",
" if \"market_overview\" in config.sections:\n",
" agents[\"market_overview\"] = Agent(\n",
" name=\"Market Overview Specialist\",\n",
" role=\"Market Analysis Expert\",\n",
" goal=f\"Analyze the {config.industry} market in {config.geography}\",\n",
" instructions=f\"\"\"\n",
" Analyze market size, trends, growth drivers for {config.industry} in {config.geography}.\n",
" Provide quantitative data and cite reliable sources.\n",
" \"\"\",\n",
" tools=[Tools.internet_search],\n",
" verbose=True\n",
" )\n",
" \n",
" if \"competitive_analysis\" in config.sections:\n",
" agents[\"competitive\"] = Agent(\n",
" name=\"Competitive Intelligence Analyst\",\n",
" role=\"Competition Research Expert\", \n",
" goal=f\"Analyze {config.company}'s competitive landscape\",\n",
" instructions=f\"\"\"\n",
" Analyze {config.company}'s competitive position in {config.industry}.\n",
" Identify key competitors, market share, and competitive advantages.\n",
" \"\"\",\n",
" tools=[Tools.internet_search],\n",
" verbose=True\n",
" )\n",
" \n",
" if \"financial_performance\" in config.sections:\n",
" agents[\"financial\"] = Agent(\n",
" name=\"Financial Performance Analyst\", \n",
" role=\"Financial Research Expert\",\n",
" goal=f\"Analyze {config.company}'s financial performance\",\n",
" instructions=f\"\"\"\n",
" Research {config.company}'s financial performance, revenue trends,\n",
" profitability, and compare with industry benchmarks.\n",
" \"\"\",\n",
" tools=[Tools.internet_search],\n",
" verbose=True\n",
" )\n",
" \n",
" if \"growth_opportunities\" in config.sections:\n",
" agents[\"growth\"] = Agent(\n",
" name=\"Growth Opportunities Researcher\",\n",
" role=\"Strategic Growth Expert\",\n",
" goal=f\"Identify growth opportunities for {config.company}\",\n",
" instructions=f\"\"\"\n",
" Identify emerging opportunities, market gaps, expansion possibilities\n",
" for {config.company} in {config.industry}.\n",
" \"\"\",\n",
" tools=[Tools.internet_search],\n",
" verbose=True\n",
" )\n",
" \n",
" # Always include synthesizer\n",
" agents[\"synthesizer\"] = Agent(\n",
" name=\"Research Report Synthesizer\",\n",
" role=\"Report Writing Expert\",\n",
" goal=\"Synthesize research into comprehensive report\",\n",
" instructions=f\"\"\"\n",
" Create a professional secondary market research report for {config.company} \n",
" in {config.industry} with clear structure and actionable recommendations.\n",
" \"\"\",\n",
" verbose=True\n",
" )\n",
" \n",
" return agents\n",
"\n",
"def create_research_tasks(agents: Dict[str, Agent], config: MarketResearchConfig):\n",
" \"\"\"Create research tasks for the agents\"\"\"\n",
" \n",
" tasks = []\n",
" \n",
" if \"market_overview\" in config.sections and \"market_overview\" in agents:\n",
" market_task = Task(\n",
" name=\"market_overview_research\",\n",
" description=f\"Research {config.industry} market overview in {config.geography}\",\n",
" expected_output=\"Market overview analysis with key metrics and trends\",\n",
" agent=agents[\"market_overview\"]\n",
" )\n",
" tasks.append(market_task)\n",
" \n",
" if \"competitive_analysis\" in config.sections and \"competitive\" in agents:\n",
" competitive_task = Task(\n",
" name=\"competitive_analysis\",\n",
" description=f\"Analyze competitive landscape for {config.company}\",\n",
" expected_output=\"Competitive analysis with key competitors and positioning\",\n",
" agent=agents[\"competitive\"]\n",
" )\n",
" tasks.append(competitive_task)\n",
" \n",
" if \"financial_performance\" in config.sections and \"financial\" in agents:\n",
" financial_task = Task(\n",
" name=\"financial_analysis\",\n",
" description=f\"Analyze {config.company}'s financial performance\",\n",
" expected_output=\"Financial performance analysis with key metrics\",\n",
" agent=agents[\"financial\"]\n",
" )\n",
" tasks.append(financial_task)\n",
" \n",
" if \"growth_opportunities\" in config.sections and \"growth\" in agents:\n",
" growth_task = Task(\n",
" name=\"growth_opportunities\",\n",
" description=f\"Identify growth opportunities for {config.company}\",\n",
" expected_output=\"Growth opportunities with strategic recommendations\",\n",
" agent=agents[\"growth\"]\n",
" )\n",
" tasks.append(growth_task)\n",
" \n",
" # Synthesis task\n",
" synthesis_task = Task(\n",
" name=\"synthesis\",\n",
" description=\"Synthesize all findings into comprehensive report\",\n",
" expected_output=\"Complete secondary market research report\",\n",
" agent=agents[\"synthesizer\"],\n",
" context=tasks\n",
" )\n",
" tasks.append(synthesis_task)\n",
" \n",
" return tasks\n",
"\n",
"async def run_market_research(config: MarketResearchConfig):\n",
" \"\"\"Run the complete market research workflow\"\"\"\n",
" \n",
" print(f\"🔍 Starting Market Research for {config.company}\")\n",
" print(f\"📍 Geography: {config.geography}\")\n",
" print(f\"🏭 Industry: {config.industry}\")\n",
" print(f\"📊 Sections: {', '.join(config.sections)}\")\n",
" print(\"=\" * 60)\n",
" \n",
" # Create agents and tasks\n",
" agents = create_market_research_agents(config)\n",
" tasks = create_research_tasks(agents, config)\n",
" \n",
" # Create workflow\n",
" workflow = PraisonAIAgents(\n",
" agents=list(agents.values()),\n",
" tasks=tasks,\n",
" process=\"workflow\",\n",
" verbose=True\n",
" )\n",
" \n",
" # Execute research\n",
" results = await workflow.astart()\n",
" \n",
" return results\n",
"\n",
"# Example with custom parameters\n",
"custom_config = MarketResearchConfig(\n",
" company=\"Apple\",\n",
" geography=\"Global\", \n",
" industry=\"Consumer Electronics\",\n",
" sections=[\"market_overview\", \"competitive_analysis\", \"financial_performance\"]\n",
")\n",
"\n",
"# Run the research\n",
"print(\"🚀 Running Custom Market Research...\")\n",
"custom_results = await run_market_research(custom_config)\n",
"\n",
"print(\"\\n📋 Research Complete!\")\n",
"print(\"=\" * 60)"
]
},
{
"cell_type": "markdown",
"id": "12",
"metadata": {},
"source": [
"## Method 3: FastAPI Integration Example"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "13",
"metadata": {},
"outputs": [],
"source": [
"# Example of using the FastAPI endpoint\n",
"import requests\n",
"import json\n",
"import time\n",
"\n",
"# Note: This requires the FastAPI server to be running\n",
"# Run: uvicorn secondary-market-research-api:app --reload --port 8000\n",
"\n",
"API_BASE_URL = \"http://localhost:8000\" # Change if running elsewhere\n",
"\n",
"def generate_report_via_api(company, geography, industry, sections):\n",
" \"\"\"Generate market research report via API\"\"\"\n",
" \n",
" # Prepare request\n",
" request_data = {\n",
" \"company\": company,\n",
" \"geography\": geography,\n",
" \"industry\": industry,\n",
" \"sections\": sections,\n",
" \"format\": \"json\"\n",
" }\n",
" \n",
" print(f\"📤 Submitting research request for {company}...\")\n",
" \n",
" # Submit research request\n",
" response = requests.post(f\"{API_BASE_URL}/research/generate\", json=request_data)\n",
" \n",
" if response.status_code == 200:\n",
" job_data = response.json()\n",
" job_id = job_data[\"job_id\"]\n",
" print(f\"✅ Job submitted successfully. Job ID: {job_id}\")\n",
" \n",
" # Poll for completion\n",
" print(\"⏳ Waiting for completion...\")\n",
" while True:\n",
" status_response = requests.get(f\"{API_BASE_URL}/research/status/{job_id}\")\n",
" if status_response.status_code == 200:\n",
" status_data = status_response.json()\n",
" print(f\"📊 Progress: {status_data['progress']}% - {status_data['message']}\")\n",
" \n",
" if status_data[\"status\"] == \"completed\":\n",
" print(\"🎉 Research completed!\")\n",
" \n",
" # Download report\n",
" report_response = requests.get(f\"{API_BASE_URL}/research/reports/{job_id}\")\n",
" if report_response.status_code == 200:\n",
" return report_response.json()\n",
" else:\n",
" print(\"❌ Failed to download report\")\n",
" return None\n",
" \n",
" elif status_data[\"status\"] == \"failed\":\n",
" print(f\"❌ Research failed: {status_data.get('error', 'Unknown error')}\")\n",
" return None\n",
" \n",
" time.sleep(5) # Wait 5 seconds before checking again\n",
" else:\n",
" print(\"❌ Failed to check status\")\n",
" return None\n",
" else:\n",
" print(f\"❌ Failed to submit request: {response.status_code}\")\n",
" return None\n",
"\n",
"# Example API usage (uncomment to run with API server)\n",
"# report = generate_report_via_api(\n",
"# company=\"Microsoft\",\n",
"# geography=\"Global\",\n",
"# industry=\"Cloud Computing\",\n",
"# sections=[\"market_overview\", \"competitive_analysis\"]\n",
"# )\n",
"# \n",
"# if report:\n",
"# print(\"\\n📋 Generated Report:\")\n",
"# print(json.dumps(report, indent=2)[:1000] + \"...\")\n",
"\n",
"print(\"💡 API integration example ready. Start the FastAPI server to test.\")"
]
},
{
"cell_type": "markdown",
"id": "14",
"metadata": {},
"source": [
"## Customization Examples"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "15",
"metadata": {},
"outputs": [],
"source": [
"# Example configurations for different industries and regions\n",
"\n",
"research_examples = {\n",
" \"tech_startup\": {\n",
" \"company\": \"OpenAI\",\n",
" \"geography\": \"United States\",\n",
" \"industry\": \"Artificial Intelligence\",\n",
" \"sections\": [\"market_overview\", \"competitive_analysis\", \"growth_opportunities\"]\n",
" },\n",
" \"automotive\": {\n",
" \"company\": \"BMW\",\n",
" \"geography\": \"Europe\",\n",
" \"industry\": \"Luxury Automobiles\",\n",
" \"sections\": [\"market_overview\", \"competitive_analysis\", \"financial_performance\", \"risk_assessment\"]\n",
" },\n",
" \"healthcare\": {\n",
" \"company\": \"Pfizer\",\n",
" \"geography\": \"Global\",\n",
" \"industry\": \"Pharmaceuticals\",\n",
" \"sections\": [\"market_overview\", \"competitive_analysis\", \"growth_opportunities\", \"risk_assessment\"]\n",
" },\n",
" \"fintech\": {\n",
" \"company\": \"Square\",\n",
" \"geography\": \"North America\",\n",
" \"industry\": \"Financial Technology\",\n",
" \"sections\": [\"market_overview\", \"competitive_analysis\", \"financial_performance\", \"growth_opportunities\"]\n",
" },\n",
" \"retail\": {\n",
" \"company\": \"Alibaba\",\n",
" \"geography\": \"Asia Pacific\",\n",
" \"industry\": \"E-commerce\",\n",
" \"sections\": [\"market_overview\", \"competitive_analysis\", \"financial_performance\", \"growth_opportunities\", \"risk_assessment\"]\n",
" }\n",
"}\n",
"\n",
"print(\"📚 Research Configuration Examples:\")\n",
"print(\"=\" * 50)\n",
"\n",
"for example_name, config in research_examples.items():\n",
" print(f\"\\n🔍 {example_name.upper()}:\")\n",
" print(f\" Company: {config['company']}\")\n",
" print(f\" Geography: {config['geography']}\")\n",
" print(f\" Industry: {config['industry']}\")\n",
" print(f\" Sections: {', '.join(config['sections'])}\")\n",
"\n",
"print(\"\\n💡 Use any of these configurations as templates for your research!\")"
]
},
{
"cell_type": "markdown",
"id": "16",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"This notebook demonstrates three different ways to use the Secondary Market Research Agent system:\n",
"\n",
"1. **YAML Configuration**: Easy to customize and modify research parameters\n",
"2. **Python API**: Programmatic control with custom configurations\n",
"3. **FastAPI Integration**: Production-ready REST API for web applications\n",
"\n",
"### Key Features:\n",
"- **Customizable Research Sections**: Choose from market overview, competitive analysis, financial performance, growth opportunities, and risk assessment\n",
"- **Geographic Flexibility**: Research any geographic region\n",
"- **Industry Agnostic**: Works across different industries and sectors\n",
"- **Professional Reports**: Generate business-ready research reports\n",
"- **Scalable Architecture**: Multi-agent system that can be extended\n",
"\n",
"### Use Cases:\n",
"- Market entry analysis\n",
"- Competitive intelligence\n",
"- Investment research\n",
"- Strategic planning\n",
"- Due diligence\n",
"\n",
"The system provides a comprehensive foundation for secondary market research that can be customized and extended based on specific business needs."
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"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.8.5"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
@@ -0,0 +1,369 @@
framework: "crewai"
topic: "Secondary Market Research Analysis with Customizable Parameters"
# Configuration variables for customization
variables:
company: "Tesla"
geography: "North America"
industry: "Electric Vehicles"
research_focus: "comprehensive market analysis"
roles:
market_overview_specialist:
role: "Market Overview Specialist"
backstory: |
You are a seasoned market research analyst with 10+ years of experience
in analyzing market dynamics, trends, and growth patterns across various
industries. You excel at synthesizing complex market data into clear,
actionable insights for business decision-makers.
goal: |
Analyze the {industry} market in {geography} to provide comprehensive
market overview including size, trends, growth drivers, and future projections.
verbose: true
allow_delegation: false
llm:
model: "gpt-4o-mini"
temperature: 0.3
tools:
- "InternetSearchTool"
tasks:
market_overview_research:
description: |
Conduct comprehensive market overview research for {industry} in {geography}:
1. Market Size & Valuation:
- Current market size and historical growth
- Market valuation and revenue projections
- Geographic market distribution
2. Market Trends:
- Key growth trends and patterns
- Emerging market developments
- Technology adoption rates
3. Market Drivers:
- Primary growth drivers and catalysts
- Consumer behavior shifts
- Regulatory and policy impacts
4. Market Segments:
- Key market segments and sub-sectors
- Segment performance and potential
- Customer demographics and preferences
Research should include quantitative data, statistics, and cite reliable sources.
expected_output: |
A comprehensive market overview report containing:
- Executive summary of market status
- Detailed market size and growth analysis
- Key trends and drivers identification
- Market segmentation analysis
- Future market projections and opportunities
Include relevant charts, statistics, and data sources.
competitive_intelligence_analyst:
role: "Competitive Intelligence Analyst"
backstory: |
You are a competitive intelligence expert specializing in analyzing
competitive landscapes, market positioning, and strategic intelligence.
You have deep expertise in competitor analysis, market share assessment,
and competitive strategy evaluation.
goal: |
Analyze {company}'s competitive landscape in the {industry} market
within {geography} to identify key competitors, market positioning,
and competitive dynamics.
verbose: true
allow_delegation: false
llm:
model: "gpt-4o-mini"
temperature: 0.3
tools:
- "InternetSearchTool"
tasks:
competitive_analysis:
description: |
Conduct detailed competitive analysis for {company} in {industry}:
1. Competitor Identification:
- Identify top 5-7 direct competitors
- Include emerging and indirect competitors
- Analyze competitor profiles and backgrounds
2. Market Share Analysis:
- Market share distribution among competitors
- {company}'s market position and ranking
- Market concentration and fragmentation
3. Competitive Positioning:
- Value proposition comparison
- Product/service differentiation
- Pricing strategies and positioning
4. Competitor Strengths & Weaknesses:
- SWOT analysis for key competitors
- Competitive advantages and disadvantages
- Performance comparison metrics
5. Competitive Strategies:
- Go-to-market strategies
- Innovation and R&D approaches
- Partnership and acquisition strategies
expected_output: |
A comprehensive competitive intelligence report including:
- Competitive landscape overview
- Detailed competitor profiles (top 5-7)
- Market share analysis and positioning
- Competitive strengths/weaknesses matrix
- Strategic implications and recommendations
financial_performance_analyst:
role: "Financial Performance Analyst"
backstory: |
You are a financial analyst with expertise in corporate finance,
financial statement analysis, and industry benchmarking. You specialize
in analyzing financial performance metrics, trends, and comparative
analysis within specific industry contexts.
goal: |
Analyze {company}'s financial performance, key metrics, and compare
with industry benchmarks to assess financial health and performance trends.
verbose: true
allow_delegation: false
llm:
model: "gpt-4o-mini"
temperature: 0.3
tools:
- "InternetSearchTool"
tasks:
financial_analysis:
description: |
Conduct comprehensive financial performance analysis for {company}:
1. Revenue Analysis:
- Revenue trends over last 3-5 years
- Revenue growth rates and patterns
- Revenue by segment/geography
2. Profitability Metrics:
- Gross, operating, and net profit margins
- EBITDA and EBITDA margins
- Return on assets (ROA) and equity (ROE)
3. Financial Ratios:
- Liquidity ratios (current, quick)
- Leverage ratios (debt-to-equity, interest coverage)
- Efficiency ratios (asset turnover, inventory turnover)
4. Industry Benchmarking:
- Compare key metrics with industry averages
- Position against top competitors
- Identify performance gaps and advantages
5. Financial Health Assessment:
- Cash flow analysis
- Debt levels and structure
- Financial stability indicators
expected_output: |
A detailed financial performance report containing:
- Financial performance summary and trends
- Key financial metrics and ratios analysis
- Industry benchmark comparison
- Financial strengths and areas for improvement
- Financial health assessment and outlook
growth_opportunities_researcher:
role: "Growth Opportunities Researcher"
backstory: |
You are a strategic business analyst specializing in identifying growth
opportunities, market expansion strategies, and emerging business models.
You excel at spotting trends, gaps, and untapped potential in markets
and industries.
goal: |
Identify and analyze growth opportunities for {company} in {industry}
within {geography}, including market expansion, innovation, and
strategic development possibilities.
verbose: true
allow_delegation: false
llm:
model: "gpt-4o-mini"
temperature: 0.4
tools:
- "InternetSearchTool"
tasks:
growth_opportunities_analysis:
description: |
Identify comprehensive growth opportunities for {company}:
1. Market Expansion:
- Geographic expansion opportunities
- New customer segments and demographics
- Untapped market niches
2. Product/Service Innovation:
- New product development opportunities
- Service enhancement possibilities
- Technology integration opportunities
3. Strategic Partnerships:
- Potential partnership opportunities
- Joint venture possibilities
- Acquisition targets and synergies
4. Technology Advancement:
- Emerging technology adoption
- Digital transformation opportunities
- Innovation and R&D investments
5. Market Trends Alignment:
- Emerging industry trends to leverage
- Consumer behavior shifts to capitalize on
- Regulatory changes creating opportunities
expected_output: |
A strategic growth opportunities report including:
- Prioritized growth opportunity matrix
- Market expansion analysis and recommendations
- Innovation and development opportunities
- Strategic partnership possibilities
- Implementation roadmap and timeline
risk_assessment_specialist:
role: "Risk Assessment Specialist"
backstory: |
You are a risk management expert with extensive experience in identifying,
analyzing, and assessing business risks across various industries. You
specialize in market risks, competitive threats, regulatory challenges,
and operational risk assessment.
goal: |
Assess and analyze potential risks, challenges, and threats facing
{company} in the {industry} market within {geography}.
verbose: true
allow_delegation: false
llm:
model: "gpt-4o-mini"
temperature: 0.3
tools:
- "InternetSearchTool"
tasks:
risk_assessment_analysis:
description: |
Conduct comprehensive risk assessment for {company}:
1. Competitive Risks:
- New entrant threats
- Competitor aggressive strategies
- Market share erosion risks
2. Market Risks:
- Market volatility and cyclicality
- Demand fluctuation risks
- Economic downturn impacts
3. Regulatory Risks:
- Regulatory changes and compliance
- Policy shifts and government intervention
- Environmental and safety regulations
4. Operational Risks:
- Supply chain vulnerabilities
- Technology and cybersecurity risks
- Talent and human capital risks
5. Financial Risks:
- Currency and exchange rate risks
- Interest rate and inflation impacts
- Credit and liquidity risks
6. Technology Disruption:
- Emerging technology threats
- Digital transformation challenges
- Innovation obsolescence risks
expected_output: |
A comprehensive risk assessment report including:
- Risk identification and categorization
- Risk impact and probability analysis
- Risk mitigation strategies and recommendations
- Risk monitoring and management framework
- Priority risk action plan
research_report_synthesizer:
role: "Research Report Synthesizer"
backstory: |
You are an expert business report writer and analyst with extensive
experience in synthesizing complex research findings into clear,
actionable business intelligence reports. You excel at creating
comprehensive, well-structured reports for executive decision-making.
goal: |
Synthesize all research findings into a comprehensive, professional
secondary market research report for {company} in {industry} that
provides actionable insights for business decision-making.
verbose: true
allow_delegation: false
llm:
model: "gpt-4o-mini"
temperature: 0.3
tasks:
report_synthesis:
description: |
Create a comprehensive secondary market research report synthesizing
all research findings:
Report Structure:
1. Executive Summary
- Key findings overview
- Critical insights and implications
- Primary recommendations
2. Market Overview
- Market size, trends, and dynamics
- Growth drivers and opportunities
- Market segmentation analysis
3. Competitive Landscape
- Competitive positioning analysis
- Key competitor profiles
- Market share and dynamics
4. Financial Performance
- Financial health assessment
- Performance benchmarking
- Financial trends and projections
5. Growth Opportunities
- Strategic growth vectors
- Market expansion possibilities
- Innovation and development opportunities
6. Risk Assessment
- Key risk factors and mitigation
- Challenge identification
- Risk management recommendations
7. Conclusions and Recommendations
- Strategic recommendations
- Action priorities
- Implementation considerations
Use professional formatting with clear headings, bullet points, and data visualization suggestions.
expected_output: |
A complete, professionally formatted secondary market research report containing:
- Executive summary with key insights
- Detailed analysis across all research areas
- Data-driven findings and conclusions
- Strategic recommendations and action items
- Professional formatting suitable for executive presentation
The report should be 8-12 pages in length with clear structure and actionable intelligence.
context:
- "market_overview_research"
- "competitive_analysis"
- "financial_analysis"
- "growth_opportunities_analysis"
- "risk_assessment_analysis"
dependencies:
- market_overview_specialist
- competitive_intelligence_analyst
- financial_performance_analyst
- growth_opportunities_researcher
- risk_assessment_specialist
- research_report_synthesizer
@@ -0,0 +1,528 @@
"""
Secondary Market Research FastAPI Application
This FastAPI application provides a REST API for generating customized
secondary market research reports. Users can specify company, geography,
industry, and research sections to generate tailored reports.
Features:
- RESTful API endpoints for market research
- Customizable research parameters
- Async report generation
- JSON and PDF report formats
- Real-time progress tracking
- Report history and caching
Usage:
uvicorn secondary-market-research-api:app --reload --port 8000
Then visit: http://localhost:8000/docs for API documentation
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from fastapi.responses import JSONResponse, FileResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
import asyncio
import json
import uuid
from datetime import datetime
import os
from pathlib import Path
# Import our market research system
from praisonaiagents import Agent, Task, PraisonAIAgents, Tools
# Create FastAPI app
app = FastAPI(
title="Secondary Market Research API",
description="Generate customized secondary market research reports",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Create directories for storing reports
REPORTS_DIR = Path("generated_reports")
REPORTS_DIR.mkdir(exist_ok=True)
# In-memory storage for job status (in production, use Redis or database)
job_status = {}
# Pydantic models for request/response
class MarketResearchRequest(BaseModel):
"""Request model for market research generation"""
company: str = Field(..., description="Company name to research", example="Tesla")
geography: str = Field(..., description="Geographic region", example="North America")
industry: str = Field(..., description="Industry sector", example="Electric Vehicles")
sections: Optional[List[str]] = Field(
default=["market_overview", "competitive_analysis", "financial_performance", "growth_opportunities", "risk_assessment"],
description="Research sections to include",
example=["market_overview", "competitive_analysis", "financial_performance"]
)
format: Optional[str] = Field(default="json", description="Output format: json or pdf", example="json")
email: Optional[str] = Field(None, description="Email for notification when complete", example="[email protected]")
class MarketResearchResponse(BaseModel):
"""Response model for market research request"""
job_id: str = Field(..., description="Unique job identifier")
status: str = Field(..., description="Job status: pending, running, completed, failed")
message: str = Field(..., description="Status message")
estimated_completion: Optional[str] = Field(None, description="Estimated completion time")
class JobStatusResponse(BaseModel):
"""Response model for job status check"""
job_id: str
status: str
progress: int = Field(..., description="Progress percentage (0-100)")
message: str
created_at: str
updated_at: str
result_url: Optional[str] = None
error: Optional[str] = None
class MarketResearchConfig:
"""Configuration class for market research parameters"""
def __init__(self, company: str, geography: str, industry: str, sections: List[str]):
self.company = company
self.geography = geography
self.industry = industry
self.sections = sections
self.timestamp = datetime.now().isoformat()
def create_market_research_agents(config: MarketResearchConfig):
"""Create specialized market research agents"""
agents = {}
if "market_overview" in config.sections:
agents["market_overview"] = Agent(
name="Market Overview Specialist",
role="Market Analysis Expert",
goal=f"Analyze the {config.industry} market in {config.geography}",
instructions=f"""
Analyze market size, trends, growth drivers, and overall market dynamics for
{config.industry} in {config.geography}. Provide quantitative data and cite sources.
""",
tools=[Tools.internet_search],
verbose=False
)
if "competitive_analysis" in config.sections:
agents["competitive"] = Agent(
name="Competitive Intelligence Analyst",
role="Competition Research Expert",
goal=f"Analyze {config.company}'s competitive landscape",
instructions=f"""
Analyze {config.company}'s competitive position in {config.industry} within {config.geography}.
Identify key competitors, market share, and competitive advantages.
""",
tools=[Tools.internet_search],
verbose=False
)
if "financial_performance" in config.sections:
agents["financial"] = Agent(
name="Financial Performance Analyst",
role="Financial Research Expert",
goal=f"Analyze {config.company}'s financial performance",
instructions=f"""
Research {config.company}'s financial performance, revenue trends, profitability,
and compare with industry benchmarks.
""",
tools=[Tools.internet_search],
verbose=False
)
if "growth_opportunities" in config.sections:
agents["growth"] = Agent(
name="Growth Opportunities Researcher",
role="Strategic Growth Expert",
goal=f"Identify growth opportunities for {config.company}",
instructions=f"""
Identify emerging opportunities, market gaps, expansion possibilities for {config.company}
in {config.industry} within {config.geography}.
""",
tools=[Tools.internet_search],
verbose=False
)
if "risk_assessment" in config.sections:
agents["risk"] = Agent(
name="Risk Assessment Specialist",
role="Risk Analysis Expert",
goal=f"Assess risks for {config.company}",
instructions=f"""
Identify and analyze potential risks, challenges, and threats facing {config.company}
in {config.geography}. Consider regulatory, competitive, and market risks.
""",
tools=[Tools.internet_search],
verbose=False
)
# Always include synthesizer
agents["synthesizer"] = Agent(
name="Research Report Synthesizer",
role="Report Writing Expert",
goal="Synthesize research into comprehensive report",
instructions=f"""
Create a professional secondary market research report for {config.company}
in {config.industry} with clear structure, insights, and recommendations.
""",
verbose=False
)
return agents
def create_research_tasks(agents: Dict[str, Agent], config: MarketResearchConfig):
"""Create research tasks for the agents"""
tasks = []
# Create tasks based on available sections
if "market_overview" in config.sections and "market_overview" in agents:
market_task = Task(
name="market_overview_research",
description=f"Research {config.industry} market overview in {config.geography}",
expected_output="Market overview analysis with key metrics and trends",
agent=agents["market_overview"]
)
tasks.append(market_task)
if "competitive_analysis" in config.sections and "competitive" in agents:
competitive_task = Task(
name="competitive_analysis",
description=f"Analyze competitive landscape for {config.company}",
expected_output="Competitive analysis with key competitors and positioning",
agent=agents["competitive"]
)
tasks.append(competitive_task)
if "financial_performance" in config.sections and "financial" in agents:
financial_task = Task(
name="financial_analysis",
description=f"Analyze {config.company}'s financial performance",
expected_output="Financial performance analysis with key metrics",
agent=agents["financial"]
)
tasks.append(financial_task)
if "growth_opportunities" in config.sections and "growth" in agents:
growth_task = Task(
name="growth_opportunities",
description=f"Identify growth opportunities for {config.company}",
expected_output="Growth opportunities analysis with strategic recommendations",
agent=agents["growth"]
)
tasks.append(growth_task)
if "risk_assessment" in config.sections and "risk" in agents:
risk_task = Task(
name="risk_assessment",
description=f"Assess risks for {config.company}",
expected_output="Risk assessment with mitigation strategies",
agent=agents["risk"]
)
tasks.append(risk_task)
# Synthesis task
synthesis_task = Task(
name="synthesis",
description=f"Synthesize all findings into comprehensive report",
expected_output="Complete secondary market research report",
agent=agents["synthesizer"],
context=tasks
)
tasks.append(synthesis_task)
return tasks
async def generate_market_research_report(job_id: str, config: MarketResearchConfig):
"""Background task to generate market research report"""
try:
# Update job status
job_status[job_id]["status"] = "running"
job_status[job_id]["progress"] = 10
job_status[job_id]["message"] = "Creating research agents..."
job_status[job_id]["updated_at"] = datetime.now().isoformat()
# Create agents and tasks
agents = create_market_research_agents(config)
tasks = create_research_tasks(agents, config)
job_status[job_id]["progress"] = 20
job_status[job_id]["message"] = "Setting up research workflow..."
# Create workflow
workflow = PraisonAIAgents(
agents=list(agents.values()),
tasks=tasks,
process="workflow",
verbose=False
)
job_status[job_id]["progress"] = 30
job_status[job_id]["message"] = "Executing research workflow..."
# Execute research
results = await workflow.astart()
job_status[job_id]["progress"] = 90
job_status[job_id]["message"] = "Generating final report..."
# Prepare final report
report_data = {
"metadata": {
"job_id": job_id,
"company": config.company,
"geography": config.geography,
"industry": config.industry,
"sections": config.sections,
"generated_at": datetime.now().isoformat()
},
"executive_summary": "Executive summary would be generated here...",
"research_findings": {}
}
# Extract results from each task
if "task_results" in results:
for task_name, result in results["task_results"].items():
if result:
report_data["research_findings"][task_name] = {
"content": result.raw,
"agent": result.agent if hasattr(result, 'agent') else "Unknown"
}
# Save report to file
report_filename = f"market_research_{job_id}.json"
report_path = REPORTS_DIR / report_filename
with open(report_path, "w") as f:
json.dump(report_data, f, indent=2, default=str)
# Update job status to completed
job_status[job_id]["status"] = "completed"
job_status[job_id]["progress"] = 100
job_status[job_id]["message"] = "Report generation completed successfully"
job_status[job_id]["result_url"] = f"/reports/{job_id}"
job_status[job_id]["updated_at"] = datetime.now().isoformat()
except Exception as e:
# Update job status to failed
job_status[job_id]["status"] = "failed"
job_status[job_id]["error"] = str(e)
job_status[job_id]["message"] = f"Report generation failed: {str(e)}"
job_status[job_id]["updated_at"] = datetime.now().isoformat()
# API Endpoints
@app.get("/", summary="API Health Check")
async def root():
"""Health check endpoint"""
return {
"service": "Secondary Market Research API",
"status": "active",
"version": "1.0.0",
"timestamp": datetime.now().isoformat()
}
@app.post("/research/generate", response_model=MarketResearchResponse, summary="Generate Market Research Report")
async def generate_research(
request: MarketResearchRequest,
background_tasks: BackgroundTasks
):
"""
Generate a secondary market research report with customizable parameters.
This endpoint starts an async job to generate a comprehensive market research
report based on the provided parameters. The report generation runs in the
background and can be tracked using the returned job_id.
"""
# Generate unique job ID
job_id = str(uuid.uuid4())
# Initialize job status
job_status[job_id] = {
"job_id": job_id,
"status": "pending",
"progress": 0,
"message": "Report generation queued",
"created_at": datetime.now().isoformat(),
"updated_at": datetime.now().isoformat(),
"config": request.dict()
}
# Create config object
config = MarketResearchConfig(
company=request.company,
geography=request.geography,
industry=request.industry,
sections=request.sections
)
# Start background task
background_tasks.add_task(generate_market_research_report, job_id, config)
return MarketResearchResponse(
job_id=job_id,
status="pending",
message="Market research report generation started",
estimated_completion=datetime.now().isoformat()
)
@app.get("/research/status/{job_id}", response_model=JobStatusResponse, summary="Check Job Status")
async def check_job_status(job_id: str):
"""
Check the status of a market research report generation job.
Returns the current status, progress percentage, and any results or errors.
"""
if job_id not in job_status:
raise HTTPException(status_code=404, detail="Job ID not found")
return JobStatusResponse(**job_status[job_id])
@app.get("/research/reports/{job_id}", summary="Download Research Report")
async def download_report(job_id: str):
"""
Download the generated market research report.
Returns the complete report in JSON format once generation is complete.
"""
if job_id not in job_status:
raise HTTPException(status_code=404, detail="Job ID not found")
if job_status[job_id]["status"] != "completed":
raise HTTPException(status_code=400, detail="Report not ready. Check job status first.")
report_path = REPORTS_DIR / f"market_research_{job_id}.json"
if not report_path.exists():
raise HTTPException(status_code=404, detail="Report file not found")
return FileResponse(
path=report_path,
media_type="application/json",
filename=f"market_research_{job_id}.json"
)
@app.get("/research/jobs", summary="List All Jobs")
async def list_jobs():
"""
List all market research jobs and their current status.
Useful for monitoring and administrative purposes.
"""
return {
"total_jobs": len(job_status),
"jobs": [
{
"job_id": job_id,
"status": status["status"],
"progress": status["progress"],
"created_at": status["created_at"],
"company": status["config"]["company"] if "config" in status else "Unknown"
}
for job_id, status in job_status.items()
]
}
@app.delete("/research/jobs/{job_id}", summary="Cancel Job")
async def cancel_job(job_id: str):
"""
Cancel a pending or running market research job.
"""
if job_id not in job_status:
raise HTTPException(status_code=404, detail="Job ID not found")
if job_status[job_id]["status"] in ["completed", "failed"]:
raise HTTPException(status_code=400, detail="Cannot cancel completed or failed job")
job_status[job_id]["status"] = "cancelled"
job_status[job_id]["message"] = "Job cancelled by user"
job_status[job_id]["updated_at"] = datetime.now().isoformat()
return {"message": "Job cancelled successfully", "job_id": job_id}
@app.get("/research/templates", summary="Get Research Templates")
async def get_research_templates():
"""
Get predefined research templates for common use cases.
"""
templates = {
"technology_company": {
"sections": ["market_overview", "competitive_analysis", "financial_performance", "growth_opportunities"],
"example": {
"company": "Tesla",
"geography": "North America",
"industry": "Electric Vehicles"
}
},
"financial_services": {
"sections": ["market_overview", "competitive_analysis", "risk_assessment", "financial_performance"],
"example": {
"company": "Goldman Sachs",
"geography": "Global",
"industry": "Investment Banking"
}
},
"healthcare": {
"sections": ["market_overview", "competitive_analysis", "growth_opportunities", "risk_assessment"],
"example": {
"company": "Johnson & Johnson",
"geography": "North America",
"industry": "Pharmaceuticals"
}
},
"retail": {
"sections": ["market_overview", "competitive_analysis", "financial_performance", "growth_opportunities"],
"example": {
"company": "Amazon",
"geography": "Global",
"industry": "E-commerce"
}
}
}
return {
"templates": templates,
"available_sections": [
"market_overview",
"competitive_analysis",
"financial_performance",
"growth_opportunities",
"risk_assessment"
]
}
if __name__ == "__main__":
import uvicorn
print("🚀 Starting Secondary Market Research API...")
print("📚 API Documentation: http://localhost:8000/docs")
print("🔄 ReDoc Documentation: http://localhost:8000/redoc")
uvicorn.run(
"secondary-market-research-api:app",
host="0.0.0.0",
port=8000,
reload=True
)
@@ -0,0 +1,205 @@
# CSV Processing with PraisonAI Agents
This document answers GitHub Issue #23: "How to pass in a CSV file" and provides multiple solutions for processing CSV files with PraisonAI agents.
## Quick Answer
**Yes, it's absolutely possible!** PraisonAI provides multiple ways to process CSV files with agents working through the list sequentially.
## The Simplest Solution
For the exact use case mentioned in the issue (CSV list of URLs processed sequentially):
```python
from praisonaiagents import Agent, Task, PraisonAIAgents
# Create your agent
agent = Agent(
name="URLProcessor",
instructions="Analyze each URL from the CSV file"
)
# Create a loop task
task = Task(
description="Process each URL from the CSV",
agent=agent,
task_type="loop", # Enable CSV loop processing
input_file="your_urls.csv" # Your CSV file
)
# Run the agents
agents = PraisonAIAgents(
agents=[agent],
tasks=[task],
process="workflow"
)
agents.start()
```
## CSV File Format
Your CSV file should have a simple structure. For URLs:
```csv
url
https://example.com
https://github.com
https://stackoverflow.com
```
Or with additional columns:
```csv
url,description
https://example.com,Example website
https://github.com,GitHub platform
https://stackoverflow.com,Q&A site
```
## Available Methods
### Method 1: Simple Loop Processing (Recommended)
- **File**: `simple-csv-url-processor.py`
- **Best for**: Beginners, simple use cases
- **How it works**: Uses `task_type="loop"` to automatically process each CSV row
### Method 2: Manual CSV Processing
- **File**: `csv-processing-agents.py` (Method 2)
- **Best for**: When you need more control over validation and processing
- **How it works**: Uses CSV tools (`read_csv`, `write_csv`) for manual processing
### Method 3: URL-Specific Processing
- **File**: `csv-processing-agents.py` (Method 3)
- **Best for**: Specifically processing URLs with detailed analysis
- **How it works**: Combines CSV tools with URL analysis logic
### Method 4: Advanced Processing with Validation
- **File**: `csv-processing-agents.py` (Method 4)
- **Best for**: Production use cases requiring data validation
- **How it works**: Validates data before processing, handles errors gracefully
## Built-in CSV Tools
PraisonAI includes powerful CSV tools:
```python
from praisonaiagents.tools import read_csv, write_csv, merge_csv
# Read CSV files
data = read_csv("input.csv")
# Write CSV files
write_csv("output.csv", data)
# Merge multiple CSV files
merge_csv(["file1.csv", "file2.csv"], "merged.csv")
```
## Examples for Different Use Cases
### Processing URLs from CSV
```python
# Your CSV: urls.csv
# url,priority
# https://example.com,high
# https://github.com,medium
agent = Agent(
name="URLAnalyzer",
tools=[read_csv, write_csv],
instructions="Analyze URLs based on priority"
)
```
### Processing Any Text File
```python
# For text files, you can convert to CSV format first
# Or process line by line using file tools
```
### Batch Processing Tasks
```python
# Your CSV: tasks.csv
# task
# "Analyze competitor website"
# "Research market trends"
# "Generate report summary"
task = Task(
description="Complete each task from the CSV",
task_type="loop",
input_file="tasks.csv"
)
```
## Prerequisites
1. **Install PraisonAI Agents**:
```bash
pip install praisonaiagents
```
2. **Set API Key**:
```bash
export OPENAI_API_KEY=your_api_key_here
```
3. **Install pandas** (for advanced CSV operations):
```bash
pip install pandas
```
## Running the Examples
1. **Simple URL Processing**:
```bash
python simple-csv-url-processor.py
```
2. **Comprehensive Examples**:
```bash
python csv-processing-agents.py
```
## Troubleshooting
### Common Issues:
1. **"CSV file not found"**
- Make sure your CSV file is in the same directory as your script
- Use absolute paths if needed: `/full/path/to/your/file.csv`
2. **"pandas not available"**
- Install pandas: `pip install pandas`
3. **"API key not set"**
- Set your OpenAI API key: `export OPENAI_API_KEY=your_key`
4. **Processing stops early**
- Increase `max_iter` parameter in PraisonAIAgents
- Check for errors in your CSV format
### CSV Format Tips:
- Use UTF-8 encoding
- Ensure consistent column headers
- Handle special characters properly
- Keep URLs properly formatted (http:// or https://)
## Related Documentation
- [CSV Tools Documentation](../../../docs/tools/csv_tools.mdx)
- [Repetitive Agents](../../../docs/features/repetitive.mdx)
- [Agent Examples](../../README.md)
## Support
If you encounter issues:
1. Check the troubleshooting section above
2. Review the example files for working implementations
3. Open an issue on the GitHub repository with your specific use case
---
*This solution was generated in response to GitHub Issue #23*
@@ -0,0 +1,369 @@
"""
CSV Processing with PraisonAI Agents
This example demonstrates multiple ways to process CSV files with agents:
1. Simple CSV Loop Processing (easiest)
2. Manual CSV Processing with read_csv tools
3. URL Processing from CSV files
4. Custom CSV Processing with validation
Author: Generated for GitHub Issue #23
"""
import os
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents.tools import read_csv, write_csv
def create_sample_urls_csv():
"""Create a sample CSV file with URLs for testing"""
urls_data = [
{"id": 1, "url": "https://example.com", "description": "Example website"},
{"id": 2, "url": "https://github.com", "description": "GitHub platform"},
{"id": 3, "url": "https://stackoverflow.com", "description": "Stack Overflow Q&A"},
{"id": 4, "url": "https://python.org", "description": "Python official site"},
{"id": 5, "url": "https://docs.python.org", "description": "Python documentation"}
]
# Ensure the CSV file exists
write_csv("sample_urls.csv", urls_data)
print("✅ Created sample_urls.csv with 5 URLs")
return "sample_urls.csv"
def create_sample_tasks_csv():
"""Create a sample CSV file with tasks for loop processing"""
tasks_data = [
{"task": "Analyze the homepage of example.com and summarize its content"},
{"task": "Check if github.com is accessible and describe its main features"},
{"task": "Research what Stack Overflow is used for"},
{"task": "Summarize the key features of Python programming language"},
{"task": "Describe what you can find in Python documentation"}
]
write_csv("sample_tasks.csv", tasks_data)
print("✅ Created sample_tasks.csv with 5 tasks")
return "sample_tasks.csv"
# ==============================================================================
# METHOD 1: SIMPLE CSV LOOP PROCESSING (RECOMMENDED FOR BEGINNERS)
# ==============================================================================
def method_1_simple_loop():
"""
The easiest way to process CSV files with PraisonAI agents.
This method automatically loops through each row in your CSV file.
Each row becomes input for the agent to process.
"""
print("\n" + "="*60)
print("METHOD 1: Simple CSV Loop Processing")
print("="*60)
# Create sample CSV file
csv_file = create_sample_tasks_csv()
# Create an agent that will process each row
loop_agent = Agent(
name="CSVLoopProcessor",
role="CSV Task Processor",
goal="Process each task from the CSV file efficiently",
backstory="Expert at handling repetitive tasks from CSV data",
instructions="Process each task thoroughly and provide detailed output",
llm="gpt-4o-mini"
)
# Create a loop task that will automatically process each CSV row
loop_task = Task(
description="Process each task from the CSV file",
expected_output="Completed task with detailed results",
agent=loop_agent,
task_type="loop", # This enables CSV loop processing
input_file=csv_file # The CSV file to process
)
# Initialize the agents system
agents = PraisonAIAgents(
agents=[loop_agent],
tasks=[loop_task],
process="workflow",
max_iter=10 # Maximum iterations to prevent infinite loops
)
print(f"🚀 Starting loop processing of {csv_file}")
print("Each row will be processed as a separate task...")
# Start processing
agents.start()
# ==============================================================================
# METHOD 2: MANUAL CSV PROCESSING WITH TOOLS
# ==============================================================================
def method_2_manual_csv():
"""
Manual CSV processing using built-in CSV tools.
This gives you more control over how the CSV is processed,
including validation, filtering, and custom logic.
"""
print("\n" + "="*60)
print("METHOD 2: Manual CSV Processing with Tools")
print("="*60)
# Create sample CSV file
csv_file = create_sample_urls_csv()
# Create an agent with CSV tools
csv_agent = Agent(
name="CSVAnalyzer",
role="CSV Data Processor",
goal="Read, validate, and process CSV data manually",
backstory="Specialist in data validation and processing",
tools=[read_csv, write_csv],
instructions="""
You are an expert at processing CSV files. When given a CSV file:
1. Read the CSV file using read_csv tool
2. Validate the data structure
3. Process each row according to the requirements
4. Create a summary of the processing results
5. Save results to a new CSV file if needed
""",
llm="gpt-4o-mini"
)
# Create a task for manual CSV processing
csv_task = Task(
description=f"""
Process the CSV file '{csv_file}' manually:
1. Read the CSV file using the read_csv tool
2. Validate that it contains 'url' and 'description' columns
3. For each URL, describe what type of website it is
4. Create a summary report of all URLs processed
5. Save the results to 'processed_urls.csv'
""",
expected_output="A detailed report of CSV processing with validation results",
agent=csv_agent
)
# Initialize and run
agents = PraisonAIAgents(
agents=[csv_agent],
tasks=[csv_task],
process="sequential"
)
print(f"🚀 Starting manual processing of {csv_file}")
agents.start()
# ==============================================================================
# METHOD 3: URL PROCESSING FROM CSV (SPECIFIC TO ISSUE REQUEST)
# ==============================================================================
def method_3_url_processing():
"""
Specific example for processing URLs from CSV files.
This addresses the original question about processing a CSV list of URLs
where agents work through the list sequentially.
"""
print("\n" + "="*60)
print("METHOD 3: URL Processing from CSV Files")
print("="*60)
# Create sample CSV file with URLs
csv_file = create_sample_urls_csv()
# Create a URL processing agent
url_agent = Agent(
name="URLProcessor",
role="Website Analyzer",
goal="Analyze websites from CSV URL list",
backstory="Expert web analyst who can evaluate websites",
tools=[read_csv, write_csv],
instructions="""
You are a website analysis expert. When given a CSV with URLs:
1. Read the CSV file to get the list of URLs
2. For each URL, analyze what kind of website it is
3. Determine the purpose and main features of each site
4. Create a detailed analysis report
5. Save results with analysis data
""",
llm="gpt-4o-mini"
)
# Create the URL processing task
url_task = Task(
description=f"""
Process URLs from '{csv_file}':
1. Read the CSV file containing URLs and descriptions
2. For each URL in the list, provide an analysis of:
- What type of website it is
- Its main purpose and features
- Target audience
- Key characteristics
3. Work through the URLs sequentially (one by one)
4. Create a comprehensive report with all analyses
5. Save the results to 'url_analysis_results.csv'
Make sure to process each URL individually and thoroughly.
""",
expected_output="Detailed analysis of each URL with comprehensive results saved to CSV",
agent=url_agent
)
# Initialize and run
agents = PraisonAIAgents(
agents=[url_agent],
tasks=[url_task],
process="sequential"
)
print(f"🚀 Starting URL processing from {csv_file}")
print("URLs will be processed sequentially...")
agents.start()
# ==============================================================================
# METHOD 4: ADVANCED CSV PROCESSING WITH VALIDATION
# ==============================================================================
def method_4_advanced_processing():
"""
Advanced CSV processing with data validation and error handling.
This method shows how to handle complex CSV processing scenarios
with validation, error handling, and structured outputs.
"""
print("\n" + "="*60)
print("METHOD 4: Advanced CSV Processing with Validation")
print("="*60)
# Create a more complex CSV file
complex_data = [
{"id": 1, "url": "https://example.com", "priority": "high", "category": "test"},
{"id": 2, "url": "https://github.com", "priority": "medium", "category": "development"},
{"id": 3, "url": "invalid-url", "priority": "low", "category": "test"}, # Invalid URL for testing
{"id": 4, "url": "https://stackoverflow.com", "priority": "high", "category": "development"},
{"id": 5, "url": "", "priority": "medium", "category": "empty"} # Empty URL for testing
]
csv_file = "complex_urls.csv"
write_csv(csv_file, complex_data)
print(f"✅ Created {csv_file} with complex data including invalid entries")
# Create a validation agent
validator_agent = Agent(
name="DataValidator",
role="Data Quality Specialist",
goal="Validate and clean CSV data before processing",
backstory="Expert in data validation and quality assurance",
tools=[read_csv, write_csv],
instructions="""
You are a data validation expert. Your job is to:
1. Read CSV files and validate data quality
2. Identify invalid or problematic entries
3. Clean and standardize data where possible
4. Create separate files for valid and invalid data
5. Provide detailed validation reports
""",
llm="gpt-4o-mini"
)
# Create a processing agent
processor_agent = Agent(
name="ValidatedProcessor",
role="Validated Data Processor",
goal="Process only validated and clean data",
backstory="Specialist in processing pre-validated data efficiently",
tools=[read_csv, write_csv],
instructions="""
You process validated data efficiently:
1. Read validated CSV data
2. Process each valid entry according to requirements
3. Generate comprehensive results
4. Create detailed output reports
""",
llm="gpt-4o-mini"
)
# Create validation task
validation_task = Task(
description=f"""
Validate the data in '{csv_file}':
1. Read the CSV file
2. Check each row for:
- Valid URL format (must start with http:// or https://)
- Non-empty required fields
- Valid priority values (high, medium, low)
- Valid category values
3. Separate valid and invalid entries
4. Save valid entries to 'valid_urls.csv'
5. Save invalid entries to 'invalid_urls.csv' with error descriptions
6. Create a validation summary report
""",
expected_output="Validation complete with separate files for valid/invalid data and summary report",
agent=validator_agent
)
# Create processing task (depends on validation)
processing_task = Task(
description="""
Process the validated data from 'valid_urls.csv':
1. Read the validated CSV file
2. For each valid URL, provide analysis based on priority and category
3. High priority items get detailed analysis
4. Medium priority items get standard analysis
5. Low priority items get basic analysis
6. Save results to 'final_results.csv'
""",
expected_output="Processed results for all valid entries saved to final CSV",
agent=processor_agent
)
# Initialize and run with sequential processing
agents = PraisonAIAgents(
agents=[validator_agent, processor_agent],
tasks=[validation_task, processing_task],
process="sequential" # Validation must complete before processing
)
print(f"🚀 Starting advanced processing with validation")
agents.start()
# ==============================================================================
# MAIN EXECUTION
# ==============================================================================
if __name__ == "__main__":
print("CSV Processing with PraisonAI Agents - Multiple Methods Demo")
print("=" * 70)
print("\nThis demo shows 4 different ways to process CSV files with agents:")
print("1. Simple CSV Loop Processing (easiest)")
print("2. Manual CSV Processing with tools")
print("3. URL Processing from CSV (addresses the GitHub issue)")
print("4. Advanced processing with validation")
# You can run all methods or comment out ones you don't need
try:
# Method 1: Simple loop processing
method_1_simple_loop()
# Method 2: Manual CSV processing
method_2_manual_csv()
# Method 3: URL processing (specific to the GitHub issue)
method_3_url_processing()
# Method 4: Advanced processing with validation
method_4_advanced_processing()
print("\n" + "="*70)
print("✅ All CSV processing methods completed!")
print("Check the generated CSV files for results.")
print("="*70)
except Exception as e:
print(f"❌ Error during processing: {e}")
print("Make sure you have:")
print("- OpenAI API key set (export OPENAI_API_KEY=your_key)")
print("- praisonaiagents package installed (pip install praisonaiagents)")
print("- pandas package installed (pip install pandas)")
@@ -0,0 +1,77 @@
"""
Simple CSV URL Processor
This example directly addresses the GitHub issue #23:
"Is it possible to pass in a CSV list of URLs and have agents work through the list each in turn?"
This is the simplest possible implementation for processing URLs from a CSV file.
Author: Generated for GitHub Issue #23
"""
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents.tools import write_csv
def create_url_csv():
"""Create a CSV file with URLs to process"""
# Example URLs - replace with your own
urls = [
{"url": "https://example.com"},
{"url": "https://github.com"},
{"url": "https://stackoverflow.com"},
{"url": "https://python.org"}
]
write_csv("urls_to_process.csv", urls)
print("✅ Created urls_to_process.csv")
return "urls_to_process.csv"
def main():
"""
Simple CSV URL processing example
This shows the easiest way to process a CSV list of URLs
where agents work through the list sequentially.
"""
print("Simple CSV URL Processor")
print("=" * 40)
# Step 1: Create the CSV file (or use your existing one)
csv_file = create_url_csv()
# Step 2: Create an agent to process URLs
url_agent = Agent(
name="URLProcessor",
role="URL Analyzer",
goal="Analyze each URL from the CSV list",
backstory="Expert at analyzing websites and URLs",
instructions="Analyze each URL and provide insights about the website",
llm="gpt-4o-mini" # You can change this to any supported model
)
# Step 3: Create a task that will loop through the CSV
url_task = Task(
description="Analyze each URL from the CSV file and provide insights",
expected_output="Analysis of the website at each URL",
agent=url_agent,
task_type="loop", # This tells PraisonAI to loop through CSV rows
input_file=csv_file # Your CSV file with URLs
)
# Step 4: Run the agents
agents = PraisonAIAgents(
agents=[url_agent],
tasks=[url_task],
process="workflow",
max_iter=10 # Adjust based on how many URLs you have
)
print(f"🚀 Processing URLs from {csv_file}")
print("The agent will work through each URL in turn...")
agents.start()
print("✅ Finished processing all URLs!")
if __name__ == "__main__":
main()
+9
View File
@@ -0,0 +1,9 @@
from praisonaiagents import Agent, MCP
search_agent = Agent(
instructions="""You help search the web.""",
llm="gpt-4o-mini",
tools=MCP("http://localhost:8931/sse")
)
search_agent.start("Find about Praison AI")
+252
View File
@@ -0,0 +1,252 @@
import streamlit as st
from praisonaiagents import Agent
from praisonaiagents.mcp import MCP
import logging
import traceback
# Configure page
st.set_page_config(
page_title="AI Airbnb Assistant",
page_icon="🏠",
layout="wide"
)
st.title("🏠 AI Airbnb Assistant")
st.write("Enter your accommodation search query below to get started!")
# Add configuration options in sidebar
with st.sidebar:
st.header("⚙️ Configuration")
# LLM Model Selection
llm_model = st.selectbox(
"Choose LLM Model",
options=[
"gpt-4o-mini",
"gpt-4o",
"gpt-3.5-turbo",
"claude-3-5-sonnet-20241022",
"claude-3-5-haiku-20241022"
],
index=0,
help="Select the language model to use. Avoid provider/model format like 'ollama/llama3.2' in Streamlit."
)
# Debug Mode
debug_mode = st.checkbox(
"Enable Debug Mode",
value=False,
help="Enable detailed logging for troubleshooting"
)
# MCP Server Command
mcp_command = st.text_input(
"MCP Server Command",
value="npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt",
help="The command to start the MCP server"
)
# Initialize session state
if "agent_initialized" not in st.session_state:
st.session_state.agent_initialized = False
st.session_state.agent = None
st.session_state.init_error = None
# Function to initialize agent
def initialize_agent():
"""Initialize the agent with proper error handling."""
try:
with st.spinner("🔄 Initializing AI agent and MCP tools..."):
if debug_mode:
st.info("🐛 Debug mode enabled - check console for detailed logs")
# Create MCP tools
mcp_tools = MCP(
mcp_command,
timeout=60,
debug=debug_mode
)
# Create agent with session state management
agent = Agent(
instructions="""You are a helpful Airbnb assistant. Help users find and book apartments.
Use the available tools to search for accommodations, get property details, and provide helpful recommendations.
Always be polite and provide comprehensive information about properties.""",
llm=llm_model,
tools=mcp_tools,
verbose=debug_mode
)
return agent, None
except Exception as e:
error_msg = f"Failed to initialize agent: {str(e)}"
if debug_mode:
error_msg += f"\n\nFull traceback:\n{traceback.format_exc()}"
return None, error_msg
# Agent initialization section
if not st.session_state.agent_initialized:
if st.button("🚀 Initialize AI Assistant", type="primary"):
agent, error = initialize_agent()
if agent:
st.session_state.agent = agent
st.session_state.agent_initialized = True
st.session_state.init_error = None
st.success("✅ AI Assistant initialized successfully!")
st.rerun()
else:
st.session_state.init_error = error
st.error(f"❌ Initialization failed: {error}")
# Show initialization status
if st.session_state.init_error:
st.error(f"❌ Initialization Error: {st.session_state.init_error}")
with st.expander("🔧 Troubleshooting Tips"):
st.markdown("""
**Common issues and solutions:**
1. **MCP Server Issues:**
- Ensure Node.js and npm are installed
- Check if the MCP server command is correct
- Try running the MCP command manually first
2. **LLM Model Issues:**
- Make sure you have API keys set as environment variables
- Avoid provider/model formats (e.g., "ollama/llama3.2") in Streamlit
- Use standard model names (e.g., "gpt-4o-mini")
3. **Network/Timeout Issues:**
- Check your internet connection
- Increase timeout if needed
- Enable debug mode for more details
4. **Environment Issues:**
- Make sure all required packages are installed
- Check Python version compatibility
""")
# Main chat interface (only show if agent is initialized)
if st.session_state.agent_initialized and st.session_state.agent:
st.success("✅ AI Assistant is ready!")
# Create two columns for better layout
col1, col2 = st.columns([2, 1])
with col1:
# Query input
query = st.text_area(
"🔍 What are you looking for?",
placeholder="e.g., 'Find a 2-bedroom apartment in Paris for next weekend' or 'Show me luxury stays in Tokyo under $200/night'",
height=100
)
# Search button
search_col1, search_col2 = st.columns([1, 4])
with search_col1:
search_button = st.button("🔍 Search", type="primary", use_container_width=True)
with col2:
# Tips and examples
with st.container():
st.markdown("### 💡 Tips")
st.markdown("""
- Be specific about location and dates
- Mention budget preferences
- Include number of guests
- Ask about amenities
""")
with st.expander("📝 Example Queries"):
st.markdown("""
- "Find a beachfront apartment in Barcelona for 4 people"
- "Show me pet-friendly stays in London under £150/night"
- "I need a place with a kitchen near downtown Tokyo"
- "Find luxury accommodations in NYC for a business trip"
""")
# Handle search
if search_button and query:
if not query.strip():
st.warning("⚠️ Please enter a search query")
else:
try:
with st.spinner("🔍 Searching for accommodations..."):
# Add query to chat history
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
st.session_state.chat_history.append({"role": "user", "content": query})
# Get response from agent
result = st.session_state.agent.start(query)
st.session_state.chat_history.append({"role": "assistant", "content": result})
# Display result
st.markdown("### 🏠 Search Results")
st.markdown(result)
except Exception as e:
error_msg = f"Search failed: {str(e)}"
st.error(f"❌ {error_msg}")
if debug_mode:
st.error(f"Full traceback:\n{traceback.format_exc()}")
# Chat history section
if "chat_history" in st.session_state and st.session_state.chat_history:
st.markdown("---")
st.markdown("### 💬 Conversation History")
for i, message in enumerate(st.session_state.chat_history):
if message["role"] == "user":
st.markdown(f"**🙋 You:** {message['content']}")
else:
st.markdown(f"**🤖 Assistant:** {message['content']}")
if i < len(st.session_state.chat_history) - 1:
st.markdown("---")
# Clear history button
if st.button("🗑️ Clear History"):
st.session_state.chat_history = []
st.rerun()
else:
# Show instructions for first-time users
st.info("👆 Please initialize the AI Assistant first using the button above.")
with st.expander("ℹ️ About this Application"):
st.markdown("""
This Streamlit application demonstrates how to properly integrate MCP (Model Context Protocol)
tools with PraisonAI Agents in a Streamlit environment.
**Key Features:**
- ✅ Proper session state management for Streamlit
- ✅ Comprehensive error handling and user feedback
- ✅ Debug mode for troubleshooting
- ✅ Flexible LLM model selection
- ✅ Clean and intuitive user interface
**Technical Implementation:**
- Uses session state to prevent agent re-initialization
- Implements proper error handling for MCP setup
- Provides detailed troubleshooting guidance
- Demonstrates best practices for Streamlit + MCP integration
""")
# Footer
st.markdown("---")
st.markdown(
"""
<div style='text-align: center; color: gray;'>
Built with PraisonAI Agents + MCP •
<a href='https://github.com/MervinPraison/PraisonAI' target='_blank'>GitHub</a> •
<a href='https://docs.praisonai.com' target='_blank'>Documentation</a>
</div>
""",
unsafe_allow_html=True
)
@@ -0,0 +1,54 @@
import streamlit as st
from praisonaiagents import Agent
from praisonaiagents.mcp import MCP
st.title("AI Airbnb")
st.write("Enter your research query below to get started!")
# CRITICAL: Initialize agent in session state to prevent re-initialization
if "agent" not in st.session_state:
try:
with st.spinner("Initializing AI agent..."):
# Initialize agent only once and store in session state
st.session_state.agent = Agent(
instructions="You help book apartments on Airbnb.",
llm="gpt-4o-mini", # ✅ Use standard format instead of "ollama/llama3.2"
tools=MCP("npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt"),
verbose=True # Enable debugging
)
st.success("✅ Agent initialized successfully!")
except Exception as e:
st.error(f"❌ Failed to initialize agent: {str(e)}")
st.stop()
# Create the input field
query = st.text_input("Query")
# Add a search button
if st.button("Search"):
if query:
try:
with st.spinner("Researching..."):
# Use the agent from session state (already initialized)
result = st.session_state.agent.start(query)
st.write(result)
except Exception as e:
st.error(f"❌ Search failed: {str(e)}")
else:
st.warning("Please enter a query")
# Add troubleshooting information
with st.expander("🔧 Troubleshooting"):
st.markdown("""
**Key differences from problematic code:**
1. **Session State**: Agent is initialized once in `st.session_state.agent`
2. **LLM Format**: Using `"gpt-4o-mini"` instead of `"ollama/llama3.2"`
3. **Error Handling**: Proper try-catch blocks with user feedback
4. **Initialization Check**: Agent only created once, not on every interaction
**If you still have issues:**
- Make sure you have Node.js installed for the MCP server
- Set your OpenAI API key: `export OPENAI_API_KEY="your-key"`
- Try the debug mode by setting `verbose=True`
""")
@@ -0,0 +1,234 @@
# Secondary Market Research Agent System
A comprehensive multi-agent system for generating customized secondary market research reports with FastAPI integration.
## 🎯 Features
- **Multi-Agent Research System**: Specialized agents for different research areas
- **Customizable Parameters**: Company, geography, industry, and research sections
- **Comprehensive Reports**: Market overview, competitive analysis, financial performance, growth opportunities, and risk assessment
- **FastAPI Integration**: Production-ready REST API
- **Professional Output**: Business-ready reports suitable for decision-making
## 📁 Files
- `secondary-market-research.py` - Main Python implementation
- `secondary-market-research-api.py` - FastAPI web service
- `secondary_market_research_agents.yaml` - YAML configuration
- `secondary_market_research_agents.ipynb` - Jupyter notebook examples
## 🚀 Quick Start
### 1. Python Implementation
```python
from praisonaiagents import Agent, Task, PraisonAIAgents, Tools
import asyncio
# Configure research parameters
config = MarketResearchConfig(
company="Tesla",
geography="North America",
industry="Electric Vehicles",
sections=["market_overview", "competitive_analysis", "financial_performance"]
)
# Run research
results = await run_market_research(config)
```
### 2. FastAPI Web Service
```bash
# Start the API server
uvicorn secondary-market-research-api:app --reload --port 8000
# Visit API documentation
http://localhost:8000/docs
```
### 3. YAML Configuration
```yaml
framework: "crewai"
topic: "Secondary Market Research Analysis"
variables:
company: "Tesla"
geography: "North America"
industry: "Electric Vehicles"
roles:
market_overview_specialist:
role: "Market Overview Specialist"
# ... configuration details
```
## 📊 Research Sections
Choose from the following research sections:
- **market_overview**: Market size, trends, and growth drivers
- **competitive_analysis**: Competitor analysis and market positioning
- **financial_performance**: Financial metrics and benchmarking
- **growth_opportunities**: Strategic growth vectors and opportunities
- **risk_assessment**: Risk factors and mitigation strategies
## 🌍 Use Cases
### Technology Companies
```python
config = MarketResearchConfig(
company="OpenAI",
geography="United States",
industry="Artificial Intelligence",
sections=["market_overview", "competitive_analysis", "growth_opportunities"]
)
```
### Automotive Industry
```python
config = MarketResearchConfig(
company="BMW",
geography="Europe",
industry="Luxury Automobiles",
sections=["market_overview", "competitive_analysis", "financial_performance", "risk_assessment"]
)
```
### Healthcare/Pharmaceuticals
```python
config = MarketResearchConfig(
company="Pfizer",
geography="Global",
industry="Pharmaceuticals",
sections=["market_overview", "competitive_analysis", "growth_opportunities", "risk_assessment"]
)
```
## 🔧 API Endpoints
### Generate Research Report
```http
POST /research/generate
Content-Type: application/json
{
"company": "Tesla",
"geography": "North America",
"industry": "Electric Vehicles",
"sections": ["market_overview", "competitive_analysis"],
"format": "json"
}
```
### Check Job Status
```http
GET /research/status/{job_id}
```
### Download Report
```http
GET /research/reports/{job_id}
```
### Get Templates
```http
GET /research/templates
```
## 📈 Output Format
The system generates comprehensive reports with the following structure:
```json
{
"metadata": {
"job_id": "uuid",
"company": "Tesla",
"geography": "North America",
"industry": "Electric Vehicles",
"generated_at": "2024-01-01T00:00:00Z"
},
"executive_summary": "...",
"research_findings": {
"market_overview_research": {
"content": "Market analysis...",
"agent": "Market Overview Specialist"
},
"competitive_analysis": {
"content": "Competitive intelligence...",
"agent": "Competitive Intelligence Analyst"
}
}
}
```
## 🛠️ Installation
```bash
pip install praisonai[crewai]
pip install fastapi uvicorn
```
## ⚙️ Configuration
Set your API keys:
```bash
export OPENAI_API_KEY="your-api-key"
export OPENAI_MODEL_NAME="gpt-4o-mini"
```
## 📝 Customization
### Adding New Research Sections
1. Create a new agent in `create_market_research_agents()`
2. Add corresponding task in `create_research_tasks()`
3. Update the sections list in configuration
### Geographic Regions
Supported regions include:
- North America
- Europe
- Asia Pacific
- Global
- Custom regions
### Industries
The system works across industries:
- Technology
- Automotive
- Healthcare/Pharmaceuticals
- Financial Services
- Retail/E-commerce
- Energy
- Manufacturing
- And more...
## 🎓 Examples
See the Jupyter notebook `secondary_market_research_agents.ipynb` for detailed examples including:
- YAML-based configuration
- Python API usage
- FastAPI integration
- Multiple industry examples
- Customization patterns
## 🚀 Production Deployment
For production use:
1. Deploy FastAPI with proper ASGI server (Gunicorn + Uvicorn)
2. Use Redis or database for job status storage
3. Implement proper error handling and logging
4. Add authentication and rate limiting
5. Set up monitoring and alerting
## 📄 License
This project is part of PraisonAI and follows the same licensing terms.
@@ -0,0 +1,334 @@
"""
Secondary Market Research Agent System
This system creates a comprehensive market research solution with customizable
inputs for company, geography, and industry sections. It generates detailed
reports suitable for secondary market research.
Features:
- Multi-agent system with specialized research roles
- Customizable geography and company focus
- Structured report generation
- Market analysis and competitive intelligence
- FastAPI integration ready
"""
from praisonaiagents import Agent, Task, PraisonAIAgents, Tools
from typing import Dict, List, Any, Optional
import asyncio
import json
from datetime import datetime
class MarketResearchConfig:
"""Configuration class for market research parameters"""
def __init__(self,
company: str = "Tesla",
geography: str = "North America",
industry: str = "Electric Vehicles",
sections: List[str] = None):
self.company = company
self.geography = geography
self.industry = industry
self.sections = sections or [
"market_overview",
"competitive_analysis",
"financial_performance",
"growth_opportunities",
"risk_assessment"
]
self.timestamp = datetime.now().isoformat()
def create_market_research_agents(config: MarketResearchConfig):
"""Create specialized market research agents"""
# Market Overview Agent
market_overview_agent = Agent(
name="Market Overview Specialist",
role="Market Analysis Expert",
goal=f"Analyze the {config.industry} market in {config.geography}",
instructions=f"""
You are a market research specialist focused on {config.industry} in {config.geography}.
Analyze market size, trends, growth drivers, and overall market dynamics.
Focus on current market conditions and future projections.
Provide quantitative data where possible and cite reliable sources.
""",
tools=[Tools.internet_search],
verbose=True
)
# Competitive Intelligence Agent
competitive_agent = Agent(
name="Competitive Intelligence Analyst",
role="Competition Research Expert",
goal=f"Analyze {config.company}'s competitive landscape",
instructions=f"""
You are a competitive intelligence specialist analyzing {config.company}'s position
in the {config.industry} market within {config.geography}.
Identify key competitors, market share, competitive advantages, and positioning.
Analyze competitor strategies, strengths, weaknesses, and market positioning.
""",
tools=[Tools.internet_search],
verbose=True
)
# Financial Performance Agent
financial_agent = Agent(
name="Financial Performance Analyst",
role="Financial Research Expert",
goal=f"Analyze {config.company}'s financial performance and metrics",
instructions=f"""
You are a financial analyst specializing in {config.industry} companies.
Research {config.company}'s financial performance, revenue trends, profitability,
key financial metrics, and compare with industry benchmarks.
Focus on recent financial data and performance indicators.
""",
tools=[Tools.internet_search],
verbose=True
)
# Growth Opportunities Agent
growth_agent = Agent(
name="Growth Opportunities Researcher",
role="Strategic Growth Expert",
goal=f"Identify growth opportunities for {config.company}",
instructions=f"""
You are a strategic growth analyst focusing on {config.industry} in {config.geography}.
Identify emerging opportunities, market gaps, expansion possibilities,
new product/service opportunities, and strategic growth vectors for {config.company}.
Consider technological trends, regulatory changes, and market evolution.
""",
tools=[Tools.internet_search],
verbose=True
)
# Risk Assessment Agent
risk_agent = Agent(
name="Risk Assessment Specialist",
role="Risk Analysis Expert",
goal=f"Assess risks and challenges for {config.company}",
instructions=f"""
You are a risk assessment specialist for {config.industry} companies.
Identify and analyze potential risks, challenges, and threats facing {config.company}
in {config.geography}. Consider regulatory, competitive, technological,
economic, and operational risks.
""",
tools=[Tools.internet_search],
verbose=True
)
# Report Synthesizer Agent
synthesizer_agent = Agent(
name="Research Report Synthesizer",
role="Report Writing Expert",
goal="Synthesize research into comprehensive market research report",
instructions=f"""
You are an expert report writer specializing in market research reports.
Synthesize all research findings into a comprehensive, well-structured
secondary market research report for {config.company} in {config.industry}.
Create a professional report with:
- Executive Summary
- Market Overview
- Competitive Analysis
- Financial Performance Analysis
- Growth Opportunities
- Risk Assessment
- Conclusions and Recommendations
Use clear headings, bullet points, and structured formatting.
Include key insights and actionable recommendations.
""",
verbose=True
)
return {
"market_overview": market_overview_agent,
"competitive": competitive_agent,
"financial": financial_agent,
"growth": growth_agent,
"risk": risk_agent,
"synthesizer": synthesizer_agent
}
def create_research_tasks(agents: Dict[str, Agent], config: MarketResearchConfig):
"""Create research tasks for the agents"""
tasks = []
# Market Overview Task
if "market_overview" in config.sections:
market_task = Task(
name="market_overview_research",
description=f"""
Conduct comprehensive market overview research for {config.industry}
in {config.geography}. Include:
- Market size and valuation
- Growth trends and projections
- Key market drivers
- Market segments
- Regulatory environment
""",
expected_output="Detailed market overview analysis with quantitative data",
agent=agents["market_overview"],
is_start=True,
next_tasks=["competitive_analysis"] if "competitive_analysis" in config.sections else ["synthesis"]
)
tasks.append(market_task)
# Competitive Analysis Task
if "competitive_analysis" in config.sections:
competitive_task = Task(
name="competitive_analysis",
description=f"""
Analyze competitive landscape for {config.company} in {config.industry}:
- Identify top 5-7 competitors
- Market share analysis
- Competitive positioning
- Competitor strengths and weaknesses
- Competitive strategies
""",
expected_output="Comprehensive competitive intelligence report",
agent=agents["competitive"],
next_tasks=["financial_analysis"] if "financial_performance" in config.sections else ["synthesis"]
)
tasks.append(competitive_task)
# Financial Performance Task
if "financial_performance" in config.sections:
financial_task = Task(
name="financial_analysis",
description=f"""
Analyze {config.company}'s financial performance:
- Revenue trends (last 3-5 years)
- Profitability metrics
- Key financial ratios
- Industry benchmark comparison
- Financial health assessment
""",
expected_output="Detailed financial performance analysis",
agent=agents["financial"],
next_tasks=["growth_opportunities"] if "growth_opportunities" in config.sections else ["synthesis"]
)
tasks.append(financial_task)
# Growth Opportunities Task
if "growth_opportunities" in config.sections:
growth_task = Task(
name="growth_opportunities",
description=f"""
Identify growth opportunities for {config.company}:
- Market expansion opportunities
- New product/service opportunities
- Strategic partnerships potential
- Technology advancement opportunities
- Emerging market trends
""",
expected_output="Strategic growth opportunities analysis",
agent=agents["growth"],
next_tasks=["risk_assessment"] if "risk_assessment" in config.sections else ["synthesis"]
)
tasks.append(growth_task)
# Risk Assessment Task
if "risk_assessment" in config.sections:
risk_task = Task(
name="risk_assessment",
description=f"""
Assess risks and challenges for {config.company}:
- Competitive threats
- Regulatory risks
- Market risks
- Operational challenges
- Technology disruption risks
""",
expected_output="Comprehensive risk assessment report",
agent=agents["risk"],
next_tasks=["synthesis"]
)
tasks.append(risk_task)
# Synthesis Task
synthesis_task = Task(
name="synthesis",
description=f"""
Synthesize all research findings into a comprehensive secondary market
research report for {config.company} in {config.industry} ({config.geography}).
Create a professional report with clear sections, insights, and recommendations.
Focus on actionable intelligence for business decision-making.
""",
expected_output="Complete secondary market research report",
agent=agents["synthesizer"],
context=tasks # Use all previous tasks as context
)
tasks.append(synthesis_task)
return tasks
async def run_market_research(config: MarketResearchConfig) -> Dict[str, Any]:
"""Run the complete market research workflow"""
print(f"\n🔍 Starting Secondary Market Research for {config.company}")
print(f"📍 Geography: {config.geography}")
print(f"🏭 Industry: {config.industry}")
print(f"📊 Sections: {', '.join(config.sections)}")
print("=" * 60)
# Create agents and tasks
agents = create_market_research_agents(config)
tasks = create_research_tasks(agents, config)
# Create workflow
workflow = PraisonAIAgents(
agents=list(agents.values()),
tasks=tasks,
process="workflow",
verbose=True
)
# Execute research
results = await workflow.astart()
return {
"config": config.__dict__,
"results": results,
"timestamp": datetime.now().isoformat()
}
# Example usage
async def main():
"""Main function demonstrating the market research system"""
# Example 1: Tesla Electric Vehicle Research
tesla_config = MarketResearchConfig(
company="Tesla",
geography="North America",
industry="Electric Vehicles",
sections=["market_overview", "competitive_analysis", "financial_performance", "growth_opportunities"]
)
tesla_results = await run_market_research(tesla_config)
print("\n📋 Research Complete!")
print("=" * 60)
# Display results
if "task_results" in tesla_results["results"]:
for task_name, result in tesla_results["results"]["task_results"].items():
if result:
print(f"\n📌 {task_name.upper()}")
print("-" * 40)
print(result.raw[:500] + "..." if len(result.raw) > 500 else result.raw)
return tesla_results
if __name__ == "__main__":
# Run the market research
results = asyncio.run(main())
# Save results to file
with open("market_research_results.json", "w") as f:
json.dump(results, f, indent=2, default=str)
print(f"\n💾 Results saved to market_research_results.json")
@@ -864,7 +864,11 @@ Your Goal: {self.goal}
if self._using_custom_llm:
try:
# Special handling for MCP tools when using provider/model format
tool_param = self.tools if tools is None else tools
# Fix: Handle empty tools list properly - use self.tools if tools is None or empty
if tools is None or (isinstance(tools, list) and len(tools) == 0):
tool_param = self.tools
else:
tool_param = tools
# Convert MCP tool objects to OpenAI format if needed
if tool_param is not None:
+77 -51
View File
@@ -205,6 +205,58 @@ class LLM:
}
logging.debug(f"LLM instance initialized with: {json.dumps(debug_info, indent=2, default=str)}")
def _is_ollama_provider(self) -> bool:
"""Detect if this is an Ollama provider regardless of naming convention"""
if not self.model:
return False
# Direct ollama/ prefix
if self.model.startswith("ollama/"):
return True
# Check environment variables for Ollama base URL
base_url = os.getenv("OPENAI_BASE_URL", "")
api_base = os.getenv("OPENAI_API_BASE", "")
# Common Ollama endpoints
ollama_endpoints = ["localhost:11434", "127.0.0.1:11434", ":11434"]
return any(endpoint in base_url or endpoint in api_base for endpoint in ollama_endpoints)
def _parse_tool_call_arguments(self, tool_call: Dict, is_ollama: bool = False) -> tuple:
"""
Safely parse tool call arguments with proper error handling
Returns:
tuple: (function_name, arguments, tool_call_id)
"""
try:
if is_ollama:
# Special handling for Ollama provider which may have different structure
if "function" in tool_call and isinstance(tool_call["function"], dict):
function_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
else:
# Try alternative format that Ollama might return
function_name = tool_call.get("name", "unknown_function")
arguments_str = tool_call.get("arguments", "{}")
arguments = json.loads(arguments_str) if arguments_str else {}
tool_call_id = tool_call.get("id", f"tool_{id(tool_call)}")
else:
# Standard format for other providers with error handling
function_name = tool_call["function"]["name"]
arguments_str = tool_call["function"]["arguments"]
arguments = json.loads(arguments_str) if arguments_str else {}
tool_call_id = tool_call["id"]
except (KeyError, json.JSONDecodeError, TypeError) as e:
logging.error(f"Error parsing tool call arguments: {e}")
function_name = tool_call.get("name", "unknown_function")
arguments = {}
tool_call_id = tool_call.get("id", f"tool_{id(tool_call)}")
return function_name, arguments, tool_call_id
def _needs_system_message_skip(self) -> bool:
"""Check if this model requires skipping system messages"""
if not self.model:
@@ -486,32 +538,19 @@ class LLM:
for tool_call in tool_calls:
# Handle both object and dict access patterns
if isinstance(tool_call, dict):
# Special handling for Ollama provider which may have a different structure
if self.model and self.model.startswith("ollama/"):
try:
# Try standard format first
if "function" in tool_call and isinstance(tool_call["function"], dict):
function_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
else:
# Try alternative format that Ollama might return
function_name = tool_call.get("name", "unknown_function")
arguments = json.loads(tool_call.get("arguments", "{}"))
tool_call_id = tool_call.get("id", f"tool_{id(tool_call)}")
except Exception as e:
logging.error(f"Error processing Ollama tool call: {e}")
function_name = "unknown_function"
arguments = {}
tool_call_id = f"tool_{id(tool_call)}"
else:
# Standard format for other providers
function_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
tool_call_id = tool_call["id"]
is_ollama = self._is_ollama_provider()
function_name, arguments, tool_call_id = self._parse_tool_call_arguments(tool_call, is_ollama)
else:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
tool_call_id = tool_call.id
# Handle object-style tool calls
try:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
tool_call_id = tool_call.id
except (json.JSONDecodeError, AttributeError) as e:
logging.error(f"Error parsing object-style tool call: {e}")
function_name = "unknown_function"
arguments = {}
tool_call_id = f"tool_{id(tool_call)}"
logging.debug(f"[TOOL_EXEC_DEBUG] About to execute tool {function_name} with args: {arguments}")
tool_result = execute_tool_fn(function_name, arguments)
@@ -1083,32 +1122,19 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
for tool_call in tool_calls:
# Handle both object and dict access patterns
if isinstance(tool_call, dict):
# Special handling for Ollama provider which may have a different structure
if self.model and self.model.startswith("ollama/"):
try:
# Try standard format first
if "function" in tool_call and isinstance(tool_call["function"], dict):
function_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
else:
# Try alternative format that Ollama might return
function_name = tool_call.get("name", "unknown_function")
arguments = json.loads(tool_call.get("arguments", "{}"))
tool_call_id = tool_call.get("id", f"tool_{id(tool_call)}")
except Exception as e:
logging.error(f"Error processing Ollama tool call: {e}")
function_name = "unknown_function"
arguments = {}
tool_call_id = f"tool_{id(tool_call)}"
else:
# Standard format for other providers
function_name = tool_call["function"]["name"]
arguments = json.loads(tool_call["function"]["arguments"])
tool_call_id = tool_call["id"]
is_ollama = self._is_ollama_provider()
function_name, arguments, tool_call_id = self._parse_tool_call_arguments(tool_call, is_ollama)
else:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
tool_call_id = tool_call.id
# Handle object-style tool calls
try:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
tool_call_id = tool_call.id
except (json.JSONDecodeError, AttributeError) as e:
logging.error(f"Error parsing object-style tool call: {e}")
function_name = "unknown_function"
arguments = {}
tool_call_id = f"tool_{id(tool_call)}"
tool_result = await execute_tool_fn(function_name, arguments)
@@ -1129,7 +1155,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
response_text = ""
# Special handling for Ollama models that don't automatically process tool results
if self.model and self.model.startswith("ollama/") and tool_result:
if self._is_ollama_provider() and tool_result:
# For Ollama models, we need to explicitly ask the model to process the tool results
# First, check if the response is just a JSON tool call
try:
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.0.92"
version = "0.0.93"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
requires-python = ">=3.10"
authors = [
Binary file not shown.
+1 -1
View File
@@ -2233,7 +2233,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.92"
version = "0.0.93"
source = { editable = "." }
dependencies = [
{ name = "mcp" },
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.2.20.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.2.20.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.2.22.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.2.22.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+82 -3
View File
@@ -5,6 +5,7 @@ import argparse
from .version import __version__
import yaml
import os
import time
from rich import print
from dotenv import load_dotenv
load_dotenv()
@@ -25,6 +26,7 @@ CALL_MODULE_AVAILABLE = False
CREWAI_AVAILABLE = False
AUTOGEN_AVAILABLE = False
PRAISONAI_AVAILABLE = False
TRAIN_AVAILABLE = False
try:
# Create necessary directories and set CHAINLIT_APP_ROOT
if "CHAINLIT_APP_ROOT" not in os.environ:
@@ -71,6 +73,12 @@ try:
except ImportError:
pass
try:
from unsloth import FastLanguageModel
TRAIN_AVAILABLE = True
except ImportError:
pass
logging.basicConfig(level=os.environ.get('LOGLEVEL', 'INFO'), format='%(asctime)s - %(levelname)s - %(message)s')
logging.getLogger('alembic').setLevel(logging.ERROR)
logging.getLogger('gradio').setLevel(logging.ERROR)
@@ -217,9 +225,66 @@ class PraisonAI:
# If no command or direct_prompt, preserve agent_file from constructor (don't overwrite)
if args.deploy:
from .deploy import CloudDeployer
deployer = CloudDeployer()
deployer.run_commands()
if args.schedule or args.schedule_config:
# Scheduled deployment
from .scheduler import create_scheduler
# Load configuration from file if provided
config = {"max_retries": args.max_retries}
schedule_expr = args.schedule
provider = args.provider
if args.schedule_config:
try:
with open(args.schedule_config, 'r') as f:
file_config = yaml.safe_load(f)
# Extract deployment config
deploy_config = file_config.get('deployment', {})
schedule_expr = schedule_expr or deploy_config.get('schedule')
provider = deploy_config.get('provider', provider)
config['max_retries'] = deploy_config.get('max_retries', config['max_retries'])
# Apply environment variables if specified
env_vars = file_config.get('environment', {})
for key, value in env_vars.items():
os.environ[key] = str(value)
except FileNotFoundError:
print(f"Configuration file not found: {args.schedule_config}")
sys.exit(1)
except yaml.YAMLError as e:
print(f"Error parsing configuration file: {e}")
sys.exit(1)
if not schedule_expr:
print("Error: Schedule expression required. Use --schedule or specify in config file.")
sys.exit(1)
scheduler = create_scheduler(provider=provider, config=config)
print(f"Starting scheduled deployment with schedule: {schedule_expr}")
print(f"Provider: {provider}")
print(f"Max retries: {config['max_retries']}")
print("Press Ctrl+C to stop the scheduler")
if scheduler.start(schedule_expr, config['max_retries']):
try:
# Keep the main thread alive
while scheduler.is_running:
time.sleep(1)
except KeyboardInterrupt:
print("\nStopping scheduler...")
scheduler.stop()
print("Scheduler stopped successfully")
else:
print("Failed to start scheduler")
sys.exit(1)
else:
# One-time deployment (backward compatible)
from .deploy import CloudDeployer
deployer = CloudDeployer()
deployer.run_commands()
return
if getattr(args, 'chat', False):
@@ -242,6 +307,11 @@ class PraisonAI:
return
if args.command == 'train':
if not TRAIN_AVAILABLE:
print("[red]ERROR: Training dependencies not installed. Install with:[/red]")
print("\npip install \"praisonai[train]\"")
print("Or run: praisonai train init\n")
sys.exit(1)
package_root = os.path.dirname(os.path.abspath(__file__))
config_yaml_destination = os.path.join(os.getcwd(), 'config.yaml')
@@ -416,6 +486,10 @@ class PraisonAI:
parser.add_argument("--init", nargs=argparse.REMAINDER, help="Initialize agents with optional topic")
parser.add_argument("command", nargs="?", help="Command to run or direct prompt")
parser.add_argument("--deploy", action="store_true", help="Deploy the application")
parser.add_argument("--schedule", type=str, help="Schedule deployment (e.g., 'daily', 'hourly', '*/6h', '3600')")
parser.add_argument("--schedule-config", type=str, help="Path to scheduling configuration file")
parser.add_argument("--provider", type=str, default="gcp", help="Deployment provider (gcp, aws, azure)")
parser.add_argument("--max-retries", type=int, default=3, help="Maximum retry attempts for scheduled deployments")
parser.add_argument("--model", type=str, help="Model name")
parser.add_argument("--llm", type=str, help="LLM model to use for direct prompts")
parser.add_argument("--hf", type=str, help="Hugging Face model name")
@@ -509,6 +583,11 @@ class PraisonAI:
sys.exit(0)
elif args.command == 'train':
if not TRAIN_AVAILABLE:
print("[red]ERROR: Training dependencies not installed. Install with:[/red]")
print("\npip install \"praisonai[train]\"")
print("Or run: praisonai train init\n")
sys.exit(1)
package_root = os.path.dirname(os.path.abspath(__file__))
config_yaml_destination = os.path.join(os.getcwd(), 'config.yaml')
+1 -1
View File
@@ -56,7 +56,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.2.20 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.2.22 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+215
View File
@@ -0,0 +1,215 @@
import re
import threading
import time
import logging
from datetime import datetime, timedelta
from typing import Union, Optional, Callable, Dict, Any
from abc import ABC, abstractmethod
logger = logging.getLogger(__name__)
class DeployerInterface(ABC):
"""Abstract interface for deployers to ensure provider compatibility."""
@abstractmethod
def deploy(self) -> bool:
"""Execute deployment. Returns True on success, False on failure."""
pass
class CloudDeployerAdapter(DeployerInterface):
"""Adapter for existing CloudDeployer to match interface."""
def __init__(self):
from .deploy import CloudDeployer
self._deployer = CloudDeployer()
def deploy(self) -> bool:
"""Execute deployment using CloudDeployer."""
try:
self._deployer.run_commands()
return True
except Exception as e:
logger.error(f"Deployment failed: {e}")
return False
class ScheduleParser:
"""Parse schedule expressions into intervals."""
@staticmethod
def parse(schedule_expr: str) -> int:
"""
Parse schedule expression and return interval in seconds.
Supported formats:
- "daily" -> 86400 seconds
- "hourly" -> 3600 seconds
- "*/30m" -> 1800 seconds (every 30 minutes)
- "*/6h" -> 21600 seconds (every 6 hours)
- "60" -> 60 seconds (plain number)
"""
schedule_expr = schedule_expr.strip().lower()
if schedule_expr == "daily":
return 86400
elif schedule_expr == "hourly":
return 3600
elif schedule_expr.isdigit():
return int(schedule_expr)
elif schedule_expr.startswith("*/"):
# Handle */30m, */6h patterns
interval_part = schedule_expr[2:]
if interval_part.endswith("m"):
minutes = int(interval_part[:-1])
return minutes * 60
elif interval_part.endswith("h"):
hours = int(interval_part[:-1])
return hours * 3600
elif interval_part.endswith("s"):
return int(interval_part[:-1])
else:
return int(interval_part)
else:
raise ValueError(f"Unsupported schedule format: {schedule_expr}")
class DeploymentScheduler:
"""
Minimal deployment scheduler with provider-agnostic design.
Features:
- Simple interval-based scheduling
- Thread-safe operation
- Extensible deployer factory pattern
- Minimal dependencies
"""
def __init__(self, schedule_config: Optional[Dict[str, Any]] = None):
self.config = schedule_config or {}
self.is_running = False
self._stop_event = threading.Event()
self._thread = None
self._deployer = None
def set_deployer(self, deployer: DeployerInterface):
"""Set custom deployer implementation."""
self._deployer = deployer
def _get_deployer(self) -> DeployerInterface:
"""Get deployer instance using factory pattern."""
if self._deployer:
return self._deployer
# Default to CloudDeployer for backward compatibility
return CloudDeployerAdapter()
def start(self, schedule_expr: str, max_retries: int = 3) -> bool:
"""
Start scheduled deployment.
Args:
schedule_expr: Schedule expression (e.g., "daily", "*/6h", "3600")
max_retries: Maximum retry attempts on failure
Returns:
True if scheduler started successfully
"""
if self.is_running:
logger.warning("Scheduler is already running")
return False
try:
interval = ScheduleParser.parse(schedule_expr)
self.is_running = True
self._stop_event.clear()
self._thread = threading.Thread(
target=self._run_schedule,
args=(interval, max_retries),
daemon=True
)
self._thread.start()
logger.info(f"Deployment scheduler started with {interval}s interval")
return True
except Exception as e:
logger.error(f"Failed to start scheduler: {e}")
self.is_running = False
return False
def stop(self) -> bool:
"""Stop the scheduler."""
if not self.is_running:
return True
self._stop_event.set()
if self._thread and self._thread.is_alive():
self._thread.join(timeout=5)
self.is_running = False
logger.info("Deployment scheduler stopped")
return True
def _run_schedule(self, interval: int, max_retries: int):
"""Internal method to run scheduled deployments."""
deployer = self._get_deployer()
while not self._stop_event.is_set():
logger.info("Starting scheduled deployment")
success = False
for attempt in range(max_retries):
try:
if deployer.deploy():
logger.info(f"Deployment successful on attempt {attempt + 1}")
success = True
break
else:
logger.warning(f"Deployment failed on attempt {attempt + 1}")
except Exception as e:
logger.error(f"Deployment error on attempt {attempt + 1}: {e}")
if attempt < max_retries - 1:
time.sleep(30) # Wait before retry
if not success:
logger.error(f"Deployment failed after {max_retries} attempts")
# Wait for next scheduled time
self._stop_event.wait(interval)
def deploy_once(self) -> bool:
"""Execute a single deployment immediately."""
deployer = self._get_deployer()
try:
return deployer.deploy()
except Exception as e:
logger.error(f"One-time deployment failed: {e}")
return False
def create_scheduler(provider: str = "gcp", config: Optional[Dict[str, Any]] = None) -> DeploymentScheduler:
"""
Factory function to create scheduler for different providers.
Args:
provider: Deployment provider ("gcp", "aws", "azure", etc.)
config: Optional configuration dict
Returns:
Configured DeploymentScheduler instance
"""
scheduler = DeploymentScheduler(config)
# Provider-specific deployer setup can be added here
if provider == "gcp":
# Default CloudDeployer for GCP
pass
elif provider == "aws":
# Future: AWS deployer implementation
logger.warning("AWS provider not yet implemented, using default")
elif provider == "azure":
# Future: Azure deployer implementation
logger.warning("Azure provider not yet implemented, using default")
else:
logger.warning(f"Unknown provider {provider}, using default")
return scheduler
+2 -2
View File
@@ -12,7 +12,7 @@ import base64
from dotenv import load_dotenv
from PIL import Image
from tavily import TavilyClient
from crawl4ai import AsyncWebCrawler
from crawl4ai import AsyncAsyncWebCrawler
# Local application/library imports
import chainlit as cl
@@ -72,7 +72,7 @@ async def tavily_web_search(query):
response = tavily_client.search(query)
logger.debug(f"Tavily search response: {response}")
async with AsyncWebCrawler() as crawler:
async with AsyncAsyncWebCrawler() as crawler:
results = []
for result in response.get('results', []):
url = result.get('url')
+3 -3
View File
@@ -12,7 +12,7 @@ from dotenv import load_dotenv
from PIL import Image
from context import ContextGatherer
from tavily import TavilyClient
from crawl4ai import AsyncWebCrawler
from crawl4ai import AsyncAsyncWebCrawler
# Local application/library imports
import chainlit as cl
@@ -153,8 +153,8 @@ async def tavily_web_search(query):
response = tavily_client.search(query)
logger.debug(f"Tavily search response: {response}")
# Create an instance of AsyncWebCrawler
async with AsyncWebCrawler() as crawler:
# Create an instance of AsyncAsyncWebCrawler
async with AsyncAsyncWebCrawler() as crawler:
# Prepare the results
results = []
for result in response.get('results', []):
@@ -7,7 +7,7 @@ from litellm import acompletion
import json
import dotenv
from tavily import TavilyClient
from crawl4ai import AsyncWebCrawler
from crawl4ai import AsyncAsyncWebCrawler
dotenv.load_dotenv()
@@ -144,7 +144,7 @@ class AICoder:
})
response = self.tavily_client.search(query)
results = []
async with AsyncWebCrawler() as crawler:
async with AsyncAsyncWebCrawler() as crawler:
for result in response.get('results', []):
url = result.get('url')
if url:
@@ -3,7 +3,7 @@ import chainlit as cl
import plotly
import json
from tavily import TavilyClient
from crawl4ai import WebCrawler
from crawl4ai import AsyncWebCrawler
import os
import logging
import asyncio
@@ -108,7 +108,7 @@ async def tavily_web_search_handler(query):
try:
response = tavily_client.search(query_with_date)
logger.debug(f"Tavily search response: {response}")
results = process_tavily_results(response)
results = await process_tavily_results(response)
except Exception as e:
logger.error(f"Error in Tavily search: {str(e)}")
results = await fallback_to_duckduckgo(query_with_date)
@@ -121,27 +121,27 @@ async def tavily_web_search_handler(query):
"results": results
})
def process_tavily_results(response):
crawler = WebCrawler()
crawler.warmup()
results = []
for result in response.get('results', []):
url = result.get('url')
if url:
try:
crawl_result = crawler.run(url=url)
results.append({
"content": result.get('content'),
"url": url,
"full_content": crawl_result.markdown
})
except Exception as e:
logger.error(f"Error crawling {url}: {str(e)}")
results.append({
"content": result.get('content'),
"url": url,
"full_content": "Error: Unable to crawl this URL"
})
async def process_tavily_results(response):
async with AsyncWebCrawler() as crawler:
results = []
for result in response.get('results', []):
url = result.get('url')
if url:
try:
crawl_result = await crawler.arun(url=url)
full_content = crawl_result.markdown if crawl_result and hasattr(crawl_result, 'markdown') and crawl_result.markdown else "No content available"
results.append({
"content": result.get('content'),
"url": url,
"full_content": full_content
})
except Exception as e:
logger.error(f"Error crawling {url}: {str(e)}")
results.append({
"content": result.get('content'),
"url": url,
"full_content": "Error: Unable to crawl this URL"
})
return results
async def fallback_to_duckduckgo(query):
@@ -151,33 +151,33 @@ async def fallback_to_duckduckgo(query):
logger.debug(f"DuckDuckGo search results: {ddg_results}")
crawler = WebCrawler()
crawler.warmup()
results = []
for result in ddg_results:
url = result.get('href')
if url:
try:
crawl_result = crawler.run(url=url)
async with AsyncWebCrawler() as crawler:
results = []
for result in ddg_results:
url = result.get('href')
if url:
try:
crawl_result = await crawler.arun(url=url)
full_content = crawl_result.markdown if crawl_result and hasattr(crawl_result, 'markdown') and crawl_result.markdown else "No content available"
results.append({
"content": result.get('body'),
"url": url,
"full_content": full_content
})
except Exception as e:
logger.error(f"Error crawling {url}: {str(e)}")
results.append({
"content": result.get('body'),
"url": url,
"full_content": "Error: Unable to crawl this URL"
})
else:
results.append({
"content": result.get('body'),
"url": url,
"full_content": crawl_result.markdown
"url": "N/A",
"full_content": "No URL provided for crawling"
})
except Exception as e:
logger.error(f"Error crawling {url}: {str(e)}")
results.append({
"content": result.get('body'),
"url": url,
"full_content": "Error: Unable to crawl this URL"
})
else:
results.append({
"content": result.get('body'),
"url": "N/A",
"full_content": "No URL provided for crawling"
})
return results
except Exception as e:
+5 -3
View File
@@ -1,6 +1,6 @@
[project]
name = "PraisonAI"
version = "2.2.20"
version = "2.2.22"
description = "PraisonAI is an AI Agents Framework with Self Reflection. PraisonAI application combines PraisonAI Agents, AutoGen, and CrewAI into a low-code solution for building and managing multi-agent LLM systems, focusing on simplicity, customisation, and efficient human-agent collaboration."
readme = "README.md"
license = ""
@@ -89,12 +89,13 @@ call = [
"rich",
"openai>=1.54.0",
]
train = []
crewai = ["crewai>=0.32.0", "praisonai-tools>=0.0.15"]
autogen = ["pyautogen>=0.2.19", "praisonai-tools>=0.0.15", "crewai"]
[tool.poetry]
name = "PraisonAI"
version = "2.2.20"
version = "2.2.22"
description = "PraisonAI is an AI Agents Framework with Self Reflection. PraisonAI application combines PraisonAI Agents, AutoGen, and CrewAI into a low-code solution for building and managing multi-agent LLM systems, focusing on simplicity, customisation, and efficient human-agent collaboration."
authors = ["Mervin Praison"]
license = ""
@@ -146,6 +147,7 @@ sqlalchemy = {version = ">=2.0.36", optional = true}
playwright = {version = ">=1.47.0", optional = true}
openai = {version = ">=1.54.0", optional = true}
pydantic = {version = "<=2.10.1", optional = true}
# unsloth = {version = ">=2024.11.7", extras = ["colab-new"], optional = true}
[tool.poetry.group.docs.dependencies]
mkdocs = "*"
@@ -249,7 +251,7 @@ code = [
"playwright",
"pydantic"
]
train = ["setup-conda-env"]
train = []
realtime = [
"chainlit",
"litellm",
@@ -0,0 +1,30 @@
# PraisonAI Deployment Scheduler Configuration
# This is an example configuration file for scheduled deployments
deployment:
# Schedule expression (e.g., "daily", "hourly", "*/6h", "3600")
schedule: "daily"
# Deployment provider (gcp, aws, azure)
provider: "gcp"
# Maximum retry attempts on deployment failure
max_retries: 3
# Optional: timezone for scheduling (future feature)
# timezone: "UTC"
# Optional: retry delay in seconds
# retry_delay: 30
# Optional: Environment variables for deployment
# environment:
# OPENAI_MODEL_NAME: "gpt-4o"
# OPENAI_API_KEY: "your-api-key"
# OPENAI_API_BASE: "https://api.openai.com/v1"
# Optional: Notification settings (future feature)
# notifications:
# on_success: true
# on_failure: true
# webhook_url: "https://your-webhook-url.com"
+132
View File
@@ -0,0 +1,132 @@
#!/usr/bin/env python3
"""
Test script to verify Ollama tool-call fixes
"""
import sys
import os
# Add the source directory to Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', 'praisonai-agents'))
from praisonaiagents.llm.llm import LLM
from praisonaiagents.agent.agent import Agent
def test_ollama_provider_detection():
"""Test the new Ollama provider detection logic"""
print("Testing Ollama provider detection...")
# Test 1: Direct ollama/ prefix
llm1 = LLM(model="ollama/qwen3:32b")
assert llm1._is_ollama_provider() == True, "Should detect ollama/ prefix"
print("✓ Direct ollama/ prefix detection works")
# Test 2: Environment variable detection
os.environ["OPENAI_BASE_URL"] = "http://localhost:11434/v1"
llm2 = LLM(model="qwen3:32b")
assert llm2._is_ollama_provider() == True, "Should detect via env var"
print("✓ Environment variable detection works")
# Test 3: Non-Ollama model
os.environ.pop("OPENAI_BASE_URL", None)
llm3 = LLM(model="gpt-4o-mini")
assert llm3._is_ollama_provider() == False, "Should not detect non-Ollama"
print("✓ Non-Ollama detection works")
def test_tool_call_parsing():
"""Test the new tool call parsing logic"""
print("\nTesting tool call parsing...")
llm = LLM(model="ollama/qwen3:32b")
# Test 1: Standard format
tool_call_std = {
"id": "call_123",
"function": {
"name": "hello_world",
"arguments": '{"name": "test"}'
}
}
name, args, id = llm._parse_tool_call_arguments(tool_call_std, is_ollama=True)
assert name == "hello_world", f"Expected 'hello_world', got '{name}'"
assert args == {"name": "test"}, f"Expected {{'name': 'test'}}, got {args}"
print("✓ Standard format parsing works")
# Test 2: Ollama alternative format
tool_call_alt = {
"name": "hello_world",
"arguments": '{"name": "test"}'
}
name, args, id = llm._parse_tool_call_arguments(tool_call_alt, is_ollama=True)
assert name == "hello_world", f"Expected 'hello_world', got '{name}'"
assert args == {"name": "test"}, f"Expected {{'name': 'test'}}, got {args}"
print("✓ Alternative format parsing works")
# Test 3: Error handling - malformed JSON
tool_call_bad = {
"function": {
"name": "hello_world",
"arguments": 'invalid json'
}
}
name, args, id = llm._parse_tool_call_arguments(tool_call_bad, is_ollama=False)
assert name == "unknown_function", f"Expected 'unknown_function' for malformed JSON, got '{name}'"
assert args == {}, f"Expected empty dict, got {args}"
print("✓ Error handling works")
def test_agent_tool_parameter_logic():
"""Test the fixed tool parameter logic in Agent"""
print("\nTesting agent tool parameter logic...")
def dummy_tool():
"""Dummy tool for testing"""
return "test"
# Test 1: None tools should use agent tools
agent = Agent(name="test", tools=[dummy_tool])
# Simulate the fixed logic
tools = None
if tools is None or (isinstance(tools, list) and len(tools) == 0):
tool_param = agent.tools
else:
tool_param = tools
assert tool_param == [dummy_tool], "Should use agent tools when tools=None"
print("✓ None tools handling works")
# Test 2: Empty list should use agent tools
tools = []
if tools is None or (isinstance(tools, list) and len(tools) == 0):
tool_param = agent.tools
else:
tool_param = tools
assert tool_param == [dummy_tool], "Should use agent tools when tools=[]"
print("✓ Empty list handling works")
# Test 3: Non-empty list should use provided tools
custom_tools = [lambda: "custom"]
tools = custom_tools
if tools is None or (isinstance(tools, list) and len(tools) == 0):
tool_param = agent.tools
else:
tool_param = tools
assert tool_param == custom_tools, "Should use provided tools when not empty"
print("✓ Non-empty list handling works")
if __name__ == "__main__":
print("Running Ollama tool-call fix tests...\n")
try:
test_ollama_provider_detection()
test_tool_call_parsing()
test_agent_tool_parameter_logic()
print("\n🎉 All tests passed! The Ollama tool-call fixes are working correctly.")
sys.exit(0)
except Exception as e:
print(f"\n❌ Test failed: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
+161
View File
@@ -0,0 +1,161 @@
#!/usr/bin/env python3
"""
Test script for the deployment scheduler functionality.
Tests the core components without actually deploying.
"""
import sys
import os
import tempfile
import yaml
# Add the src directory to the path so we can import modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src/praisonai'))
def test_schedule_parser():
"""Test the schedule parser functionality."""
try:
from praisonai.scheduler import ScheduleParser
# Test various schedule expressions
test_cases = [
("daily", 86400),
("hourly", 3600),
("*/30m", 1800),
("*/6h", 21600),
("60", 60),
("3600", 3600)
]
print("Testing ScheduleParser...")
for expr, expected in test_cases:
result = ScheduleParser.parse(expr)
assert result == expected, f"Expected {expected}, got {result} for '{expr}'"
print(f" ✓ '{expr}' -> {result} seconds")
print("ScheduleParser tests passed!")
return True
except Exception as e:
print(f"ScheduleParser test failed: {e}")
return False
def test_scheduler_creation():
"""Test scheduler creation and basic functionality."""
try:
from praisonai.scheduler import create_scheduler, DeploymentScheduler
print("Testing scheduler creation...")
# Test default scheduler
scheduler = create_scheduler()
assert isinstance(scheduler, DeploymentScheduler)
print(" ✓ Default scheduler created")
# Test with different providers
for provider in ["gcp", "aws", "azure"]:
scheduler = create_scheduler(provider=provider)
assert isinstance(scheduler, DeploymentScheduler)
print(f" ✓ {provider} scheduler created")
print("Scheduler creation tests passed!")
return True
except Exception as e:
print(f"Scheduler creation test failed: {e}")
return False
def test_config_file_parsing():
"""Test configuration file parsing."""
try:
# Create a temporary config file
config_data = {
'deployment': {
'schedule': 'daily',
'provider': 'gcp',
'max_retries': 5
},
'environment': {
'TEST_VAR': 'test_value'
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
yaml.dump(config_data, f)
config_file = f.name
print("Testing config file parsing...")
# Test loading the config
with open(config_file, 'r') as f:
loaded_config = yaml.safe_load(f)
assert loaded_config['deployment']['schedule'] == 'daily'
assert loaded_config['deployment']['provider'] == 'gcp'
assert loaded_config['deployment']['max_retries'] == 5
print(" ✓ Config file parsed correctly")
# Clean up
os.unlink(config_file)
print("Config file parsing tests passed!")
return True
except Exception as e:
print(f"Config file parsing test failed: {e}")
return False
def test_cli_argument_parsing():
"""Test CLI argument parsing for scheduling options."""
try:
from praisonai.cli import PraisonAI
print("Testing CLI argument parsing...")
# Test basic CLI instantiation
praison = PraisonAI()
assert praison is not None
print(" ✓ PraisonAI CLI instance created")
print("CLI argument parsing tests passed!")
return True
except Exception as e:
print(f"CLI argument parsing test failed: {e}")
return False
def main():
"""Run all tests."""
print("Running PraisonAI Scheduler Tests")
print("=" * 40)
tests = [
test_schedule_parser,
test_scheduler_creation,
test_config_file_parsing,
test_cli_argument_parsing
]
passed = 0
total = len(tests)
for test in tests:
print()
if test():
passed += 1
else:
print("❌ Test failed")
print()
print("=" * 40)
print(f"Tests passed: {passed}/{total}")
if passed == total:
print("🎉 All tests passed!")
return 0
else:
print("❌ Some tests failed")
return 1
if __name__ == "__main__":
sys.exit(main())
+1 -1
View File
@@ -3931,7 +3931,7 @@ wheels = [
[[package]]
name = "praisonai"
version = "2.2.20"
version = "2.2.22"
source = { editable = "." }
dependencies = [
{ name = "instructor" },