### Install LoopGuard

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Install the package via pip.

```bash
pip install loopguard
```

--------------------------------

### Concurrent execution with loopguard

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Demonstrates using the loopguard decorator on a function and invoking it from multiple threads.

```python
import threading

@loopguard(max_repeats=100, window=60)
def my_func(x):
    return x

# Safe to call from multiple threads
threads = [threading.Thread(target=lambda: my_func(1)) for _ in range(10)]
```

--------------------------------

### Async Support

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Use async_loopguard for asynchronous functions, with optional async handlers.

```python
from loopguard import async_loopguard

@async_loopguard(max_repeats=3, window=60)
async def async_agent_action(query: str) -> str:
    return await llm.complete(query)
```

```python
async def my_handler(func, args, kwargs):
    await log_loop_event()
    return "fallback response"

@async_loopguard(max_repeats=3, on_loop=my_handler)
async def agent_action(query: str) -> str:
    return await llm.complete(query)
```

--------------------------------

### Integration with AI Frameworks

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Apply loopguard to tools or agents in LangChain or CrewAI.

```python
from langchain.tools import tool
from loopguard import loopguard

@tool
@loopguard(max_repeats=3, window=120)
def search_tool(query: str) -> str:
    """Search the web."""
    return search_api.search(query)
```

```python
from crewai import Agent, Task
from loopguard import loopguard

@loopguard(max_repeats=5, window=300)
def execute_task(task: Task) -> str:
    return agent.execute(task)
```

--------------------------------

### Basic LoopGuard Usage

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Apply the loopguard decorator to limit function execution frequency.

```python
from loopguard import loopguard

@loopguard(max_repeats=3, window=60)
def agent_action(query: str) -> str:
    return llm.complete(query)

# Third call with same query within 60s raises LoopDetectedError
```

```python
from loopguard import loopguard, LoopDetectedError

@loopguard(max_repeats=3, window=60)
def search(query: str) -> str:
    return search_api.search(query)

try:
    for _ in range(10):
        search("same query")  # Raises on 4th call
except LoopDetectedError as e:
    print(f"Loop stopped: {e}")
```

--------------------------------

### Sub-second Rate Limiting

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Use float values for the window parameter to achieve sub-second precision.

```python
# Allow max 5 calls per 500ms
@loopguard(max_repeats=5, window=0.5)
def rate_limited_api(query: str) -> str:
    return api.call(query)
```

--------------------------------

### Custom Loop Handler

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Define a custom callback function to execute when a loop is detected.

```python
@loopguard(max_repeats=3, on_loop=lambda f, a, k: "Loop detected, stopping")
def agent_step(state: dict) -> str:
    return llm.complete(state["query"])
```

--------------------------------

### loopguard(max_repeats=3, window=60, on_loop=None)

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Decorator for synchronous functions to detect repeated calls with identical arguments. It is thread-safe and uses monotonic time.

```APIDOC
## loopguard(max_repeats=3, window=60, on_loop=None)

### Description
Decorator for sync functions. Thread-safe. Uses monotonic time internally, so immune to system clock adjustments.

### Parameters
- **max_repeats** (int) - Optional - Max calls with identical args within window (default: 3)
- **window** (float/int) - Optional - Time window in seconds (default: 60)
- **on_loop** (callable) - Optional - Callback (func, args, kwargs) -> Any. If provided, return value is used instead of raising.

### Attached Methods
- **func.reset()** - Clear all call history
- **func.get_count(args_tuple, kwargs_dict=None)** - Get current count for specific arguments
- **func.would_trigger(args_tuple, kwargs_dict=None)** - Check if next call would trigger loop detection
- **func.get_signatures()** - Get list of tracked signature hashes
```

--------------------------------

### Manage LoopGuard State

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Access call counts or reset the history of a decorated function.

```python
@loopguard(max_repeats=5, window=60)
def my_func(x):
    return x

my_func(10)
my_func(10)
print(my_func.get_count((10,)))  # 2
```

```python
@loopguard(max_repeats=2, window=60)
def my_func(x):
    return x

my_func(5)
my_func(5)
my_func.reset()  # Clear all history
my_func(5)  # Works again
```

--------------------------------

### async_loopguard(max_repeats=3, window=60, on_loop=None)

Source: https://github.com/yolojewjitsu/loopguard/blob/main/README.md

Decorator for asynchronous functions to detect repeated calls with identical arguments. It is coroutine-safe.

```APIDOC
## async_loopguard(max_repeats=3, window=60, on_loop=None)

### Description
Same as loopguard, for async functions. Coroutine-safe. The on_loop callback can be sync or async.

### Parameters
- **max_repeats** (int) - Optional - Max calls with identical args within window (default: 3)
- **window** (float/int) - Optional - Time window in seconds (default: 60)
- **on_loop** (callable) - Optional - Callback (func, args, kwargs) -> Any. If provided, return value is used instead of raising.
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.