Pipecat Integration

pip install pipecat-agenthuman

from agenthuman.video import Agent HumanVideoService
from pipecat.pipeline import Pipeline

# Configure your avatar
transport = Agent HumanVideoService(
    api_key="ah_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    avatar="https://example.com/avatar.jpg",  # URL or base64 image (data:image/...;base64,...)
    room_url="https://your-domain.daily.co/your-room",
    room_token="your-daily-token"
)

# Build your pipeline
pipeline = Pipeline([
    # Your pipeline processors here
    transport
])

# Run the pipeline
await pipeline.run()

import asyncio
from pipecat.pipeline import Pipeline
from pipecat.processors.audio import AudioProcessor
from pipecat.processors.speech import SpeechProcessor
from pipecat_agenthuman import Agent HumanVideoService

async def main():
    # Configure avatar transport
    avatar_transport = Agent HumanVideoService(
        api_key="ah_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxx",
        avatar="https://example.com/avatar.jpg",  # URL or base64 image
        room_url="https://your-domain.daily.co/your-room",
        room_token="your-daily-token",
        room_platform="daily",
        display_name="AI Assistant",
        aspect_ratio="4:3"
    )
    
    # Build voice AI pipeline
    pipeline = Pipeline([
        AudioProcessor(),           # Process incoming audio
        SpeechProcessor(),          # Speech recognition/synthesis
        avatar_transport            # Send to avatar
    ])
    
    # Run the pipeline
    try:
        await pipeline.run()
    except KeyboardInterrupt:
        print("Shutting down...")
    finally:
        await avatar_transport.close()

if __name__ == "__main__":
    asyncio.run(main())

Microphone → AudioProcessor → SpeechProcessor → Agent HumanVideoService → Avatar

from pipecat_agenthuman import Agent HumanVideoService

transport = Agent HumanVideoService(...)

# Listen for events
@transport.on("connected")
async def on_connected():
    print("Avatar connected and ready")

@transport.on("audio_sent")
async def on_audio_sent(samples):
    print(f"Sent {samples} audio samples")

@transport.on("error")
async def on_error(error):
    print(f"Error: {error}")

from daily import Daily

# Create Daily room
daily = Daily(api_key="your-daily-api-key")
room = daily.rooms.create()

# Use room credentials with transport
transport = Agent HumanVideoService(
    api_key="ah_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    avatar="https://example.com/avatar.jpg",  # URL or base64 image
    room_url=room['url'],
    room_token=room['token'],
    room_platform="daily"
)

from livekit import api

# Create LiveKit room
livekit_api = api.LiveKitAPI(
    url="wss://your-livekit-server.com",
    api_key="your-livekit-key",
    api_secret="your-livekit-secret"
)

room = await livekit_api.room.create_room(name="my-room")
token = await livekit_api.create_token(room_name=room.name)

# Use room credentials with transport
transport = Agent HumanVideoService(
    api_key="ah_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    avatar="https://example.com/avatar.jpg",  # URL or base64 image
    room_url=f"wss://your-livekit-server.com/{room.name}",
    room_token=token,
    room_platform="livekit"
)

# Proper cleanup
try:
    await pipeline.run()
finally:
    await transport.close()  # Closes WebSocket and ends session

@transport.on("error")
async def on_error(error):
    if "Session not found" in str(error):
        # Recreate session
        await transport.reconnect()
    else:
        # Log error
        logger.error(f"Avatar error: {error}")

# Configure for optimal quality
transport = Agent HumanVideoService(
    api_key="...",
    avatar="...",  # URL or base64 image
    room_url="...",
    room_token="...",
    aspect_ratio="4:3"
)

# Test connection
try:
    await transport.connect()
    print("Connected successfully")
except Exception as e:
    print(f"Connection failed: {e}")

# Add audio format processor before transport
pipeline = Pipeline([
    AudioFormatProcessor(sample_rate=48000, channels=1),
    avatar_transport
])