freemo/cleveragents-core
0
0
Fork 0
forked from cleveragents/cleveragents-core
Code Pull Requests Activity
Files
config/memory-yaml
cleveragents-core/docs/advanced_usage.rst
T

600 lines
18 KiB
ReStructuredText
Raw Permalink Blame History

==============
Advanced Usage
==============
Agent Configuration
===================
Agents are defined in the ``agents`` section of your configuration file. Here's an overview of common configuration for the built-in ``llm`` agent type.
LLMAgent
~~~~~~~~
The ``LLMAgent`` is powered by a large language model. Its configuration includes:
- ``provider``: The LLM provider to use (e.g., ``openai``, ``anthropic``).
- ``model``: The specific model to use (e.g., ``gpt-4``, ``claude-3-opus-20240229``).
- ``role``: An inline string defining the agent's role or system message.
- ``role_reference``: The name of a template from the top-level ``prompts`` section to use as the role. Cannot be used with ``role``.
- ``prompt``: An inline template for formatting the user's input message.
- ``prompt_reference``: The name of a template from ``prompts`` to format the user's input. Cannot be used with ``prompt``.
- ``response``: An inline template for formatting the LLM's output.
- ``response_reference``: The name of a template from ``prompts`` to format the LLM's output. Cannot be used with ``response``.
- ``api_key``: (Optional) The API key for the provider. If not provided, it will be read from the corresponding environment variable (e.g., ``OPENAI_API_KEY``).
Example:
.. code-block:: yaml
agents:
researcher:
type: llm
config:
provider: openai
model: gpt-4
role: "You are a research assistant."
prompt_reference: research_task
temperature: 0.7
prompts:
research_task:
content: "Research the topic: {{ message }}"
Custom Agent Implementation
==========================
This section covers advanced usage scenarios for CleverAgents.
Custom Agent Implementation
==========================
Creating custom agents allows you to extend CleverAgents with specialized functionality.
Basic Custom Agent
----------------
Here's how to create a basic custom agent:
.. code-block:: python
from cleveragents.agents.base import Agent
class MyCustomAgent(Agent):
def __init__(self, name, config, template_renderer):
super().__init__(name, config, template_renderer)
# Initialize your custom agent
self.custom_parameter = config.get("custom_parameter", "default")
async def process(self, message, context=None):
# Process the message
return f"Custom agent processed: {message}"
def get_capabilities(self):
return ["custom_processing"]
# Register the custom agent with the factory
from cleveragents.agents.factory import AgentFactory
factory = AgentFactory(config, template_renderer)
factory.register_agent_type("custom", MyCustomAgent)
# Create an instance of the custom agent
custom_agent = factory.create_agent("my_custom_agent")
Stateful Custom Agent
-------------------
For agents that need to maintain state:
.. code-block:: python
from cleveragents.agents.base import AgentWithMemory
class MyStatefulAgent(AgentWithMemory):
def __init__(self, name, config, template_renderer):
super().__init__(name, config, template_renderer)
# Initialize memory
self.memory = {
"conversation_count": 0,
"last_message": None
}
async def process(self, message, context=None):
# Update memory
self.memory["conversation_count"] += 1
self.memory["last_message"] = message
# Process the message
return f"Message {self.memory['conversation_count']}: {message}"
def get_capabilities(self):
return ["stateful_processing"]
def save_memory(self):
# Return the memory for persistence
return self.memory
def load_memory(self, memory):
# Load the memory from persistence
self.memory = memory
Custom Tools
===========
You can create custom tools to extend the functionality of tool agents.
Basic Custom Tool
--------------
Here's how to create a basic custom tool:
.. code-block:: python
from cleveragents.agents.tool import Tool
class WeatherTool(Tool):
def __init__(self, config=None):
super().__init__(
name="weather",
description="Gets weather information for a location",
config=config or {}
)
self.api_key = self.config.get("api_key", "")
async def execute(self, input_data, context=None):
# Parse the location from the input
location = input_data.strip()
# In a real implementation, you would call a weather API here
# For this example, we'll return a mock response
return f"The weather in {location} is sunny with a temperature of 25°C."
# Create an instance of the custom tool
weather_tool = WeatherTool({"api_key": "your-api-key"})
# Use the tool
result = await weather_tool.execute("New York")
# Result: "The weather in New York is sunny with a temperature of 25°C."
Integrating Custom Tools
---------------------
To use custom tools in a tool agent:
.. code-block:: python
from cleveragents.agents.tool import ToolAgent
# Create a tool agent with custom tools
tool_agent = ToolAgent(
name="tools",
config={
"tools": [
{
"name": "weather",
"description": "Gets weather information for a location",
"class": "path.to.WeatherTool",
"config": {
"api_key": "your-api-key"
}
}
]
},
template_renderer=template_renderer
)
# Process a message
response = await tool_agent.process("What's the weather in New York?")
Advanced Routing
==============
CleverAgents supports advanced routing scenarios for complex agent networks.
Conditional Routing and Transformation
--------------------------------------
Route messages based on conditions and transform the message payload using Jinja2 templates.
.. code-block:: yaml
routes:
main_workflow:
- source: input
destination: classifier
transform: "{{ context.get('initial_message', message) }}"
- source: classifier
destination: calculator
condition: "'CALCULATION' in message"
transform: "{{ context.get('initial_message', message) }}"
- source: calculator
destination: responder
transform: |-
The result of {{ context.history[-1].message }} is {{ message }}.
Dynamic Routing
------------
Implement dynamic routing based on message content:
.. code-block:: python
from cleveragents.routing.router import Router
class DynamicRouter(Router):
async def route_message(self, message, source="input", context=None):
# Analyze the message to determine the destination
if "weather" in message.lower():
destination = "weather_agent"
elif "news" in message.lower():
destination = "news_agent"
elif "calculate" in message.lower():
destination = "calculator_agent"
else:
destination = "general_agent"
# Route to the determined destination
if destination in self.agents:
return await self.agents[destination].process(message, context)
else:
return f"No agent found for: {destination}"
Advanced Templates
================
CleverAgents supports advanced template features for complex prompt engineering.
Template Inheritance
-----------------
Create template hierarchies:
.. code-block:: python
from cleveragents.templates.renderer import TemplateRenderer, TemplateEngine
# Create a template renderer
renderer = TemplateRenderer(engine_type=TemplateEngine.JINJA2)
# Register a base template
renderer.register_template(
"base_prompt",
"""
You are a helpful assistant.
User: {message}
Assistant:
"""
)
# Register a specialized template that extends the base
renderer.register_template(
"expert_prompt",
"""
{% extends "base_prompt" %}
{% block preamble %}
You are an expert in {domain}.
{% endblock %}
"""
)
# Render the specialized template
prompt = renderer.render(
"expert_prompt",
{"message": "How does quantum computing work?", "domain": "quantum physics"}
)
Template Includes
--------------
Include templates within other templates:
.. code-block:: python
# Register component templates
renderer.register_template(
"header",
"# {title}\n\n"
)
renderer.register_template(
"footer",
"\n\nRespond in a {tone} tone."
)
# Register a template that includes components
renderer.register_template(
"complete_prompt",
"""
{% include "header" with {"title": "Expert Consultation"} %}
You are an expert in {domain}.
User: {message}
Assistant:
{% include "footer" with {"tone": "professional"} %}
"""
)
# Render the complete template
prompt = renderer.render(
"complete_prompt",
{"message": "How does quantum computing work?", "domain": "quantum physics"}
)
Conditional Templates
-----------------
Use conditions in templates:
.. code-block:: python
# Register a template with conditions
renderer.register_template(
"conditional_prompt",
"""
You are a helpful assistant.
{% if context.user_level == "beginner" %}
Please provide a simple explanation suitable for beginners.
{% elif context.user_level == "intermediate" %}
Please provide a detailed explanation with some technical terms.
{% else %}
Please provide an advanced technical explanation.
{% endif %}
User: {message}
Assistant:
"""
)
# Render with different contexts
beginner_prompt = renderer.render(
"conditional_prompt",
{"message": "How does quantum computing work?", "user_level": "beginner"}
)
expert_prompt = renderer.render(
"conditional_prompt",
{"message": "How does quantum computing work?", "user_level": "expert"}
)
Distributed Agent Networks
========================
CleverAgents can be used to create distributed agent networks across multiple machines.
Agent Network Distribution
-----------------------
Distribute agents across multiple machines:
.. code-block:: python
from cleveragents.network import AgentNetwork
from cleveragents.distribution import RemoteAgent
# Create a network
network = AgentNetwork()
# Add a local agent
network.add_agent(local_agent)
# Add a remote agent
remote_agent = RemoteAgent(
name="remote_assistant",
endpoint="http://remote-server:8000/agent",
api_key="your-api-key"
)
network.add_agent(remote_agent)
# Define connections
router = network.get_router()
router.add_route({
"from": "input",
"to": "local_agent",
"condition": "true"
})
router.add_route({
"from": "local_agent",
"to": "remote_assistant",
"condition": "true"
})
router.add_route({
"from": "remote_assistant",
"to": "output",
"condition": "true"
})
Remote Agent Server
----------------
Create a server for remote agents:
.. code-block:: python
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from cleveragents.network import AgentNetwork
app = FastAPI()
# Create a network with agents
network = AgentNetwork()
class MessageRequest(BaseModel):
message: str
context: dict = None
agent: str
@app.post("/agent")
async def process_message(request: MessageRequest):
try:
# Get the specified agent
agent = network.get_agent(request.agent)
if not agent:
raise HTTPException(status_code=404, detail=f"Agent {request.agent} not found")
# Process the message
response = await agent.process(request.message, request.context)
return {"response": response}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Performance Optimization
======================
Optimize CleverAgents for better performance.
Caching
------
Implement caching to reduce redundant processing:
.. code-block:: python
from cleveragents.agents.base import Agent
from functools import lru_cache
class CachedAgent(Agent):
def __init__(self, name, config, template_renderer):
super().__init__(name, config, template_renderer)
self.cache_size = config.get("cache_size", 100)
self.process_with_cache = lru_cache(maxsize=self.cache_size)(self._process_uncached)
async def process(self, message, context=None):
# Convert context to a hashable form for caching
hashable_context = None
if context:
hashable_context = tuple(sorted((k, str(v)) for k, v in context.items()))
return await self.process_with_cache(message, hashable_context)
async def _process_uncached(self, message, hashable_context):
# Convert hashable context back to a dict
context = None
if hashable_context:
context = {k: v for k, v in hashable_context}
# Actual processing logic
return f"Processed: {message}"
def get_capabilities(self):
return ["cached_processing"]
Batching
-------
Batch process messages for better throughput:
.. code-block:: python
from cleveragents.agents.base import Agent
import asyncio
class BatchAgent(Agent):
def __init__(self, name, config, template_renderer):
super().__init__(name, config, template_renderer)
self.batch_size = config.get("batch_size", 10)
self.batch_timeout = config.get("batch_timeout", 1.0)
self.batch = []
self.batch_lock = asyncio.Lock()
self.batch_event = asyncio.Event()
self.results = {}
self.next_id = 0
# Start the batch processor
asyncio.create_task(self._process_batches())
async def process(self, message, context=None):
async with self.batch_lock:
# Assign an ID to this request
request_id = self.next_id
self.next_id += 1
# Add to the batch
self.batch.append((request_id, message, context))
# Signal that a new item is in the batch
self.batch_event.set()
# Wait for the result
while request_id not in self.results:
await asyncio.sleep(0.1)
# Get and remove the result
result = self.results.pop(request_id)
return result
async def _process_batches(self):
while True:
# Wait for items in the batch
await self.batch_event.wait()
# Wait for more items or timeout
await asyncio.sleep(self.batch_timeout)
# Get the current batch
async with self.batch_lock:
current_batch = self.batch[:self.batch_size]
self.batch = self.batch[self.batch_size:]
# Reset the event if the batch is empty
if not self.batch:
self.batch_event.clear()
# Process the batch
if current_batch:
batch_results = await self._process_batch(current_batch)
# Store the results
for request_id, result in batch_results:
self.results[request_id] = result
async def _process_batch(self, batch):
# Process the batch and return results
# This is where you would implement your batch processing logic
return [(request_id, f"Batch processed: {message}") for request_id, message, _ in batch]
def get_capabilities(self):
return ["batch_processing"]
Parallel Processing
----------------
Process messages in parallel:
.. code-block:: python
from cleveragents.agents.base import Agent
import asyncio
class ParallelAgent(Agent):
def __init__(self, name, config, template_renderer):
super().__init__(name, config, template_renderer)
self.max_parallel = config.get("max_parallel", 10)
self.semaphore = asyncio.Semaphore(self.max_parallel)
async def process(self, message, context=None):
async with self.semaphore:
# Process the message with a limit on parallel executions
return await self._process_message(message, context)
async def _process_message(self, message, context=None):
# Actual processing logic
return f"Processed: {message}"
def get_capabilities(self):
return ["parallel_processing"]
View Git Blame Copy Permalink