Skip to main content

​
Thread Safety CLI

Commands for testing and verifying thread-safe operations in PraisonAI Agents.

​
Verification Commands

​
Run Thread Safety Tests

Run the built-in thread safety tests: 
cd /path/to/praisonai-agents
python -m pytest tests/unit/test_thread_safety.py -v
Output: 
tests/unit/test_thread_safety.py::TestLiteAgentThreadSafety::test_concurrent_chat_history_access PASSED
tests/unit/test_thread_safety.py::TestLiteAgentThreadSafety::test_concurrent_clear_history PASSED
tests/unit/test_thread_safety.py::TestAgentThreadSafety::test_agent_has_history_lock PASSED
tests/unit/test_thread_safety.py::TestAgentThreadSafety::test_agent_has_cache_lock PASSED

​
Quick Thread Safety Check

Verify thread safety with a quick Python command: 
python -c "
from praisonaiagents import Agent
import threading

agent = Agent(name='Test', instructions='You are a helpful assistant.')

# Verify locks exist
assert hasattr(agent, '_history_lock'), 'Missing history lock'
assert hasattr(agent, '_cache_lock'), 'Missing cache lock'

print('Thread safety locks: OK')
"

​
Lite Agent Thread Safety Check

python -c "
from praisonaiagents.lite import LiteAgent
import threading

agent = LiteAgent(name='Test', llm_fn=lambda m: 'Response')

# Verify lock exists
assert hasattr(agent, '_lock'), 'Missing lock'

# Test concurrent access
errors = []
def worker():
    try:
        for _ in range(50):
            agent.chat('Test')
    except Exception as e:
        errors.append(e)

threads = [threading.Thread(target=worker) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

assert len(errors) == 0, f'Errors: {errors}'
print('LiteAgent thread safety: OK')
"

​
Testing Concurrent Access

​
With Real API

Test concurrent API calls (requires OPENAI_API_KEY): 
export OPENAI_API_KEY="your-key"
python -c "
from praisonaiagents import Agent
import threading
import os

if not os.environ.get('OPENAI_API_KEY'):
    print('Skipping: OPENAI_API_KEY not set')
    exit(0)

agent = Agent(
    name='ConcurrentTest',
    instructions='Reply with one word only.',
    llm='gpt-4o-mini',
    output="silent"
)

results = []
lock = threading.Lock()

def worker(i):
    response = agent.chat(f'Say hello {i}')
    with lock:
        results.append(response)

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f'Completed {len(results)} concurrent requests')
print('Concurrent access test: OK')
"

​
Lock Verification

​
Check Agent Locks

python -c "
from praisonaiagents import Agent
import threading

agent = Agent(name='Test', instructions='You are a helpful assistant.')

# Check lock types
assert isinstance(agent._history_lock, type(threading.Lock()))
assert isinstance(agent._cache_lock, type(threading.RLock()))

print('Lock types verified:')
print('  _history_lock: Lock')
print('  _cache_lock: RLock')
"

​
Check LiteAgent Locks

python -c "
from praisonaiagents.lite import LiteAgent
import threading

agent = LiteAgent(name='Test', llm_fn=lambda m: 'ok')

assert hasattr(agent, '_lock')
print('LiteAgent lock verified')
"

​
CI/CD Integration

Add thread safety checks to your CI pipeline: 
# .github/workflows/test.yml
- name: Thread Safety Tests
  run: |
    pip install praisonaiagents pytest
    python -m pytest tests/unit/test_thread_safety.py -v