How to use RetrieveUserProxyAgent in AutoGen
Quick answer
Use
RetrieveUserProxyAgent in AutoGen by instantiating it with your user retrieval logic and integrating it into your agent workflow to handle user data requests via proxy. This agent acts as a proxy layer that fetches user information dynamically during conversations or workflows.PREREQUISITES
Python 3.8+pip install crewai>=0.1.0Basic knowledge of AutoGen frameworkOpenAI API key (for underlying LLM usage)
Setup
Install the crewai package and set your environment variables for API keys. This example assumes you have an OpenAI API key configured in your environment.
pip install crewai>=0.1.0 Step by step
This example demonstrates how to create and use RetrieveUserProxyAgent in AutoGen to retrieve user data through a proxy agent. The agent is initialized with a retrieval function and then queried for user information.
import os
from crewai.autogen import RetrieveUserProxyAgent
# Define a simple user retrieval function
# This simulates fetching user data from a database or API
def fetch_user_data(user_id: str) -> dict:
# Example static user data
users = {
"user123": {"name": "Alice", "email": "alice@example.com"},
"user456": {"name": "Bob", "email": "bob@example.com"}
}
return users.get(user_id, {"error": "User not found"})
# Instantiate the RetrieveUserProxyAgent with the retrieval function
proxy_agent = RetrieveUserProxyAgent(user_retriever=fetch_user_data)
# Use the proxy agent to retrieve user data
user_id = "user123"
result = proxy_agent.retrieve(user_id)
print(f"Retrieved data for {user_id}: {result}") output
Retrieved data for user123: {'name': 'Alice', 'email': 'alice@example.com'} Common variations
You can customize RetrieveUserProxyAgent by using asynchronous retrieval functions or integrating it with different LLMs for enhanced user interaction. For example, use async retrieval for database calls or wrap the agent in a conversational workflow.
import asyncio
from crewai.autogen import RetrieveUserProxyAgent
async def async_fetch_user_data(user_id: str) -> dict:
# Simulate async call
await asyncio.sleep(0.1)
users = {
"user123": {"name": "Alice", "email": "alice@example.com"},
"user456": {"name": "Bob", "email": "bob@example.com"}
}
return users.get(user_id, {"error": "User not found"})
proxy_agent = RetrieveUserProxyAgent(user_retriever=async_fetch_user_data, is_async=True)
async def main():
user_id = "user456"
result = await proxy_agent.retrieve(user_id)
print(f"Async retrieved data for {user_id}: {result}")
asyncio.run(main()) output
Async retrieved data for user456: {'name': 'Bob', 'email': 'bob@example.com'} Troubleshooting
- If
retrievereturns{"error": "User not found"}, verify the user ID exists in your retrieval function. - For async retrieval, ensure you pass
is_async=Truewhen instantiatingRetrieveUserProxyAgent. - If you encounter import errors, confirm
crewaiis installed and up to date.
Key Takeaways
- Instantiate
RetrieveUserProxyAgentwith a user retrieval function to proxy user data requests. - Support both synchronous and asynchronous retrieval by setting
is_asyncaccordingly. - Integrate the proxy agent into AutoGen workflows to dynamically fetch user info during conversations.