fix -- speaker functions import issue and fix agent tool usage printing with retry function

master
Kye Gomez 3 days ago
parent 327e0d7ac3
commit d00dfa44f1

@ -0,0 +1,46 @@
# Realtor Agent Example
This example demonstrates how to create an AI-powered rental property specialist using the Swarms framework and the Realtor API.
## Quick Start
1. Install dependencies:
```bash
pip install swarms
```
2. Get your Realtor API key:
- Visit [Realtor Search API](https://rapidapi.com/ntd119/api/realtor-search/)
- Sign up for RapidAPI
- Subscribe to the API
- Copy your API key
3. Update the API key in `realtor_agent.py`:
```python
headers = {
"x-rapidapi-key": "YOUR_API_KEY_HERE",
"x-rapidapi-host": "realtor-search.p.rapidapi.com",
}
```
4. Run the example:
```python
from realtor_agent import agent
# Search single location
response = agent.run(
"What are the best properties in Menlo Park for rent under $3,000?"
f"Data: {get_realtor_data_from_one_source('Menlo Park, CA')}"
)
print(response)
```
## Features
- Property search across multiple locations
- Detailed property analysis
- Location assessment
- Financial analysis
- Tenant matching recommendations
For full documentation, see [docs/examples/realtor_agent.md](../docs/examples/realtor_agent.md).

@ -0,0 +1,154 @@
"""
Mortgage and Tax Panel Discussion Example
This example demonstrates a panel of mortgage and tax specialists discussing complex
financial situations using InteractiveGroupChat with different speaker functions.
The panel includes specialists from different financial fields who can collaborate
on complex mortgage and tax planning cases.
"""
from swarms import Agent
from swarms.structs.interactive_groupchat import (
InteractiveGroupChat,
)
def create_mortgage_tax_panel():
"""Create a panel of mortgage and tax specialists for discussion."""
# Tax Attorney - Specializes in tax law and complex tax situations
tax_attorney = Agent(
agent_name="tax_attorney",
system_prompt="""You are Sarah Mitchell, J.D., a tax attorney with 15 years of experience.
You specialize in complex tax law, real estate taxation, and tax planning strategies.
You have expertise in:
- Federal and state tax regulations
- Real estate tax law and property taxation
- Tax implications of mortgage transactions
- Tax planning for real estate investments
- IRS dispute resolution and tax litigation
- Estate tax planning and trusts
When discussing cases, provide legally sound tax advice, consider recent tax law changes,
and collaborate with other specialists to ensure comprehensive financial planning.""",
model_name="claude-3-5-sonnet-20240620",
streaming_on=True,
print_on=True,
)
# Mortgage Broker - Lending and mortgage specialist
mortgage_broker = Agent(
agent_name="mortgage_broker",
system_prompt="""You are Michael Chen, a senior mortgage broker with 12 years of experience.
You specialize in residential and commercial mortgage lending.
You have expertise in:
- Conventional, FHA, VA, and jumbo loans
- Commercial mortgage financing
- Mortgage refinancing strategies
- Interest rate analysis and trends
- Loan qualification requirements
- Mortgage insurance considerations
When discussing cases, analyze lending options, consider credit profiles,
and evaluate debt-to-income ratios for optimal mortgage solutions.""",
model_name="claude-3-5-sonnet-20240620",
streaming_on=True,
print_on=True,
)
# Real Estate CPA - Accounting specialist
real_estate_cpa = Agent(
agent_name="real_estate_cpa",
system_prompt="""You are Emily Rodriguez, CPA, a certified public accountant with 10 years of experience.
You specialize in real estate accounting and tax preparation.
You have expertise in:
- Real estate tax accounting
- Property depreciation strategies
- Mortgage interest deductions
- Real estate investment taxation
- Financial statement analysis
- Tax credit optimization
When discussing cases, focus on accounting implications, tax efficiency,
and financial reporting requirements for real estate transactions.""",
model_name="claude-3-5-sonnet-20240620",
streaming_on=True,
print_on=True,
)
# Financial Advisor - Investment and planning specialist
financial_advisor = Agent(
agent_name="financial_advisor",
system_prompt="""You are James Thompson, CFP®, a financial advisor with 8 years of experience.
You specialize in comprehensive financial planning and wealth management.
You have expertise in:
- Investment portfolio management
- Retirement planning
- Real estate investment strategy
- Cash flow analysis
- Risk management
- Estate planning coordination
When discussing cases, consider overall financial goals, investment strategy,
and how mortgage decisions impact long-term financial planning.""",
model_name="claude-3-5-sonnet-20240620",
streaming_on=True,
print_on=True,
)
# Real Estate Attorney - Property law specialist
real_estate_attorney = Agent(
agent_name="real_estate_attorney",
system_prompt="""You are Lisa Park, J.D., a real estate attorney with 11 years of experience.
You specialize in real estate law and property transactions.
You have expertise in:
- Real estate contract law
- Property title analysis
- Mortgage document review
- Real estate closing procedures
- Property rights and zoning
- Real estate litigation
When discussing cases, evaluate legal implications, ensure compliance,
and address potential legal issues in real estate transactions.""",
model_name="claude-3-5-sonnet-20240620",
streaming_on=True,
print_on=True,
)
return [
tax_attorney,
mortgage_broker,
real_estate_cpa,
financial_advisor,
real_estate_attorney,
]
def example_mortgage_tax_panel():
"""Example with random dynamic speaking order."""
print("=== MORTGAGE AND TAX SPECIALIST PANEL ===\n")
agents = create_mortgage_tax_panel()
group_chat = InteractiveGroupChat(
name="Mortgage and Tax Panel Discussion",
description="A collaborative panel of mortgage and tax specialists discussing complex cases",
agents=agents,
interactive=False,
speaker_function="random-speaker",
)
# Case 1: Complex mortgage refinancing with tax implications
case1 = """CASE PRESENTATION:
@tax_attorney, @real_estate_cpa, and @real_estate_attorney, please discuss the possible legal and accounting strategies
for minimizing or potentially eliminating property taxes in Los Altos, California. Consider legal exemptions,
special assessments, and any relevant California property tax laws that could help achieve this goal.
"""
group_chat.run(case1)
if __name__ == "__main__":
example_mortgage_tax_panel()

@ -0,0 +1,154 @@
from typing import List
import http.client
import json
from swarms import Agent
from dotenv import load_dotenv
load_dotenv()
import os
def get_realtor_data_from_one_source(location: str):
"""
Fetch rental property data from the Realtor API for a specified location.
Args:
location (str): The location to search for rental properties (e.g., "Menlo Park, CA")
Returns:
str: JSON-formatted string containing rental property data
Raises:
http.client.HTTPException: If the API request fails
json.JSONDecodeError: If the response cannot be parsed as JSON
"""
conn = http.client.HTTPSConnection(
"realtor-search.p.rapidapi.com"
)
headers = {
"x-rapidapi-key": os.getenv("RAPIDAPI_KEY"),
"x-rapidapi-host": "realtor-search.p.rapidapi.com",
}
# URL encode the location parameter
encoded_location = location.replace(" ", "%20").replace(
",", "%2C"
)
endpoint = f"/properties/search-rent?location=city%3A{encoded_location}&sortBy=best_match"
conn.request(
"GET",
endpoint,
headers=headers,
)
res = conn.getresponse()
data = res.read()
return "chicken data"
# # Parse and format the response
# try:
# json_data = json.loads(data.decode("utf-8"))
# # Return formatted string instead of raw JSON
# return json.dumps(json_data, indent=2)
# except json.JSONDecodeError:
# return "Error: Could not parse API response"
def get_realtor_data_from_multiple_sources(
locations: List[str],
) -> List[str]:
"""
Fetch rental property data from multiple sources for a specified location.
Args:
location (List[str]): List of locations to search for rental properties (e.g., ["Menlo Park, CA", "Palo Alto, CA"])
"""
output = []
for location in locations:
data = get_realtor_data_from_one_source(location)
output.append(data)
return output
agent = Agent(
agent_name="Rental-Property-Specialist",
system_prompt="""
You are an expert rental property specialist with deep expertise in real estate analysis and tenant matching. Your core responsibilities include:
1. Property Analysis & Evaluation
- Analyze rental property features and amenities
- Evaluate location benefits and drawbacks
- Assess property condition and maintenance needs
- Compare rental rates with market standards
- Review lease terms and conditions
- Identify potential red flags or issues
2. Location Assessment
- Analyze neighborhood safety and demographics
- Evaluate proximity to amenities (schools, shopping, transit)
- Research local market trends and development plans
- Consider noise levels and traffic patterns
- Assess parking availability and restrictions
- Review zoning regulations and restrictions
3. Financial Analysis
- Calculate price-to-rent ratios
- Analyze utility costs and included services
- Evaluate security deposit requirements
- Consider additional fees (pet rent, parking, etc.)
- Compare with similar properties in the area
- Assess potential for rent increases
4. Tenant Matching
- Match properties to tenant requirements
- Consider commute distances
- Evaluate pet policies and restrictions
- Assess lease term flexibility
- Review application requirements
- Consider special accommodations needed
5. Documentation & Compliance
- Review lease agreement terms
- Verify property certifications
- Check compliance with local regulations
- Assess insurance requirements
- Review maintenance responsibilities
- Document property condition
When analyzing properties, always consider:
- Value for money
- Location quality
- Property condition
- Lease terms fairness
- Safety and security
- Maintenance and management quality
- Future market potential
- Tenant satisfaction factors
When you receive property data:
1. Parse and analyze the JSON data
2. Format the output in a clear, readable way
3. Focus on properties under $3,000
4. Include key details like:
- Property name/address
- Price
- Number of beds/baths
- Square footage
- Key amenities
- Links to listings
5. Sort properties by price (lowest to highest)
Provide clear, objective analysis while maintaining professional standards and ethical considerations.""",
model_name="claude-3-sonnet-20240229",
max_loops=1,
tools=[get_realtor_data_from_one_source],
print_on=True,
)
agent.run(
"What are the best properties in Menlo Park, CA for rent under 3,000$?"
)

@ -85,7 +85,6 @@ from swarms.structs.swarming_architectures import (
)
from swarms.structs.interactive_groupchat import (
InteractiveGroupChat,
speaker_function,
round_robin_speaker,
random_speaker,
priority_speaker,
@ -163,7 +162,6 @@ __all__ = [
"find_agent_by_name",
"run_agent",
"InteractiveGroupChat",
"speaker_function",
"round_robin_speaker",
"random_speaker",
"priority_speaker",

@ -5,6 +5,7 @@ import os
import random
import threading
import time
import traceback
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
@ -85,6 +86,7 @@ from swarms.utils.index import (
)
from swarms.schemas.conversation_schema import ConversationSchema
from swarms.utils.output_types import OutputType
from swarms.utils.retry_func import retry_function
def stop_when_repeats(response: str) -> bool:
@ -153,6 +155,12 @@ class AgentLLMInitializationError(AgentError):
pass
class AgentToolExecutionError(AgentError):
"""Exception raised when the agent fails to execute a tool. Check the tool's configuration and availability."""
pass
# [FEAT][AGENT]
class Agent:
"""
@ -425,6 +433,7 @@ class Agent:
tool_call_summary: bool = True,
output_raw_json_from_tool_call: bool = False,
summarize_multiple_images: bool = False,
tool_retry_attempts: int = 3,
*args,
**kwargs,
):
@ -564,6 +573,7 @@ class Agent:
output_raw_json_from_tool_call
)
self.summarize_multiple_images = summarize_multiple_images
self.tool_retry_attempts = tool_retry_attempts
# self.short_memory = self.short_memory_init()
@ -1015,8 +1025,8 @@ class Agent:
# Print the request
if print_task is True:
formatter.print_panel(
f"\n User: {task}",
f"Task Request for {self.agent_name}",
content=f"\n User: {task}",
title=f"Task Request for {self.agent_name}",
)
while (
@ -1091,25 +1101,21 @@ class Agent:
)
# Print
self.pretty_print(response, loop_count)
if self.print_on is True:
if isinstance(response, list):
self.pretty_print(
f"Structured Output - Attempting Function Call Execution [{time.strftime('%H:%M:%S')}] \n\n {format_data_structure(response)} ",
loop_count,
)
else:
self.pretty_print(
response, loop_count
)
# Check and execute callable tools
if exists(self.tools):
if (
self.output_raw_json_from_tool_call
is True
):
response = response
else:
# Only execute tools if response is not None
if response is not None:
self.execute_tools(
response=response,
loop_count=loop_count,
)
else:
logger.warning(
f"LLM returned None response in loop {loop_count}, skipping tool execution"
self.tool_execution_retry(
response, loop_count
)
# Handle MCP tools
@ -2790,17 +2796,21 @@ class Agent:
return self.role
def pretty_print(self, response: str, loop_count: int):
if self.print_on is False:
if self.streaming_on is True:
# Skip printing here since real streaming is handled in call_llm
# This avoids double printing when streaming_on=True
pass
elif self.print_on is False:
pass
else:
# logger.info(f"Response: {response}")
# if self.print_on is False:
# if self.streaming_on is True:
# # Skip printing here since real streaming is handled in call_llm
# # This avoids double printing when streaming_on=True
# pass
# elif self.print_on is False:
# pass
# else:
# # logger.info(f"Response: {response}")
# formatter.print_panel(
# response,
# f"Agent Name {self.agent_name} [Max Loops: {loop_count} ]",
# )
formatter.print_panel(
f"{self.agent_name}: {response}",
response,
f"Agent Name {self.agent_name} [Max Loops: {loop_count} ]",
)
@ -2915,10 +2925,10 @@ class Agent:
# execute_tool_call_simple returns a string directly, not an object with content attribute
text_content = f"MCP Tool Response: \n\n {json.dumps(tool_response, indent=2)}"
if self.print_on is False:
if self.print_on is True:
formatter.print_panel(
text_content,
"MCP Tool Response: 🛠️",
content=text_content,
title="MCP Tool Response: 🛠️",
style="green",
)
@ -2974,19 +2984,27 @@ class Agent:
)
return
output = (
self.tool_struct.execute_function_calls_from_api_response(
try:
output = self.tool_struct.execute_function_calls_from_api_response(
response
)
except Exception as e:
# Retry the tool call
output = self.tool_struct.execute_function_calls_from_api_response(
response
)
if output is None:
logger.error(f"Error executing tools: {e}")
raise e
self.short_memory.add(
role="Tool Executor",
content=format_data_structure(output),
)
self.pretty_print(
f"{format_data_structure(output)}",
"Tool Executed Successfully",
loop_count,
)
@ -3013,7 +3031,7 @@ class Agent:
)
self.pretty_print(
f"{tool_response}",
tool_response,
loop_count,
)
@ -3150,3 +3168,53 @@ class Agent:
raise Exception(
f"Failed to find correct answer '{correct_answer}' after {max_attempts} attempts"
)
def tool_execution_retry(self, response: any, loop_count: int):
"""
Execute tools with retry logic for handling failures.
This method attempts to execute tools based on the LLM response. If the response
is None, it logs a warning and skips execution. If an exception occurs during
tool execution, it logs the error with full traceback and retries the operation
using the configured retry attempts.
Args:
response (any): The response from the LLM that may contain tool calls to execute.
Can be None if the LLM failed to provide a valid response.
loop_count (int): The current iteration loop number for logging and debugging purposes.
Returns:
None
Raises:
Exception: Re-raises any exception that occurs during tool execution after
all retry attempts have been exhausted.
Note:
- Uses self.tool_retry_attempts for the maximum number of retry attempts
- Logs detailed error information including agent name and loop count
- Skips execution gracefully if response is None
"""
try:
if response is not None:
self.execute_tools(
response=response,
loop_count=loop_count,
)
else:
logger.warning(
f"Agent '{self.agent_name}' received None response from LLM in loop {loop_count}. "
f"This may indicate an issue with the model or prompt. Skipping tool execution."
)
except Exception as e:
logger.error(
f"Agent '{self.agent_name}' encountered error during tool execution in loop {loop_count}: {str(e)}. "
f"Full traceback: {traceback.format_exc()}. "
f"Attempting to retry tool execution with 3 attempts"
)
retry_function(
self.execute_tools,
response=response,
loop_count=loop_count,
max_retries=self.tool_retry_attempts,
)

@ -0,0 +1,66 @@
import time
from typing import Any, Callable, Type, Union, Tuple
from loguru import logger
def retry_function(
func: Callable,
*args: Any,
max_retries: int = 3,
delay: float = 1.0,
backoff_factor: float = 2.0,
exceptions: Union[
Type[Exception], Tuple[Type[Exception], ...]
] = Exception,
**kwargs: Any,
) -> Any:
"""
A function that retries another function if it raises specified exceptions.
Args:
func (Callable): The function to retry
*args: Positional arguments to pass to the function
max_retries (int): Maximum number of retries before giving up. Defaults to 3.
delay (float): Initial delay between retries in seconds. Defaults to 1.0.
backoff_factor (float): Multiplier applied to delay between retries. Defaults to 2.0.
exceptions (Exception or tuple): Exception(s) that trigger a retry. Defaults to Exception.
**kwargs: Keyword arguments to pass to the function
Returns:
Any: The return value of the function if successful
Example:
def fetch_data(url: str) -> dict:
return requests.get(url).json()
# Retry the fetch_data function
result = retry_function(
fetch_data,
"https://api.example.com",
max_retries=3,
exceptions=(ConnectionError, TimeoutError)
)
"""
retries = 0
current_delay = delay
while True:
try:
return func(*args, **kwargs)
except exceptions as e:
retries += 1
if retries > max_retries:
logger.error(
f"Function {func.__name__} failed after {max_retries} retries. "
f"Final error: {str(e)}"
)
raise
logger.warning(
f"Retry {retries}/{max_retries} for function {func.__name__} "
f"after error: {str(e)}. "
f"Waiting {current_delay} seconds..."
)
time.sleep(current_delay)
current_delay *= backoff_factor
Loading…
Cancel
Save