Thread-Safe Agent State

from praisonaiagents import Agent
import threading

agent = Agent(name="Assistant", instructions="You are helpful.")

def worker(prompt):
    agent.chat(prompt)

threads = [threading.Thread(target=worker, args=(f"Question {i}",)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

from praisonaiagents import Agent
import threading

agent = Agent(
    name="ThreadSafeAgent",
    instructions="You are helpful."
)

def worker(prompt):
    # Safe to call from multiple threads
    response = agent.chat(prompt)
    print(f"Response: {response[:50]}...")

# Create multiple threads
threads = [
    threading.Thread(target=worker, args=(f"Question {i}",))
    for i in range(5)
]

# Start all threads
for t in threads:
    t.start()

# Wait for completion
for t in threads:
    t.join()

# These operations are now thread-safe out of the box:
agent.chat_history = []  # Full replacement - uses locked setter
agent.chat("Hello")      # Appends safely via internal methods

# Reading is always safe:
history = agent.chat_history
print(f"History has {len(history)} messages")

from praisonaiagents.lite import LiteAgent, create_openai_llm_fn
import threading

llm_fn = create_openai_llm_fn(model="gpt-4o-mini")
agent = LiteAgent(name="LiteThreadSafe", llm_fn=llm_fn)

def concurrent_chat(message):
    return agent.chat(message)

# Safe concurrent access
with threading.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(concurrent_chat, f"Q{i}") for i in range(10)]
    results = [f.result() for f in futures]

# Internal implementation pattern
class Agent:
    def __init__(self):
        self.__chat_history_state = AsyncSafeState([])
        self._cache_lock = threading.RLock()
    
    @property
    def _history_lock(self):
        return self.__chat_history_state
    
    def _append_to_chat_history(self, message):
        with self._history_lock.lock():
            self._history_lock.value.append(message)

from praisonaiagents import Agent
from praisonai.persistence import PersistenceOrchestrator
import threading

# Create shared orchestrator
orchestrator = PersistenceOrchestrator()

def agent_worker(agent_name, session_id):
    agent = Agent(name=agent_name)
    
    # Safe: orchestrator returns deep copy of cached session
    history = orchestrator.on_agent_start(agent, session_id=session_id)
    
    # Multiple agents can safely read/update sessions concurrently
    orchestrator.on_message(session_id, "user", "Hello from " + agent_name)
    orchestrator.on_agent_end(agent, session_id)

# Multiple threads can safely share the orchestrator
threads = [
    threading.Thread(target=agent_worker, args=(f"Agent{i}", "session-123"))
    for i in range(3)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

import asyncio
from praisonaiagents import Agent

agent = Agent(name="AsyncAgent")

async def async_chat(prompt):
    # No external lock needed - AsyncSafeState handles this
    return agent.chat(prompt)

async def main():
    tasks = [async_chat(f"Question {i}") for i in range(5)]
    results = await asyncio.gather(*tasks)

import asyncio
import threading
from praisonaiagents import Agent

agent = Agent(name="MixedAgent")
external_lock = threading.Lock()

def sync_worker(prompt):
    with external_lock:
        return agent.chat(prompt)

async def async_worker(prompt):
    # Convert to sync context for coordination
    loop = asyncio.get_event_loop()
    with external_lock:
        return await loop.run_in_executor(None, agent.chat, prompt)

import threading
from praisonaiagents.lite import LiteAgent

def test_thread_safety():
    agent = LiteAgent(
        name="Test",
        llm_fn=lambda m: "Response"
    )
    
    errors = []
    
    def worker():
        try:
            for _ in range(100):
                agent.chat("Test")
        except Exception as e:
            errors.append(e)
    
    threads = [threading.Thread(target=worker) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    
    assert len(errors) == 0, f"Thread safety errors: {errors}"
    print("Thread safety test passed!")

test_thread_safety()

import threading
from praisonaiagents import AgentTeam

def launch_team(team_name, port, path):
    team = AgentTeam(name=team_name)
    team.launch(port=port, path=path)

# Safe concurrent launches
thread1 = threading.Thread(target=launch_team, args=("TeamA", 8000, "/team_a"))
thread2 = threading.Thread(target=launch_team, args=("TeamB", 8000, "/team_b"))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

# Both teams available at:
# - http://localhost:8000/team_a
# - http://localhost:8000/team_b

import os
from praisonai import PraisonAI

# Each PraisonAI instance constructs its own auto-generator, which owns
# a single OpenAI client. Two concurrent instances cannot evict each
# other's client — even with identical API keys.
team_a = PraisonAI(auto="Draft a marketing plan", api_key=os.getenv("OPENAI_API_KEY"))
team_b = PraisonAI(auto="Draft a technical doc", api_key=os.getenv("OPENAI_API_KEY"))

team_a.run()  # safe to run in parallel threads
team_b.run()

import os
from praisonai.auto import BaseAutoGenerator

# Sync context manager
with BaseAutoGenerator(config_list=[{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": None,
}]) as gen:
    # _get_core_client() is lock-guarded; calling it from multiple threads
    # on the same instance returns the same core OpenAIClient.
    result = gen._structured_completion(MyModel, messages=[...])

# Async context manager
async with BaseAutoGenerator(config_list=[{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": None,
}]) as gen:
    result = await gen._structured_completion_async(MyModel, messages=[...], is_async=True)

import threading
import subprocess

def run_cli_command(command):
    # Safe to call from multiple threads
    result = subprocess.run(
        ["python", "-m", "praisonai"] + command,
        capture_output=True, text=True
    )
    return result.stdout

# Multiple threads can safely use the CLI
threads = [
    threading.Thread(target=run_cli_command, args=(["--version"],))
    for _ in range(5)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

from praisonai.scheduler import AgentScheduler

scheduler = AgentScheduler(agent, "Hourly news check")
scheduler.start("hourly", max_retries=3)
# ... later, from another thread or signal handler:
scheduler.stop()  # returns within ms, even if a backoff sleep was in progress

import asyncio
from praisonaiagents import Agent
from praisonaiagents import parallel_handoffs

billing = Agent(name="Billing", instructions="Handle billing inquiries.")
refund = Agent(name="Refunds", instructions="Process refund requests.")
tech = Agent(name="TechSupport", instructions="Solve technical problems.")
triage = Agent(name="Triage", instructions="Route inquiries to specialists.")

async def main():
    results = await parallel_handoffs(
        triage,
        targets=[
            (billing, "Why was I charged twice?"),
            (refund, "I want a refund for order #42"),
            (tech, "The app crashes on upload"),
        ],
    )
    for r in results:
        print(r.target_agent, r.success)

asyncio.run(main())