Device Monitoring Intelligent Agent¶

An intelligent device monitoring and maintenance agent based on the Qwen3-32B large language model, providing device status analysis, fault diagnosis, preventive maintenance recommendations, and automated report generation.

📋 Table of Contents¶

Features
Use Cases
Quick Start
User Guide
API Reference
Sample Data
Best Practices
FAQ

✨ Features¶

1. 🔍 Real-time Device Status Analysis¶

Multi-dimensional sensor data analysis (temperature, pressure, vibration, energy consumption, etc.)
Anomaly detection and early warning
Health score rating (0-100)
Potential risk identification

2. 📊 Intelligent Historical Log Analysis¶

Error and warning pattern recognition
Critical event extraction
Time series trend analysis
Root cause inference

3. 🔧 Maintenance Record Correlation Analysis¶

Failure frequency statistics
High-frequency fault component identification
Maintenance cost analysis
Preventive maintenance recommendations

4. 🎯 Comprehensive Diagnostic Reports¶

Holistic diagnosis integrating multiple data sources
Structured Markdown reports
Priority-ranked issue lists
Detailed action recommendations

5. 📋 Intelligent Action Plans¶

Priority-based action items
Resource requirement assessment
Execution step planning
Success criteria definition

6. 💬 Interactive Q&A¶

Natural language question answering
Context-aware responses
Professional technical advice

🎯 Use Cases¶

Industrial Manufacturing¶

Production equipment monitoring
Production line fault diagnosis
Equipment maintenance planning

Energy & Power¶

Generator set monitoring
Transformer health management
Distribution equipment operations

Transportation¶

Vehicle fleet management
Railway equipment monitoring
Airport facility maintenance

Building Facilities¶

HVAC systems
Elevator equipment management
Fire protection system monitoring

Data Centers¶

Server monitoring
Cooling system management
UPS equipment maintenance

🚀 Quick Start¶

Requirements¶

Python 3.8+
XPULink API Key (obtain from www.xpulink.ai)

Install Dependencies¶

pip install requests python-dotenv

Configure API Key¶

Method 1: Set environment variable

export XPULINK_API_KEY=your_api_key_here

Method 2: Specify in code

from device_agent import DeviceMonitorAgent

agent = DeviceMonitorAgent(api_key="your_api_key_here")

5-Minute Quick Experience¶

Method 1: Using Jupyter Notebook (Recommended)¶

cd DeviceAgent
jupyter notebook device_agent_example.ipynb

Follow the steps in the notebook.

Method 2: Python Script¶

from device_agent import DeviceMonitorAgent
import json

# Initialize Agent
agent = DeviceMonitorAgent()

# Prepare device data
device_data = {
    "device_id": "PUMP-001",
    "device_type": "Water Pump",
    "temperature": 85.5,
    "pressure": 3.2,
    "vibration": 2.8,
    "power_consumption": 45.2,
    "runtime_hours": 15420,
    "status": "running"
}

# Analyze device status
result = agent.analyze_device_status(device_data)
print(result['analysis'])

Method 3: Command Line Test¶

cd DeviceAgent
python device_agent.py

📖 User Guide¶

1. Device Status Analysis¶

from device_agent import DeviceMonitorAgent
import json

agent = DeviceMonitorAgent()

# Load device data
with open('data/examples/device_data.json', 'r') as f:
    device_data = json.load(f)

# Perform analysis
result = agent.analyze_device_status(device_data)

print(result['analysis'])

Sample Output:

{
  "status": "warning",
  "health_score": 75,
  "key_findings": [
    "Temperature exceeds normal threshold by 5.5°C",
    "Vibration level is high",
    "Power consumption slightly increased"
  ],
  "risks": [
    "Sustained high temperature may damage seals",
    "Increased vibration may cause bearing wear"
  ],
  "summary": "Device is operating normally but showing warning signs. Recommend checking cooling system and bearing condition soon"
}

2. Log Analysis¶

# Load logs
with open('data/examples/device_logs.txt', 'r') as f:
    logs = [line.strip() for line in f.readlines()]

# Analyze logs
result = agent.analyze_logs(
    logs,
    context="Device ID: PUMP-001, Device Type: Centrifugal Pump"
)

print(result['analysis'])

3. Maintenance History Analysis¶

# Load maintenance records
with open('data/examples/maintenance_records.json', 'r') as f:
    maintenance_records = json.load(f)

# Analyze maintenance history
result = agent.analyze_maintenance_history(maintenance_records)

print(result['analysis'])

4. Comprehensive Diagnosis¶

# Integrate all data sources
comprehensive_report = agent.comprehensive_diagnosis(
    device_data=device_data,
    logs=logs[:30],  # Last 30 log entries
    maintenance_records=maintenance_records
)

print(comprehensive_report['report'])

# Export report
agent.export_report(
    comprehensive_report,
    output_path="reports/diagnosis_report.md",
    format="markdown"
)

5. Generate Action Plan¶

# Generate action plan based on diagnosis report
action_plan = agent.generate_action_plan(
    diagnosis_report=comprehensive_report['report'],
    priority="immediate"  # immediate/short_term/medium_term/long_term/all
)

print(action_plan['action_plan'])

6. Interactive Q&A¶

# Ask questions about the device
question = "What is the most critical issue with the current device?"

context = {
    "device_data": device_data,
    "recent_logs": logs[:10]
}

answer = agent.query(question, context=context)
print(answer)

📚 API Reference¶

DeviceMonitorAgent Class¶

Initialization¶

agent = DeviceMonitorAgent(api_key=None, model="qwen3-32b")

Parameters: - api_key (str, optional): XPULink API Key, reads from environment variable if not provided - model (str): Model name to use, defaults to "qwen3-32b"

Main Methods¶

analyze_device_status()¶

Analyze device operating status

result = agent.analyze_device_status(device_data: Dict[str, Any]) -> Dict[str, Any]

Parameters: - device_data: Device data dictionary

Returns: - Dictionary containing analysis results

analyze_logs()¶

Analyze device logs

result = agent.analyze_logs(
    logs: List[str],
    context: Optional[str] = None
) -> Dict[str, Any]

Parameters: - logs: List of log entries - context: Additional context information

Returns: - Log analysis results dictionary

analyze_maintenance_history()¶

Analyze maintenance history records

result = agent.analyze_maintenance_history(
    maintenance_records: List[Dict[str, Any]]
) -> Dict[str, Any]

Parameters: - maintenance_records: List of maintenance records

Returns: - Maintenance history analysis results dictionary

comprehensive_diagnosis()¶

Comprehensive diagnosis

result = agent.comprehensive_diagnosis(
    device_data: Optional[Dict[str, Any]] = None,
    logs: Optional[List[str]] = None,
    maintenance_records: Optional[List[Dict[str, Any]]] = None
) -> Dict[str, Any]

Parameters: - device_data: Current device status data - logs: Historical logs - maintenance_records: Maintenance history records

Returns: - Comprehensive diagnosis report dictionary

generate_action_plan()¶

Generate action plan

result = agent.generate_action_plan(
    diagnosis_report: str,
    priority: str = "all"
) -> Dict[str, Any]

Parameters: - diagnosis_report: Diagnosis report content - priority: Priority filter (immediate/short_term/medium_term/long_term/all)

Returns: - Action plan dictionary

query()¶

Free-form Q&A

answer = agent.query(
    question: str,
    context: Optional[Dict[str, Any]] = None
) -> str

Parameters: - question: User question - context: Relevant context data

Returns: - AI response text

export_report()¶

Export report

path = agent.export_report(
    report_data: Dict[str, Any],
    output_path: str,
    format: str = "markdown"
) -> str

Parameters: - report_data: Report data - output_path: Output file path - format: Output format (markdown/json/txt)

Returns: - Output file path

Convenience Functions¶

quick_diagnosis()¶

Quick diagnosis function

from device_agent import quick_diagnosis

result = quick_diagnosis(device_data, api_key=None)

📁 Sample Data¶

The project includes complete sample data located in the data/examples/ directory:

device_data.json¶

Contains complete device status data, including: - Basic device information - Sensor readings (temperature, pressure, vibration, etc.) - Performance metrics - Alert information - Environmental conditions

device_logs.txt¶

50 realistic device operation logs, including: - Info logs - Warning logs - Error logs - Timestamps

maintenance_records.json¶

8 complete maintenance records, including: - Preventive maintenance - Corrective maintenance - Parts replacement - Cost information - Technical findings

🎓 Best Practices¶

1. Data Preparation¶

Device Data Specification:

device_data = {
    "device_id": "Unique device ID",
    "device_type": "Device type",
    "sensors": {
        "sensor_name": {
            "value": current_value,
            "unit": "unit",
            "threshold_max": max_threshold,
            "threshold_min": min_threshold
        }
    },
    "status": "Operating status",
    # Other relevant fields
}

Recommended Log Format:

YYYY-MM-DD HH:MM:SS [LEVEL] DEVICE-ID: Message

Maintenance Record Format:

{
  "date": "YYYY-MM-DD",
  "type": "preventive/corrective",
  "description": "Description",
  "work_performed": ["Work item 1", "Work item 2"],
  "parts_replaced": [{"part_name": "Name", "cost": cost}],
  "findings": "Findings",
  "recommendations": "Recommendations"
}

2. Cost Control¶

Use temperature parameter to control output randomness (0.3 for diagnosis, 0.7 for creative responses)
Set max_tokens reasonably to limit output length
For large log volumes, only analyze recent critical logs (e.g., last 50 entries)
Use caching to avoid repeatedly analyzing the same data

3. Data Privacy¶

In production environments, ensure sensitive data is anonymized
Don't include personal information or trade secrets in logs
Control access permissions for exported reports

4. Result Validation¶

AI analysis results should be used as decision support tools
Critical decisions need validation by professionals
Establish manual review processes
Track accuracy rate of AI recommendations

5. Integration Recommendations¶

Integration with Monitoring Systems:

# Scheduled task example
import schedule
import time

def monitor_device():
    agent = DeviceMonitorAgent()
    device_data = get_device_data()  # Get data from monitoring system
    result = agent.analyze_device_status(device_data)

    if result['health_score'] < 70:
        send_alert(result)  # Send alert

schedule.every(1).hours.do(monitor_device)

while True:
    schedule.run_pending()
    time.sleep(60)

Integration with Ticketing Systems:

def auto_create_ticket():
    report = agent.comprehensive_diagnosis(...)
    action_plan = agent.generate_action_plan(report['report'], priority="immediate")

    # Auto-create ticket
    create_ticket(
        title=f"Device {device_id} requires urgent maintenance",
        description=action_plan['action_plan'],
        priority="high"
    )

❓ FAQ¶

Q1: What if API calls fail?¶

A: Check the following: 1. Is the API Key configured correctly? 2. Is network connection normal? 3. Is API quota sufficient? 4. Is request timeout setting reasonable (default 60 seconds)?

# Increase timeout
agent = DeviceMonitorAgent()
agent.timeout = 120  # Set to 120 seconds

Q2: How to handle large amounts of historical data?¶

A: Recommend batch processing:

# Only analyze recent critical data
recent_logs = logs[-50:]  # Last 50 entries
recent_maintenance = maintenance_records[:5]  # Last 5 maintenance events

result = agent.comprehensive_diagnosis(
    device_data=device_data,
    logs=recent_logs,
    maintenance_records=recent_maintenance
)

Q3: What if analysis results are inaccurate?¶

A: Try the following methods to improve: 1. Provide more complete device data 2. Add more context information 3. Use lower temperature value (e.g., 0.2) for higher determinism 4. Clearly specify focus points in the question

Q4: How to customize analysis dimensions?¶

A: Can be achieved through free-form Q&A:

custom_analysis = agent.query(
    question="""
    Please analyze the device from the following perspectives:
    1. Energy efficiency
    2. Environmental impact
    3. Operational safety
    """,
    context={"device_data": device_data}
)

Q5: What device types are supported?¶

A: Theoretically supports all industrial equipment, including but not limited to: - Pumps, fans, compressors - Motors, transformers - Machine tools, robots - Conveyors, cranes - Boilers, cooling towers - And more

As long as device operating data can be provided, the agent can perform analysis.

Q6: How to improve analysis speed?¶

A: 1. Reduce input data volume (include only key information) 2. Lower max_tokens setting 3. Use faster models (if available) 4. Implement asynchronous calls

import asyncio
import concurrent.futures

def parallel_analysis(devices):
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = [
            executor.submit(agent.analyze_device_status, device)
            for device in devices
        ]
        results = [f.result() for f in futures]
    return results

Q7: Can report formats be customized?¶

A: Yes, through modifying system prompts or post-processing:

# Method 1: Custom prompt
agent.system_prompt += "\n\nAlways present key metrics in table format."

# Method 2: Post-processing
def format_report(report):
    # Add company logo, formatting, etc.
    return formatted_report

📊 Project Structure¶

DeviceAgent/
├── README.md                           # This document
├── device_agent.py                     # Core Agent implementation
├── device_agent_example.ipynb          # Jupyter Notebook example
├── data/
│   ├── examples/                       # Sample data
│   │   ├── device_data.json           # Device data sample
│   │   ├── device_logs.txt            # Log sample
│   │   └── maintenance_records.json   # Maintenance records sample
│   └── reports/                        # Report output directory (auto-created)
└── tests/                              # Test files (optional)

📝 Development Roadmap¶

Support for multi-device batch analysis
Add trend prediction functionality
Integrate visualization charts
Support multi-language reports
Provide REST API interface
Add Web UI interface

🤝 Contributing¶

Issues and Pull Requests are welcome!

📄 License¶

This project is licensed under the MIT License.

💡 Technical Support¶

For questions or suggestions, please: 1. Check the FAQ section of this document 2. Visit XPULink Official Website for API support 3. Submit a GitHub Issue

Notes: - Please keep your API Key secure and do not commit it to public repositories - AI analysis results are for reference only; critical decisions should be made by professionals - Be mindful of API call costs and use resources reasonably