myxagent 0.2.8

Multi-Modal AI Agent System

xAgent - Multi-Modal AI Agent System

🚀 A powerful and easy-to-use AI Agent system with real-time streaming responses

xAgent provides a complete AI assistant experience with text and image processing, tool execution, HTTP server, web interface, and streaming CLI. Perfect for both quick prototypes and production deployments.

📋 Table of Contents

• 🚀 Quick Start
• 🔧 Installation
• 🌐 HTTP Server
• 🌐 Web Interface
• 💻 Command Line Interface
• 🤖 Python API
• 🔄 Multi-Agent Workflows
• 📚 Documentation
• 🤝 Contributing
• 📄 License

🚀 Quick Start

Get up and running with xAgent in just a few commands:

# Install xAgent
pip install myxagent

# Set your OpenAI API key
export OPENAI_API_KEY=your_openai_api_key

# Start interactive CLI chat
xagent-cli

# Or quickly ask a question
xagent-cli --ask "What's the weather like today?"

That's it!

HTTP Server

The easiest way to deploy xAgent in production is through the HTTP server. Once you start the server, you can interact with your AI agent via HTTP API.

# Start the HTTP server
xagent-server

# Server runs at http://localhost:8010

Chat via HTTP API

curl -X POST "http://localhost:8010/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user123",
    "session_id": "session456", 
    "user_message": "Hello, how are you?"
  }'

Or launch web interface

xagent-web

🔧 Installation

Prerequisites

• Python 3.12+
• OpenAI API Key

Install from PyPI

# Latest version
pip install myxagent

# Upgrade existing installation
pip install --upgrade myxagent

# Using different mirrors
pip install myxagent -i https://pypi.org/simple
pip install myxagent -i https://mirrors.aliyun.com/pypi/simple  # China users

Environment Setup

Create a .env file in your project directory:

# Required
OPENAI_API_KEY=your_openai_api_key

# Redis persistence
REDIS_URL=your_redis_url_with_password

# Observability
LANGFUSE_SECRET_KEY=your_langfuse_key
LANGFUSE_PUBLIC_KEY=your_langfuse_public_key
LANGFUSE_HOST=https://cloud.langfuse.com

# Image upload to S3
AWS_ACCESS_KEY_ID=your_aws_access_key_id
AWS_SECRET_ACCESS_KEY=your_aws_secret_access_key
AWS_REGION=us-east-1
BUCKET_NAME=your_bucket_name

🌐 HTTP Server

Here's how to run the xAgent HTTP server:

Quick Start

# Start with default settings
xagent-server

# Server runs at http://localhost:8010

Basic Configuration

Create agent_config.yaml:

agent:
  name: "MyAgent"
  system_prompt: "You are a helpful assistant"
  model: "gpt-4.1-mini"
  capabilities:
    tools:
      - "web_search"      # Built-in web search
      - "draw_image"      # Built-in image generation (need set aws credentials for image upload)
      - "custom_tool"     # Your custom tools

server:
  host: "0.0.0.0"
  port: 8010

more advanced configurations can be found in Configuration Reference.

Custom Tools

Initialize your project with default toolkit structure:

# Create default config and toolkit structure
xagent-cli --init

This creates:

• config/agent.yaml - Configuration file
• my_toolkit/ - Custom tools directory with examples

Example toolkit structure:

# my_toolkit/__init__.py
from .tools import *

TOOLKIT_REGISTRY = {
    "calculate_square": calculate_square,
    "fetch_weather": fetch_weather
}

# my_toolkit/tools.py
import asyncio
from xagent.utils.tool_decorator import function_tool

@function_tool()
def calculate_square(n: int) -> int:
    """Calculate the square of a number."""
    return n * n

@function_tool()
async def fetch_weather(city: str) -> str:
    """Fetch weather data from an API."""
    # Simulate API call
    await asyncio.sleep(0.5)
    return f"Weather in {city}: 22°C, Sunny"

Start with Custom Config

# Use the generated config and toolkit
xagent-server --config config/agent.yaml --toolkit_path my_toolkit

API Usage

# Simple chat
curl -X POST "http://localhost:8010/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user123",
    "session_id": "session456", 
    "user_message": "Calculate the square of 15"
  }'

# Streaming response
curl -X POST "http://localhost:8010/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user123",
    "session_id": "session456",
    "user_message": "Tell me a story",
    "stream": true
  }'

For advanced configurations including multi-agent systems and structured outputs, see Configuration Reference.

🌐 Web Interface

User-friendly Streamlit chat interface for interactive conversations with your AI agent.

# Start the chat interface with default settings
xagent-web

# With custom agent server URL
xagent-web --agent-server http://localhost:8010

# With custom host and port
xagent-web --host 0.0.0.0 --port 8501 --agent-server http://localhost:8010

💻 Command Line Interface

Interactive CLI with real-time streaming for quick testing and development.

Basic Usage

# Interactive chat mode (default with streaming)
xagent-cli

# Ask a single question
xagent-cli --ask "What is the capital of France?"

# Use custom configuration
xagent-cli --config my_config.yaml --toolkit_path my_toolkit

Interactive Chat Session

$ xagent-cli
╭──────────────────────────────────────────────────────────╮
│                  🤖 Welcome to xAgent CLI!               │
╰──────────────────────────────────────────────────────────╯

📁 Config: Default configuration
🤖 Agent: Agent
🧠 Model: gpt-4o-mini
🛠️  Tools: 1 loaded
🔗 Session: cli_session_d02daf21
⚙️  Status: 🔇 Verbose: Off | 🌊 Stream: On

────────────────────────────────────────────────────────────
🚀 Quick Start:
  • Type your message to chat with the agent
  • Use 'help' to see all available commands
  • Use 'exit', 'quit', or 'bye' to end session
  • Use 'clear' to reset conversation history
  • Use 'stream on/off' to toggle response streaming
────────────────────────────────────────────────────────────

👤 You: Hello, how are you?
🤖 Agent: Hello! I'm doing well, thank you for asking...

👤 You: help

╭─ 📋 Commands ─────────────────────────────────────────────╮
│ exit, quit, bye    Exit the chat session                  │
│ clear              Clear conversation history             │
│ stream on/off      Toggle streaming response mode         │
│ help               Show this help message                 │
╰───────────────────────────────────────────────────────────╯

╭─ 🔧 Built-in Tools ───────────────────────────────────────╮
│  1. web_search                                            │
╰───────────────────────────────────────────────────────────╯

👤 You: exit

╭───────────────────────────────────────╮
│  👋 Thank you for using xAgent CLI!   │
│         See you next time! 🚀         │
╰───────────────────────────────────────╯

CLI Options

Option	Description	Example
--config	Configuration file path	--config my_config.yaml
--toolkit_path	Custom toolkit directory	--toolkit_path my_toolkit
--user_id	User ID for session	--user_id user123
--session_id	Session ID for chat	--session_id session456
--ask	Ask single question and exit	--ask "Hello world"
--init	Create default config and toolkit	--init
--verbose, -v	Enable verbose logging	--verbose

🤖 Python API

Use xAgent directly in your Python applications with full control and customization.

Basic Usage

import asyncio
from xagent.core import Agent

async def main():
    # Create agent
    agent = Agent(
        name="my_assistant",
        system_prompt="You are a helpful AI assistant.",
        model="gpt-4.1-mini"
    )

    # Chat interaction
    response = await agent.chat(
        user_message="Hello, how are you?",
        user_id="user123", 
        session_id="session456"
    )
    print(response)

asyncio.run(main())

Streaming Responses

async def streaming_example():
    agent = Agent()
    
    response = await agent.chat(
        user_message="Tell me a story",
        user_id="user123",
        session_id="session456",
        stream=True
    )
    
    async for chunk in response:
        print(chunk, end="")

asyncio.run(streaming_example())

Adding Custom Tools

import asyncio
import time
import httpx
from xagent.utils.tool_decorator import function_tool
from xagent.core import Agent

# Sync tools - automatically converted to async
@function_tool()
def calculate_square(n: int) -> int:
    """Calculate square of a number."""
    time.sleep(0.1)  # Simulate CPU work
    return n * n

# Async tools - used directly for I/O operations
@function_tool()
async def fetch_weather(city: str) -> str:
    """Fetch weather data from API."""
    async with httpx.AsyncClient() as client:
        await asyncio.sleep(0.5)  # Simulate API call
        return f"Weather in {city}: 22°C, Sunny"

async def main():
    # Create agent with custom tools
    agent = Agent(
        tools=[calculate_square, fetch_weather],
        model="gpt-4.1-mini"
    )
    
    # Agent handles all tools automatically
    response = await agent.chat(
        user_message="Calculate the square of 15 and get weather for Tokyo",
        user_id="user123",
        session_id="session456"
    )
    print(response)

asyncio.run(main())

Structured Outputs

With Pydantic structured outputs, you can:

• Parse and validate an agent’s response into typed data
• Easily extract specific fields
• Ensure the response matches the expected format
• Guarantee type safety in your application
• Reliably chain multi-step tasks using structured data

import asyncio
from pydantic import BaseModel
from xagent.core import Agent
from xagent.tools import web_search

class WeatherReport(BaseModel):
    location: str
    temperature: int
    condition: str
    humidity: int

class Step(BaseModel):
    explanation: str
    output: str

class MathReasoning(BaseModel):
    steps: list[Step]
    final_answer: str


async def get_structured_response():
    
    agent = Agent(model="gpt-4.1-mini", 
                  tools=[web_search], 
                  output_type=WeatherReport) # You can set a default output type here or leave it None
    
    # Request structured output for weather
    weather_data = await agent.chat(
        user_message="what's the weather like in Hangzhou?",
        user_id="user123",
        session_id="session456"
    )
    
    print(f"Location: {weather_data.location}")
    print(f"Temperature: {weather_data.temperature}°F")
    print(f"Condition: {weather_data.condition}")
    print(f"Humidity: {weather_data.humidity}%")


    # Request structured output for mathematical reasoning (overrides output_type)
    reply = await agent.chat(
        user_message="how can I solve 8x + 7 = -23",
        user_id="user123",
        session_id="session456",
        output_type=MathReasoning
    ) # Override output_type for this call
    for index, step in enumerate(reply.steps):
        print(f"Step {index + 1}: {step.explanation} => Output: {step.output}")
    print("Final Answer:", reply.final_answer)

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

Agent as Tool Pattern

import asyncio
from xagent.core import Agent
from xagent.components import MessageStorageLocal
from xagent.tools import web_search

async def agent_as_tool_example():
    # Create specialized agents with message storage
    message_storage = MessageStorageLocal()
    
    researcher_agent = Agent(
        name="research_specialist",
        system_prompt="Research expert. Gather information and provide insights.",
        model="gpt-4.1-mini",
        tools=[web_search],
        message_storage=message_storage
    )
    
    # Convert agent to tool
    research_tool = researcher_agent.as_tool(
        name="researcher",
        description="Research topics and provide detailed analysis"
    )
    
    # Main coordinator agent with specialist tools
    coordinator = Agent(
        name="coordinator",
        tools=[research_tool],
        system_prompt="Coordination agent that delegates to specialists.",
        model="gpt-4.1",
        message_storage=message_storage
    )
    
    # Complex multi-step task
    response = await coordinator.chat(
        user_message="Research renewable energy benefits and write a brief summary",
        user_id="user123",
        session_id="session456"
    )
    print(response)

asyncio.run(agent_as_tool_example())

Persistent Sessions with Redis

import asyncio
from xagent.core import Agent
from xagent.components import MessageStorageRedis

async def chat_with_persistence():
    # Initialize Redis-backed message storage
    message_storage = MessageStorageRedis()
    
    # Create agent with Redis persistence
    agent = Agent(
        name="persistent_agent",
        model="gpt-4.1-mini",
        message_storage=message_storage
    )

    # Chat with automatic message persistence
    response = await agent.chat(
        user_message="Remember this: my favorite color is blue",
        user_id="user123",
        session_id="persistent_session"
    )
    print(response)
    
    # Later conversation - context is preserved in Redis
    response = await agent.chat(
        user_message="What's my favorite color?",
        user_id="user123",
        session_id="persistent_session"
    )
    print(response)

asyncio.run(chat_with_persistence())

you can implement your own message storage by inheriting from MessageStorageBase and implementing the required methods like add_messages, get_messages, etc.

🔄 Multi-Agent Workflows

xAgent enables powerful multi-agent coordination patterns for complex tasks.

Workflow Patterns

Pattern	Use Case	Example
Sequential	Pipeline processing	Research -> Analysis -> Summary
Parallel	Multiple perspectives	Expert panel reviews
Graph	Complex dependencies	A->B, A->C, B&C->D
Hybrid	Multi-stage workflows	Sequential + Parallel + Graph stages

Quick Example

import asyncio
from xagent import Agent
from xagent.multi.workflow import Workflow

async def workflow_example():
    # Create specialized agents with detailed expertise
    market_researcher = Agent(
        name="MarketResearcher",
        system_prompt="""You are a senior market research analyst with 10+ years of experience. 
        Your expertise includes:
        - Industry trend analysis and forecasting
        - Competitive landscape assessment
        - Market size estimation and growth projections
        - Consumer behavior analysis
        - Technology adoption patterns
        
        Always provide data-driven insights with specific metrics, sources, and actionable recommendations.""",
        model="gpt-4o"
    )
    
    data_scientist = Agent(
        name="DataScientist", 
        system_prompt="""You are a senior data scientist specializing in business intelligence and predictive analytics.
        Your core competencies:
        - Statistical analysis and hypothesis testing
        - Predictive modeling and machine learning
        - Data visualization and storytelling
        - Risk assessment and scenario planning
        - Performance metrics and KPI development
        
        Transform raw research into quantitative insights, identify patterns, and build predictive models.""",
        model="gpt-4o"
    )
    
    business_writer = Agent(
        name="BusinessWriter",
        system_prompt="""You are an executive business writer and strategic communications expert.
        Your specializations:
        - Executive summary creation
        - Strategic recommendation development
        - Stakeholder communication
        - Risk and opportunity assessment
        - Implementation roadmap design
        
        Create compelling, actionable business reports that drive decision-making at the C-level.""",
        model="gpt-4o"
    )
    
    financial_analyst = Agent(
        name="FinancialAnalyst",
        system_prompt="""You are a CFA-certified financial analyst with expertise in valuation and investment analysis.
        Your focus areas:
        - Financial modeling and valuation
        - Investment risk assessment
        - ROI and NPV calculations
        - Capital allocation strategies
        - Market opportunity sizing
        
        Provide detailed financial analysis with concrete numbers, projections, and investment recommendations.""",
        model="gpt-4o"
    )
    
    strategy_consultant = Agent(
        name="StrategyConsultant",
        system_prompt="""You are a senior strategy consultant from a top-tier consulting firm.
        Your expertise includes:
        - Strategic planning and execution
        - Business model innovation
        - Competitive strategy development
        - Organizational transformation
        - Change management
        
        Synthesize complex information into clear strategic recommendations with implementation timelines.""",
        model="gpt-4o"
    )
    
    workflow = Workflow()
    
    # Sequential workflow - Research to Analysis to Report Pipeline
    result = await workflow.run_sequential(
        agents=[market_researcher, data_scientist, business_writer],
        task="Analyze the electric vehicle charging infrastructure market opportunity in North America for 2024-2027"
    )
    print("Sequential Pipeline Result:", result.result)
    
    # Parallel workflow - Multiple expert perspectives on same challenge
    result = await workflow.run_parallel(
        agents=[financial_analyst, strategy_consultant, data_scientist],
        task="Evaluate the investment potential and strategic implications of generative AI adoption in enterprise software companies"
    )
    print("Expert Panel Analysis:", result.result)
    
    # Graph workflow - Complex dependency analysis
    dependencies = "MarketResearcher->DataScientist, MarketResearcher->FinancialAnalyst, DataScientist&FinancialAnalyst->StrategyConsultant, StrategyConsultant->BusinessWriter"
    result = await workflow.run_graph(
        agents=[market_researcher, data_scientist, financial_analyst, strategy_consultant, business_writer],
        dependencies=dependencies,
        task="Develop a comprehensive go-to-market strategy for a B2B SaaS startup entering the healthcare analytics space"
    )
    print("Strategic Analysis Result:", result.result)
    
    # Hybrid workflow - Multi-stage comprehensive business analysis
    quality_reviewer = Agent(
        name="QualityReviewer",
        system_prompt="""You are a senior partner-level consultant specializing in quality assurance and risk management.
        Your responsibilities:
        - Strategic recommendation validation
        - Risk identification and mitigation
        - Stakeholder impact assessment
        - Implementation feasibility review
        - Compliance and regulatory considerations
        
        Ensure all recommendations are practical, well-researched, and aligned with business objectives.""",
        model="gpt-4o"
    )
    
    stages = [
        {
            "pattern": "sequential",
            "agents": [market_researcher, financial_analyst],
            "task": "Conduct market and financial analysis for: {original_task}",
            "name": "market_financial_analysis"
        },
        {
            "pattern": "parallel", 
            "agents": [data_scientist, strategy_consultant],
            "task": "Analyze strategic implications and develop data-driven insights based on: {previous_result}",
            "name": "strategic_data_analysis"
        },
        {
            "pattern": "graph",
            "agents": [business_writer, quality_reviewer, strategy_consultant],
            "dependencies": "BusinessWriter->QualityReviewer, StrategyConsultant->QualityReviewer",
            "task": "Create final strategic report with quality validation from: {previous_result}",
            "name": "report_synthesis_validation"
        }
    ]
    
    result = await workflow.run_hybrid(
        task="Develop a digital transformation strategy for a traditional manufacturing company looking to implement IoT and predictive maintenance solutions",
        stages=stages
    )
    print("Comprehensive Strategy Report:", result["final_result"])

asyncio.run(workflow_example())

DSL Syntax

Use intuitive arrow notation for complex dependencies:

# Simple chain
"A->B->C"

# Parallel branches
"A->B, A->C"

# Fan-in pattern
"A->C, B->C"

# Complex dependencies
"A->B, A->C, B&C->D"

For detailed workflow patterns and examples, see Multi-Agent Workflows and Workflow DSL.

📚 Documentation

Core Documentation

• Configuration Reference - Complete YAML configuration guide and examples
• API Reference - Complete API documentation and parameter reference
• Multi-Agent Workflows - Workflow patterns and orchestration examples
• Workflow DSL - Domain-specific language for defining agent dependencies

Examples

• examples/ - Complete usage examples and demos
• config/ - Configuration file templates
• toolkit/ - Custom tool development examples

Architecture Overview

xAgent is built with a modular architecture:

• Core (xagent/core/) - Agent and session management
• Interfaces (xagent/interfaces/) - CLI, HTTP server, and web interface
• Components (xagent/components/) - Message storage and persistence
• Tools (xagent/tools/) - Built-in and custom tool ecosystem
• Multi (xagent/multi/) - Multi-agent coordination and workflows

Key Features

• � Easy to Use - Simple API for quick prototyping
• ⚡ High Performance - Async/await throughout, concurrent tool execution
• 🔧 Extensible - Custom tools, MCP integration, plugin system
• 🌐 Multiple Interfaces - CLI, HTTP API, web interface
• 💾 Persistent - Redis-backed conversation storage
• 🤖 Multi-Agent - Hierarchical agent systems and workflows
• 📊 Observable - Built-in logging and monitoring

**Key Methods:**
- `async chat(user_message, user_id, session_id, **kwargs) -> str | BaseModel`: Main chat interface
- `async __call__(user_message, user_id, session_id, **kwargs) -> str | BaseModel`: Shorthand for chat
- `as_tool(name, description) -> Callable`: Convert agent to tool

**Chat Method Parameters:**
- `user_message`: The user's message (string or Message object)
- `user_id`: User identifier for message storage (default: "default_user")
- `session_id`: Session identifier for message storage (default: "default_session")
- `history_count`: Number of previous messages to include (default: 16)
- `max_iter`: Maximum model call attempts (default: 10)
- `image_source`: Optional image for analysis (URL, path, or base64)
- `output_type`: Pydantic model for structured output
- `stream`: Enable streaming response (default: False)

**Agent Parameters:**
- `name`: Agent identifier (default: "default_agent")
- `system_prompt`: Instructions for the agent behavior
- `model`: OpenAI model to use (default: "gpt-4.1-mini")
- `client`: Custom AsyncOpenAI client instance
- `tools`: List of function tools
- `mcp_servers`: MCP server URLs for dynamic tool loading
- `sub_agents`: List of sub-agent configurations (name, description, server URL)
## 🤝 Contributing

We welcome contributions! Here's how to get started:

### Development Workflow

1. **Fork** the repository
2. **Create** a feature branch: `git checkout -b feature/amazing-feature`  
3. **Commit** your changes: `git commit -m 'Add amazing feature'`
4. **Push** to the branch: `git push origin feature/amazing-feature`
5. **Open** a Pull Request

### Development Guidelines

- Follow PEP 8 standards for code style
- Add tests for new features
- Update documentation as needed
- Use type hints throughout
- Follow conventional commit messages

## 📄 License

This project is licensed under the **MIT License** - see the [LICENSE](LICENSE) file for details.

## 🙏 Acknowledgments

Special thanks to these amazing open source projects:

- **[OpenAI](https://openai.com/)** - GPT models powering our AI
- **[FastAPI](https://fastapi.tiangolo.com/)** - Robust async API framework  
- **[Streamlit](https://streamlit.io/)** - Intuitive web interface
- **[Redis](https://redis.io/)** - High-performance data storage

## 📞 Support & Community

| Resource | Purpose |
|----------|---------|
| **[GitHub Issues](https://github.com/ZJCODE/xAgent/issues)** | Bug reports & feature requests |
| **[GitHub Discussions](https://github.com/ZJCODE/xAgent/discussions)** | Community chat & Q&A |
| **Email: [e-mail]** | Direct support |

---

<div align="center">

**xAgent** - Empowering conversations with AI 🚀

[![GitHub stars](https://img.shields.io/github/stars/ZJCODE/xAgent?style=social)](https://github.com/ZJCODE/xAgent)
[![GitHub forks](https://img.shields.io/github/forks/ZJCODE/xAgent?style=social)](https://github.com/ZJCODE/xAgent)

*Built with ❤️ for the AI community*

</div>