コラム

HOME
コラム
社内データを活用した意思決定AIエージェントの作り方【2025年版開発ガイド】

2025年8月20日 / 最終更新日 : 2025年8月20日 fujifuji コラム

社内データを活用した意思決定AIエージェントの作り方【2025年版開発ガイド】

意思決定AIエージェントとは

意思決定AIエージェントは、企業内の様々なデータソースから情報を収集・分析し、ビジネス判断をサポートする自動化システムです。従来の単純なチャットボットとは異なり、複数のデータ源を統合して複雑な意思決定プロセスを支援します。

主な機能

データ統合: 売上データ、在庫情報、顧客データなど複数ソースの統合
パターン認識: 過去のデータから傾向やパターンを識別
予測分析: 将来のトレンドや結果を予測
推奨提案: データに基づいた具体的なアクション提案

社内データ活用のメリット

1. データの信頼性向上

社内データは品質管理されており、外部データよりも信頼性が高く、ビジネス固有の文脈を含んでいます。

2. コンプライアンス対応

機密情報を外部に送信せず、GDPR・個人情報保護法などの規制要件に対応できます。

3. カスタマイゼーション

業界特有の専門用語や社内の業務プロセスに最適化されたエージェントを構築できます。

開発に必要な技術スタック

フロントエンド

React/Vue.js: ユーザーインターフェース構築
TypeScript: 型安全性の確保

バックエンド

Python: データ処理・機械学習
FastAPI/Flask: API構築
Node.js: リアルタイム処理

AI・機械学習

LangChain: LLMアプリケーション開発フレームワーク
OpenAI API/Azure OpenAI: 大規模言語モデル
Hugging Face Transformers: オープンソースモデル

データベース

PostgreSQL: 構造化データ
MongoDB: ドキュメントデータ
Vector Database (Pinecone/Chroma): ベクトル検索

実装手順とサンプルコード

Step 1: データ収集システムの構築

まず、社内の複数データソースから情報を収集するシステムを構築します。

import asyncio
import pandas as pd
from typing import Dict, List
import sqlite3

class DataCollector:
    def __init__(self):
        self.sources = {}
    
    async def collect_sales_data(self) -> pd.DataFrame:
        """売上データを収集"""
        conn = sqlite3.connect('sales.db')
        query = "SELECT * FROM sales WHERE date >= date('now', '-30 days')"
        return pd.read_sql_query(query, conn)
    
    async def collect_inventory_data(self) -> pd.DataFrame:
        """在庫データを収集"""
        conn = sqlite3.connect('inventory.db')
        query = "SELECT product_id, stock_level, last_updated FROM inventory"
        return pd.read_sql_query(query, conn)
    
    async def aggregate_data(self) -> Dict:
        """複数ソースのデータを統合"""
        sales_data = await self.collect_sales_data()
        inventory_data = await self.collect_inventory_data()
        
        return {
            'sales': sales_data.to_dict('records'),
            'inventory': inventory_data.to_dict('records'),
            'timestamp': pd.Timestamp.now()
        }

Step 2: AIエージェントクラスの実装

LangChainを使用してAIエージェントを構築します。

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from langchain.memory import ConversationBufferMemory

class DecisionAgent:
    def __init__(self, api_key: str):
        self.llm = OpenAI(openai_api_key=api_key)
        self.memory = ConversationBufferMemory()
        self.data_collector = DataCollector()
        self.tools = self._create_tools()
        self.agent = initialize_agent(
            self.tools, 
            self.llm, 
            agent="conversational-react-description",
            memory=self.memory
        )
    
    def _create_tools(self) -> List[Tool]:
        """エージェント用のツールを定義"""
        return [
            Tool(
                name="sales_analysis",
                description="売上データを分析して傾向を調べる",
                func=self._analyze_sales
            ),
            Tool(
                name="inventory_check",
                description="在庫レベルを確認する",
                func=self._check_inventory
            )
        ]
    
    async def _analyze_sales(self, query: str) -> str:
        """売上分析ツール"""
        data = await self.data_collector.collect_sales_data()
        avg_sales = data['amount'].mean()
        return f"過去30日の平均売上: {avg_sales:.2f}円"
    
    async def _check_inventory(self, product_id: str) -> str:
        """在庫確認ツール"""
        data = await self.data_collector.collect_inventory_data()
        product = data[data['product_id'] == product_id]
        if not product.empty:
            stock = product.iloc[0]['stock_level']
            return f"商品ID {product_id}の在庫: {stock}個"
        return f"商品ID {product_id}が見つかりません"

Step 3: RESTful APIの構築

FastAPIを使用してWebAPIを構築します。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="Decision AI Agent API")

class QueryRequest(BaseModel):
    question: str
    user_id: str

class QueryResponse(BaseModel):
    answer: str
    confidence: float
    sources: List[str]

@app.post("/ask", response_model=QueryResponse)
async def ask_question(request: QueryRequest):
    """質問に対する回答を生成"""
    try:
        agent = DecisionAgent(api_key="your-openai-key")
        result = await agent.agent.arun(request.question)
        
        return QueryResponse(
            answer=result,
            confidence=0.8,
            sources=["sales_db", "inventory_db"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

Step 4: フロントエンド実装（React）

import React, { useState } from 'react';
import axios from 'axios';

const DecisionAgent = () => {
    const [question, setQuestion] = useState('');
    const [response, setResponse] = useState(null);
    const [loading, setLoading] = useState(false);
    
    const handleSubmit = async (e) => {
        e.preventDefault();
        setLoading(true);
        
        try {
            const result = await axios.post('/api/ask', {
                question: question,
                user_id: 'current_user'
            });
            setResponse(result.data);
        } catch (error) {
            console.error('Error:', error);
        } finally {
            setLoading(false);
        }
    };
    
    return (
        <div className="agent-interface">
            <h2>意思決定AIエージェント</h2>
            <form onSubmit={handleSubmit}>
                <textarea 
                    value={question}
                    onChange={(e) => setQuestion(e.target.value)}
                    placeholder="質問を入力してください"
                    rows="4"
                />
                <button type="submit" disabled={loading}>
                    {loading ? '処理中...' : '質問する'}
                </button>
            </form>
            
            {response && (
                <div className="response">
                    <h3>回答</h3>
                    <p>{response.answer}</p>
                    <p>信頼度: {response.confidence * 100}%</p>
                </div>
            )}
        </div>
    );
};

export default DecisionAgent;

セキュリティ対策

1. データ暗号化

from cryptography.fernet import Fernet

class SecureDataHandler:
    def __init__(self):
        self.key = Fernet.generate_key()
        self.cipher = Fernet(self.key)
    
    def encrypt_data(self, data: str) -> bytes:
        return self.cipher.encrypt(data.encode())
    
    def decrypt_data(self, encrypted_data: bytes) -> str:
        return self.cipher.decrypt(encrypted_data).decode()

2. アクセス制御

ロールベースアクセス制御（RBAC）の実装
API認証トークンの使用
データアクセスログの記録

3. データマスキング

機密性の高いデータは適切にマスキングして処理します。

def mask_personal_info(text: str) -> str:
    import re
    # 電話番号のマスキング
    text = re.sub(r'\d{3}-\d{4}-\d{4}', 'XXX-XXXX-XXXX', text)
    # メールアドレスのマスキング
    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 
                  'XXX@XXX.com', text)
    return text

運用・保守のポイント

1. モニタリング

システムの性能とエージェントの回答品質を継続的に監視します。

import logging
from datetime import datetime

class AgentMonitor:
    def __init__(self):
        self.setup_logging()
    
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('agent.log'),
                logging.StreamHandler()
            ]
        )
    
    def log_interaction(self, question: str, answer: str, response_time: float):
        logging.info(f"Q: {question[:50]}... | Response time: {response_time}s")

2. 定期的な再学習

データの変化に対応するため、定期的にモデルを再学習させます。

3. パフォーマンス最適化

データベースクエリの最適化
キャッシュ機能の活用
非同期処理の活用

まとめ

社内データを活用した意思決定AIエージェントの開発には、適切な技術選択とセキュリティ対策が重要です。本記事で紹介したサンプルコードを参考に、自社の要件に合わせてカスタマイズを行ってください。

継続的な改善とモニタリングにより、ビジネス価値の高いAIエージェントを構築できます。

プログラミング学習に関するお役立ちコンテンツ

カテゴリー: コラム

コラム

2025年8月20日

コラム