forked from cleveragents/cleveragents-core
============
CleverAgents
============
A powerful, reactive Agent Framework using RxPy streams for complex AI agent orchestration and message routing.
.. image:: https://img.shields.io/pypi/v/cleveragents.svg
:target: https://pypi.org/project/cleveragents/
:alt: PyPI Package
.. image:: https://img.shields.io/travis/cleverthis/cleveragents.svg
:target: https://travis-ci.org/cleverthis/cleveragents
:alt: Travis-CI Build Status
🌊 Reactive Architecture
========================
CleverAgents is built on **RxPy reactive streams**, providing powerful stream processing capabilities:
- **Full RxPy Integration**: Access to all RxPy operators (map, filter, merge, split, buffer, throttle, etc.)
- **Hot & Cold Streams**: Support for different stream types with replay capabilities
- **Async/Await Support**: Native async processing throughout the pipeline
- **Stream Composition**: Complex routing patterns with merge and split operations
🔄 Why RxPy + LangGraph?
========================
CleverAgents uniquely combines RxPy's reactive streams with LangGraph's stateful workflows. This integration is more powerful than either library alone:
**What Only RxPy Can Do:**
1. **Real-time Stream Processing**
.. code-block:: yaml
# Throttle API calls to 10/second while buffering bursts
operators:
- type: throttle
params: {duration: 0.1}
- type: buffer
params: {time: 1.0}
2. **Hot Streams with Live Updates**
.. code-block:: yaml
# Dashboard that shows latest value to new subscribers
dashboard_feed:
type: hot
initial_value: "System OK"
3. **Time-Window Aggregations**
.. code-block:: yaml
# Calculate 5-minute rolling averages
operators:
- type: buffer
params: {time: 300}
- type: scan
params: {accumulator: {type: average}}
**What Only LangGraph Can Do:**
1. **Stateful Conversations with Memory**
.. code-block:: yaml
# Remember context across messages
graphs:
chat:
checkpointing: true
enable_time_travel: true
2. **Conditional Workflow Routing**
.. code-block:: yaml
# Route based on runtime decisions
edges:
- source: classify
target: urgent_handler
condition: {type: priority_check}
3. **Iterative Refinement Loops**
.. code-block:: yaml
# Retry until quality threshold met
edges:
- source: review
target: refine
condition: {type: needs_improvement}
- source: refine
target: review # Loop back
**The Power of Both Together:**
1. **Streaming LangGraph Results**: Process partial results from long-running graphs in real-time
2. **Reactive Graph Triggers**: Use RxPy's debounce to prevent graph overload while maintaining responsiveness
3. **Windowed State Checkpoints**: Checkpoint graph state only when stream windows complete
4. **Parallel Stream-Graph Pipelines**: Run multiple graphs in parallel based on stream splits
This combination enables building AI systems that are both **reactive** (responding to real-time events) and **stateful** (maintaining context and memory) - something neither library can achieve alone.
🎯 LangGraph Integration
========================
CleverAgents provides full LangGraph integration, allowing you to combine stateful workflows with reactive stream processing:
**Key Features:**
- **Stateful Workflows**: Build complex, stateful agent graphs with memory and context
- **Conditional Routing**: Dynamic flow control based on state and conditions
- **Checkpointing**: Save and restore graph state with time travel capabilities
- **Cycles and Loops**: Support for iterative refinement and feedback loops
- **Subgraphs**: Compose complex workflows from reusable components
- **Parallel Execution**: Run multiple nodes concurrently for better performance
- **Seamless RxPy Integration**: Use LangGraphs as RxPy operators and vice versa
**LangGraph Definition in YAML:**
.. code-block:: yaml
graphs:
my_workflow:
name: my_workflow
entry_point: start
checkpointing: true
enable_time_travel: true
nodes:
process:
type: agent
agent: my_agent
validate:
type: function
function: validate
route:
type: conditional
condition:
type: has_messages
edges:
- source: start
target: process
- source: process
target: validate
- source: validate
target: route
- source: route
target: end
condition:
type: metadata_check
key: valid
value: true
**Node Types:**
- **Agent Nodes**: Execute CleverAgents agents
- **Function Nodes**: Run built-in or custom functions
- **Tool Nodes**: Execute tools with parallel support
- **Conditional Nodes**: Make routing decisions
- **Subgraph Nodes**: Embed other graphs
**State Management:**
LangGraph state flows through the graph and can be:
- Checkpointed for persistence
- Restored from checkpoints
- Time-traveled to previous states
- Updated incrementally or replaced
**Using LangGraphs as RxPy Operators:**
.. code-block:: yaml
streams:
my_stream:
type: cold
operators:
- type: graph_execute
params:
graph: my_workflow
- type: state_checkpoint
params:
graph: my_workflow
**Hybrid Pipelines:**
Combine RxPy streams and LangGraphs in complex pipelines:
.. code-block:: yaml
pipelines:
hybrid_flow:
stages:
- type: stream
name: preprocess
operators:
- type: debounce
params:
duration: 0.5
- type: graph
config:
name: process_graph
nodes:
analyze:
type: agent
agent: analyzer
input_from: preprocess
output_to: postprocess
- type: stream
name: postprocess
operators:
- type: buffer
params:
count: 5
**CLI Graph Commands:**
.. code-block:: bash
# Execute graphs interactively
cleveragents interactive -c my_config.yaml
>>> /graph my_workflow "Process this message"
**Programmatic Usage:**
.. code-block:: python
from cleveragents import ReactiveCleverAgentsApp
app = ReactiveCleverAgentsApp()
app.load_configuration([Path("my_config.yaml")])
# Get a graph
graph = app.langgraph_bridge.get_graph("my_workflow")
# Execute with input
result = await graph.execute({
"messages": [{"role": "user", "content": "Hello"}]
})
# Access state
state = graph.get_state()
print(state.messages)
**Advanced Patterns:**
1. **Iterative Refinement**: Use cycles to refine outputs until quality threshold is met
2. **Parallel Research**: Execute multiple research tasks in parallel and combine results
3. **Streaming Responses**: Process streaming LLM responses with real-time updates
4. **A/B Testing**: Route traffic between different graph variants for experimentation
5. **Error Recovery**: Implement retry policies and fallback paths for robustness
**Best Practices:**
1. **State Design**: Keep state minimal and well-structured
2. **Checkpointing**: Enable for long-running or critical workflows
3. **Parallel Execution**: Mark independent nodes as parallel
4. **Error Handling**: Use retry policies and error edges
5. **Monitoring**: Connect state streams to monitoring systems
**Integration with Existing Features:**
- **Templates**: Use Jinja2/Mustache templates in prompts
- **Tools**: Integrate existing CleverAgents tools in graphs
- **Agents**: All agent types work seamlessly in graphs
- **Streams**: Graphs can publish to and subscribe from streams
- **Context**: Global context is available in all graph nodes
**Troubleshooting:**
*Graph Not Executing:*
- Check edge definitions connect all nodes
- Verify entry point exists
- Ensure agents referenced in nodes are defined
*State Not Persisting:*
- Enable checkpointing in graph config
- Specify checkpoint directory
- Check filesystem permissions
*Parallel Execution Issues:*
- Mark nodes with ``parallel: true``
- Enable ``parallel_execution`` in graph config
- Check for dependencies between parallel nodes
*Stream Integration Problems:*
- Ensure stream router has LangGraph bridge reference
- Use correct operator names (graph_execute, state_update, etc.)
- Verify graph exists before referencing in streams
🤖 Agent Types
==============
**LLM Agents**
- OpenAI GPT models (GPT-4, GPT-3.5-turbo)
- Anthropic Claude models
- Google Gemini models
- Memory management and conversation history
**Tool Agents**
- Built-in tools: math, HTTP requests, file operations, JSON parsing
- Shell command execution (with safety controls)
- Custom tool integration
**Reactive Processing**
- Agents work as stream processors
- Can be used in map operations or connect between streams
- Support for both sync and async processing
📋 Quick Start
==============
Installation
------------
.. code-block:: bash
pip install cleveragents
Basic Usage
-----------
1. **Create a configuration file** (``config.yaml``):
.. code-block:: yaml
agents:
chat_agent:
type: llm
config:
provider: openai
model: gpt-4
temperature: 0.7
streams:
chat_stream:
type: cold
operators:
- type: map
params:
agent: chat_agent
publications:
- __output__
merges:
- sources: [__input__]
target: chat_stream
2. **Run single-shot processing**:
.. code-block:: bash
cleveragents run -c config.yaml -p "Hello, how are you?"
3. **Start interactive session**:
.. code-block:: bash
cleveragents interactive -c config.yaml
🔧 API Key Configuration
========================
CleverAgents requires API keys for LLM providers. Configure them in two ways:
1. **In Agent Configuration**:
.. code-block:: yaml
agents:
my_agent:
type: llm
config:
provider: openai
model: gpt-4
api_key: "sk-your-key-here"
2. **Via Environment Variables**:
- **OpenAI**: ``OPENAI_API_KEY``
- **Anthropic**: ``ANTHROPIC_API_KEY``
- **Google Gemini**: ``GOOGLE_GEMINI_API_KEY``
🌊 Stream Processing
====================
**Stream Types**
.. code-block:: yaml
streams:
cold_stream:
type: cold # Starts when subscribed
hot_stream:
type: hot # Always active, replays last value
replay_stream:
type: replay # Replays all previous values
**RxPy Operators**
.. code-block:: yaml
streams:
processing_stream:
operators:
- type: map
params:
agent: my_agent
- type: filter
params:
condition:
type: content_contains
text: "important"
- type: debounce
params:
duration: 1.0
- type: buffer
params:
count: 5
- type: throttle
params:
duration: 2.0
**Stream Operations**
.. code-block:: yaml
# Merge multiple streams
merges:
- sources: [stream1, stream2, stream3]
target: combined_stream
# Split stream based on conditions
splits:
- source: input_stream
targets:
questions:
type: content_contains
text: "?"
commands:
type: content_contains
text: "execute"
🏗️ Advanced Patterns
====================
**Multi-Agent Collaboration**
.. code-block:: yaml
# Research → Analysis → Writing → Editing pipeline
streams:
research_stream:
operators:
- type: map
params:
agent: researcher
publications:
- analysis_stream
analysis_stream:
operators:
- type: map
params:
agent: analyzer
publications:
- writing_stream
**Real-time Analytics**
.. code-block:: yaml
streams:
analytics_stream:
type: hot
operators:
- type: scan
params:
accumulator:
type: collect
- type: sample
params:
interval: 10.0
**Error Handling & Retry**
.. code-block:: yaml
streams:
robust_processing:
operators:
- type: map
params:
agent: my_agent
- type: catch
- type: retry
params:
count: 3
📊 CLI Commands
===============
**Run Commands**
.. code-block:: bash
# Single-shot processing
cleveragents run -c config.yaml -p "Your prompt here"
# Interactive session
cleveragents interactive -c config.yaml
# With verbose output
cleveragents run -c config.yaml -p "Hello" --verbose
# Unsafe mode (for file operations)
cleveragents run -c config.yaml -p "Command" --unsafe
**Utility Commands**
.. code-block:: bash
# Generate example configurations
cleveragents generate-examples -o ./my-examples
# Visualize stream network
cleveragents visualize -c config.yaml -f mermaid -o diagram.md
cleveragents visualize -c config.yaml -f dot -o network.dot
cleveragents visualize -c config.yaml -f ascii
📚 Examples
===========
The repository includes comprehensive examples:
**RxPy Stream Examples:**
- **basic_chat.yaml**: Simple conversational agent
- **advanced_pipeline.yaml**: Complex multi-stage processing with RxPy operators
- **multi_agent_collaboration.yaml**: Agent-to-agent communication
- **stream_analytics.yaml**: Real-time monitoring and analytics
**LangGraph Examples:**
- **simple_langgraph.yaml**: Basic linear workflow with a single agent
- **langgraph_conditional.yaml**: Conditional routing based on classification
- **langgraph_stateful.yaml**: Stateful conversations with checkpointing and time travel
- **hybrid_rxpy_langgraph.yaml**: Combine reactive streams with stateful graphs
**Advanced LangGraph Examples:**
- **advanced_langgraph_cycles.yaml**: Iterative refinement with cycles, parallel execution, and subgraphs
- **advanced_multi_graph_system.yaml**: Multi-graph orchestration with stream-based routing
- **advanced_streaming_langgraph.yaml**: Real-time streaming with partial results and windowed aggregation
🛡️ Safety & Security
====================
- **Safe Mode**: Blocks dangerous shell commands by default
- **Unsafe Flag**: Required for file operations and risky commands
- **API Key Protection**: Environment variable support
- **Error Boundaries**: Graceful error handling and recovery
🔄 Stream Visualization
=======================
Generate diagrams of your stream networks:
**Mermaid Format**:
.. code-block:: bash
cleveragents visualize -c config.yaml -f mermaid
**Graphviz DOT**:
.. code-block:: bash
cleveragents visualize -c config.yaml -f dot
**ASCII Diagram**:
.. code-block:: bash
cleveragents visualize -c config.yaml -f ascii
🎯 Use Cases
============
- **Conversational AI**: Multi-stage chat processing
- **Content Generation**: Research → Analysis → Writing workflows
- **Data Processing**: ETL pipelines with LLM processing
- **Decision Support**: Multi-agent decision making
- **Real-time Analytics**: Stream monitoring and alerting
- **API Orchestration**: Complex service integration
📋 Template System
==================
CleverAgents includes a powerful template system for creating reusable, parameterizable components across agents, graphs, and streams.
**Template Features**
- **Parameter Support**: Define parameters with types, defaults, and descriptions
- **Jinja2 Integration**: Full Jinja2 templating for dynamic content
- **Component References**: Reference agents, graphs, and streams across templates
- **Inheritance**: Build complex templates from simpler ones
- **Type Safety**: Parameter validation and type conversion
**Template Definition**
.. code-block:: yaml
templates:
agents:
my_template:
type: llm
parameters:
model:
description: "LLM model to use"
type: string
default: "gpt-3.5-turbo"
temperature:
description: "Temperature setting"
type: number
default: 0.7
config:
provider: openai
model: "{{ model }}"
temperature: "{{ temperature }}"
**Template Instantiation**
.. code-block:: yaml
agents:
my_agent:
template: my_template
params:
model: "gpt-4"
temperature: 0.9
**Template Types**
1. **Agent Templates**: Create reusable agent configurations
2. **Graph Templates**: Define parameterizable LangGraph workflows
3. **Stream Templates**: Build configurable stream processing pipelines
4. **Composite Templates**: Combine multiple components into reusable units
**Advanced Templating**
.. code-block:: yaml
templates:
graphs:
adaptive_workflow:
parameters:
stages:
type: number
default: 3
enable_validation:
type: boolean
default: true
nodes:
{% for i in range(stages) %}
stage_{{ i }}:
type: agent
agent: processor_{{ i }}
{% endfor %}
{% if enable_validation %}
validator:
type: function
function: validate
{% endif %}
edges:
- source: start
target: stage_0
{% for i in range(stages - 1) %}
- source: stage_{{ i }}
target: stage_{{ i + 1 }}
{% endfor %}
{% if enable_validation %}
- source: stage_{{ stages - 1 }}
target: validator
- source: validator
target: end
{% else %}
- source: stage_{{ stages - 1 }}
target: end
{% endif %}
**Composite Agent Templates**
.. code-block:: yaml
templates:
agents:
research_assistant:
type: composite
parameters:
research_depth:
type: string
default: "detailed"
enum: ["quick", "standard", "detailed"]
components:
agents:
researcher:
type: llm
config:
model: gpt-4
system_prompt: |
Research depth: {{ research_depth }}
Provide {{ research_depth }} analysis.
summarizer:
type: llm
config:
model: gpt-3.5-turbo
graphs:
workflow:
nodes:
research:
type: agent
agent: researcher
summarize:
type: agent
agent: summarizer
edges:
- source: start
target: research
- source: research
target: summarize
- source: summarize
target: end
routing:
input: workflow
output: workflow
**Template Registry**
.. code-block:: python
from cleveragents.templates import TemplateRegistry
# Access templates programmatically
registry = app.template_registry
# Get a template
template = registry.get_template("agent", "my_template")
# Instantiate with parameters
agent_config = template.instantiate({
"model": "gpt-4",
"temperature": 0.9
})
**Template Best Practices**
1. **Parameter Naming**: Use descriptive parameter names
2. **Default Values**: Provide sensible defaults for all parameters
3. **Type Constraints**: Define parameter types for validation
4. **Documentation**: Include descriptions for all parameters
5. **Modularity**: Build complex templates from simpler ones
**Use Cases**
1. **Multi-Environment Configs**: Same template, different parameters for dev/prod
2. **A/B Testing**: Create variants with different parameter values
3. **Dynamic Workflows**: Generate workflows based on runtime parameters
4. **Agent Libraries**: Build reusable agent components for teams
5. **Configuration Management**: Centralize common patterns
**Template Limitations**
1. **YAML Parsing**: Complex Jinja2 may require quoted strings
2. **Type Conversion**: Values rendered as strings need conversion
3. **Debugging**: Template errors may be harder to trace
🧪 Development
==============
**Running Tests**
.. code-block:: bash
# Run all tests
python -m pytest
# Run with coverage
python -m pytest --cov=cleveragents
# Run BDD tests
python -m behave
# Run with tox for multiple environments
tox
**Contributing**
1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request
📄 License
==========
Apache License 2.0
🔗 Links
========
- **Documentation**: https://cleveragents.readthedocs.io/
- **PyPI**: https://pypi.org/project/cleveragents/
- **GitHub**: https://github.com/cleverthis/cleveragents
- **Issues**: https://github.com/cleverthis/cleveragents/issues