--- title: "データベースにデータを保存するAIエージェントをステップごとに構築する" slug: ai-agent-with-a-database date: 2026-02-01T13:03:41+00:00 modified: 2026-08-26T14:03:57+00:00 permalink: https://brightdata.jp/blog/ai/ai-agent-with-a-database type: blog --- [ ブログ ](https://brightdata.jp/blog "ブログ") / [AI](https://brightdata.jp/blog/ai) [AI](https://brightdata.jp/blog/ai) # データベースにデータを保存するAIエージェントをステップごとに構築する 会話履歴をデータベースに完全保存し、エンティティを追跡してリアルタイムウェブデータと統合するスマートAIエージェントの開発方法を学びます。 27 分読 [ ![Arindam Majumder](https://media.brightdata.jp/2025/07/Arindam-Majumder-50x50.jpg) ](https://brightdata.com/blog/authors/arindam-majumder) [Arindam Majumder Technical Writer ](https://brightdata.com/blog/authors/arindam-majumder) ![Build an AI Agent that Saves Data to Database](https://media.brightdata.jp/2025/12/Build-an-AI-Agent-that-Saves-Data-to-Database.png) この記事では、以下のことを学びます： - 会話をデータベースに永続化する本番環境対応のAIエージェントの構築方法 - インテリジェントなデータ抽出とエンティティトラッキングの実装方法 - 自動リカバリを備えた堅牢なエラーハンドリングの作成方法 - Bright Dataのリアルタイムウェブデータでエージェントを強化する方法 それでは始めましょう！ ## ステートレスなAI会話の課題 現在のAIエージェントは通常、ステートレスなシステムとして機能します。各会話を独立したイベントとして扱うため、履歴コンテキストが欠如しており、ユーザーは情報を繰り返し入力しなければなりません。その結果、業務効率が低下し、ユーザーの不満につながります。さらに、企業はパーソナライズやサービス改善のための長期的なデータを活用できません。 データ永続型AIはすべてのインタラクションを構造化されたデータベースに記録することでこの問題を解決します。継続的な記録を保持することで、履歴コンテキストを記憶し、特定のエンティティを経時的に追跡し、過去のインタラクションパターンを活用して一貫したパーソナライズされたユーザー体験を提供できます。 ## 構築するもの：データベース接続型AIエージェントシステム [LangChain](https://docs.brightdata.com/integrations/langchain)とGPT-4でメッセージを処理し、各会話を[PostgreSQL](https://www.postgresql.org/)に保存する本番環境対応のAIエージェントを構築します。リアルタイムでエンティティとインサイトを抽出し、セッション間で完全な会話履歴を保持します。自動リトライシステムによるエラー管理とロギングによる監視も実装します。 このシステムは以下を処理します： - 適切なリレーションシップとインデックスを持つデータベーススキーマ - カスタムデータベースツールを持つLangChainエージェント - 自動的な会話永続化とエンティティ抽出 - データ収集のためのバックグラウンド処理パイプライン - トランザクション管理を備えたエラーハンドリング - 履歴データ取得のためのクエリインターフェース - ウェブインテリジェンスのための[Bright Data](/)とのRAG統合 ## 前提条件 開発環境を以下のように設定してください： - **Python 3.10以上**。最新の非同期機能と型ヒントに必要 - **PostgreSQL 14+**または**SQLite 3.35+**。データ永続化用データベース - **OpenAI APIキー**。GPT-4アクセス用。[OpenAIプラットフォーム](https://platform.openai.com/api-keys)から取得 ![Creating an OpenAI Key](https://media.brightdata.com/2025/12/BJFxjSjWZx.png) - **LangChain**。AIエージェント構築フレームワーク。[ドキュメント](https://python.langchain.com/docs/get_started/introduction)を参照 - **Python仮想環境**。依存関係を分離。[`venv`ドキュメント](https://docs.python.org/3/library/venv.html)を参照 ### 環境セットアップ プロジェクトディレクトリを作成し、依存関係をインストールします： ```none mkdir database-agent cd database-agent python -m venv venv # macOS/Linux: source venv/bin/activate # Windows: venv\\Scripts\\activate pip install langchain langchain-openai sqlalchemy psycopg2-binary python-dotenv pydantic ``` `agent.py`という新しいファイルを作成し、以下のインポートを追加します： ```none import os import json import logging import time from datetime import datetime, timedelta from typing import List, Dict, Any, Optional from queue import Queue from threading import Thread # SQLAlchemy imports from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Float, JSON, ForeignKey, text from sqlalchemy.orm import sessionmaker, relationship, Session, declarative_base from sqlalchemy.pool import QueuePool from sqlalchemy.exc import SQLAlchemyError # LangChain imports from langchain.agents import AgentExecutor, create_openai_functions_agent from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.schema import HumanMessage, AIMessage, SystemMessage # RAG imports from langchain_community.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter import requests # Environment setup from dotenv import load_dotenv load_dotenv() # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) ``` 認証情報を含む`.env`ファイルを作成します： ```none # Database Configuration DATABASE_URL="postgresql://username:password@localhost:5432/agent_db" # Or for SQLite: DATABASE_URL="sqlite:///./agent_data.db" # API Keys OPENAI_API_KEY="your-openai-api-key" # Optional: Bright Data (for Step 7) BRIGHT_DATA_API_KEY="your-bright-data-api-key" # Application Settings AGENT_MODEL="gpt-4-turbo-preview" CONNECTION_POOL_SIZE=5 MAX_RETRIES=3 ``` 必要なもの： - **データベースURL**：PostgreSQLまたはSQLiteの接続文字列 - **OpenAI APIキー**：GPT-4経由のエージェントインテリジェンス用 - **Bright Data APIキー**：オプション。ステップ7のリアルタイムウェブデータ用 ![Creating a BrightData API Key](https://media.brightdata.com/2025/12/B1LbZwsZZx.png) ## データベース接続型AIエージェントの構築 ### ステップ1：データベーススキーマの設計 ユーザー、会話、メッセージ、抽出エンティティのテーブルを設計します。スキーマは外部キーとリレーションシップを使用してデータの整合性を維持します。 ```none Base = declarative_base() class User(Base): """User profile table - stores user information and preferences.""" __tablename__ = 'users' id = Column(Integer, primary_key=True) user_id = Column(String(255), unique=True, nullable=False, index=True) name = Column(String(255)) email = Column(String(255)) preferences = Column(JSON, default={}) created_at = Column(DateTime, default=datetime.utcnow) last_active = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # Relationships conversations = relationship("Conversation", back_populates="user", cascade="all, delete-orphan") def __repr__(self): return f"" class Conversation(Base): """Conversation session table - tracks individual conversation sessions.""" __tablename__ = 'conversations' id = Column(Integer, primary_key=True) conversation_id = Column(String(255), unique=True, nullable=False, index=True) user_id = Column(Integer, ForeignKey('users.id'), nullable=False) title = Column(String(500)) summary = Column(Text) status = Column(String(50), default='active') # active, archived, deleted meta_data = Column(JSON, default={}) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # Relationships user = relationship("User", back_populates="conversations") messages = relationship("Message", back_populates="conversation", cascade="all, delete-orphan") entities = relationship("Entity", back_populates="conversation", cascade="all, delete-orphan") def __repr__(self): return f"" class Message(Base): """Individual message table - stores each message in a conversation.""" __tablename__ = 'messages' id = Column(Integer, primary_key=True) conversation_id = Column(Integer, ForeignKey('conversations.id'), nullable=False, index=True) role = Column(String(50), nullable=False) # user, assistant, system content = Column(Text, nullable=False) tokens = Column(Integer) model = Column(String(100)) meta_data = Column(JSON, default={}) created_at = Column(DateTime, default=datetime.utcnow) # Relationships conversation = relationship("Conversation", back_populates="messages") def __repr__(self): return f"" class Entity(Base): """Extracted entities table - stores named entities extracted from conversations.""" __tablename__ = 'entities' id = Column(Integer, primary_key=True) conversation_id = Column(Integer, ForeignKey('conversations.id'), nullable=False, index=True) entity_type = Column(String(100), nullable=False, index=True) # person, organization, location, etc. entity_value = Column(String(500), nullable=False) context = Column(Text) confidence = Column(Float, default=0.0) meta_data = Column(JSON, default={}) extracted_at = Column(DateTime, default=datetime.utcnow) # Relationships conversation = relationship("Conversation", back_populates="entities") def __repr__(self): return f"" class AgentLog(Base): """Agent operation logs table - stores operational logs for monitoring.""" __tablename__ = 'agent_logs' id = Column(Integer, primary_key=True) conversation_id = Column(String(255), index=True) level = Column(String(50), nullable=False) # INFO, WARNING, ERROR operation = Column(String(255), nullable=False) message = Column(Text, nullable=False) error_details = Column(JSON) execution_time = Column(Float) # in seconds created_at = Column(DateTime, default=datetime.utcnow) def __repr__(self): return f"" ``` スキーマは5つのコアテーブルを定義します。`User`は柔軟なデータのためにJSON設定でプロファイルを保存します。`Conversation`はステータストラッキング付きでセッションを管理します。`Message`はユーザーとアシスタントのメッセージを区別するロール指示子付きで個別のやり取りを保持します。`Entity`は信頼スコア付きで抽出された情報を記録します。`AgentLog`は監視のための操作トラッキングを提供します。外部キーは参照整合性を維持します。頻繁にクエリされるフィールドのインデックスはパフォーマンスを最適化します。`cascade="all, delete-orphan"`設定は親レコードが削除された際に関連レコードをクリーンアップします。 ### ステップ2：データベース接続レイヤーのセットアップ SQLAlchemyでデータベース接続マネージャーを設定します。マネージャーは接続プーリング、ヘルスチェック、信頼性のための自動リトライロジックを処理します。 ```none class DatabaseManager: """ Manages database connections and operations. Features: - Connection pooling for efficient resource usage - Health checks to verify database connectivity - Automatic table creation """ def __init__(self, database_url: str, pool_size: int = 5, max_retries: int = 3): """ Initialize database manager. Args: database_url: Database connection string (e.g., 'sqlite:///./agent_data.db') pool_size: Number of connections to maintain in the pool max_retries: Maximum number of retry attempts for failed operations """ self.database_url = database_url self.max_retries = max_retries # Create engine with connection pooling self.engine = create_engine( database_url, poolclass=QueuePool, pool_size=pool_size, max_overflow=10, pool_pre_ping=True, # Verify connections before using echo=False # Set to True for SQL debugging ) # Create session factory self.SessionLocal = sessionmaker( bind=self.engine, autocommit=False, autoflush=False ) logger.info(f"✓ Database engine created with {pool_size} connection pool") def initialize_database(self): """Create all tables in the database.""" try: Base.metadata.create_all(bind=self.engine) logger.info("✓ Database tables created successfully") except Exception as e: logger.error(f"❌ Failed to create database tables: {e}") raise def get_session(self) -> Session: """Get a new database session for performing operations.""" return self.SessionLocal() def health_check(self) -> bool: """ Check database connectivity. Returns: bool: True if database is healthy, False otherwise """ try: with self.engine.connect() as conn: conn.execute(text("SELECT 1")) logger.info("✓ Database health check passed") return True except Exception as e: logger.error(f"❌ Database health check failed: {e}") return False ``` `DatabaseManager`はSQLAlchemyの接続プーリングを使用して接続を確立します。`pool_size=5`の設定は効率化のために5つの永続的な接続を維持します。`pool_pre_ping`オプションは使用前に接続を検証し、古い接続エラーを防ぎます。リトライメカニズムは失敗した操作を指数バックオフで最大3回試みます。一時的なネットワーク問題に対処します。 ### ステップ3：LangChainエージェントコアの構築 データベースと連携するカスタムツールを持つLangChainを使用してAIエージェントを作成します。エージェントはファンクションコーリングを使用して情報を保存し、会話履歴を取得します。 ```none class DataPersistentAgent: """ AI Agent with database persistence capabilities. This agent: - Remembers conversations across sessions - Saves and retrieves user information - Extracts and stores important entities - Provides personalized responses based on history """ def __init__( self, db_manager: DatabaseManager, model_name: str = "gpt-4-turbo-preview", temperature: float = 0.7 ): """ Initialize the data-persistent agent. Args: db_manager: Database manager instance model_name: LLM model to use (default: gpt-4-turbo-preview) temperature: Model temperature for response generation """ self.db_manager = db_manager self.model_name = model_name # Initialize LLM self.llm = ChatOpenAI( model=model_name, temperature=temperature, openai_api_key=os.getenv("OPENAI_API_KEY") ) # Create tools for agent self.tools = self._create_agent_tools() # Create agent prompt self.prompt = self._create_agent_prompt() # Initialize memory self.memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True ) # Create agent self.agent = create_openai_functions_agent( llm=self.llm, tools=self.tools, prompt=self.prompt ) # Create agent executor self.agent_executor = AgentExecutor( agent=self.agent, tools=self.tools, memory=self.memory, verbose=True, handle_parsing_errors=True, max_iterations=5 ) logger.info(f"✓ Data-persistent agent initialized with {model_name}") def _create_agent_tools(self) -> List[Tool]: """Create custom tools for database operations.""" def save_user_info(user_data: str) -> str: """Save user information to database.""" try: data = json.loads(user_data) session = self.db_manager.get_session() user = session.query(User).filter_by(user_id=data['user_id']).first() if not user: user = User(**data) session.add(user) else: for key, value in data.items(): setattr(user, key, value) session.commit() session.close() return f"✓ User information saved successfully" except Exception as e: logger.error(f"Failed to save user info: {e}") return f"❌ Error saving user info: {str(e)}" def retrieve_user_history(user_id: str) -> str: """Retrieve user's conversation history.""" try: session = self.db_manager.get_session() user = session.query(User).filter_by(user_id=user_id).first() if not user: return "No user found" conversations = session.query(Conversation).filter_by(user_id=user.id).order_by(Conversation.created_at.desc()).limit(5).all() history = [] for conv in conversations: messages = session.query(Message).filter_by(conversation_id=conv.id).all() history.append({ 'conversation_id': conv.conversation_id, 'created_at': conv.created_at.isoformat(), 'message_count': len(messages), 'summary': conv.summary }) session.close() return json.dumps(history, indent=2) except Exception as e: logger.error(f"Failed to retrieve history: {e}") return f"❌ Error retrieving history: {str(e)}" def extract_entities(text: str) -> str: """Extract entities from text and save to database.""" try: entities = [] # Simple keyword extraction (replace with proper NER) keywords = ['important', 'key', 'critical'] for keyword in keywords: if keyword in text.lower(): entities.append({ 'entity_type': 'keyword', 'entity_value': keyword, 'confidence': 0.8 }) return json.dumps(entities, indent=2) except Exception as e: logger.error(f"Failed to extract entities: {e}") return f"❌ Error extracting entities: {str(e)}" tools = [ Tool( name="SaveUserInfo", func=save_user_info, description="Save user information to the database. Input should be a JSON string with user details." ), Tool( name="RetrieveUserHistory", func=retrieve_user_history, description="Retrieve a user's conversation history from the database. Input should be the user_id." ), Tool( name="ExtractEntities", func=extract_entities, description="Extract important entities from text and save to database. Input should be the text to analyze." ) ] return tools def _create_agent_prompt(self) -> ChatPromptTemplate: """Create agent prompt template.""" system_message = """You are a helpful AI assistant with the ability to remember and learn from conversations. You have access to the following tools: - SaveUserInfo: Save user information to remember for future conversations - RetrieveUserHistory: Look up past conversations with a user - ExtractEntities: Extract and save important information from conversations Use these tools to provide personalized, context-aware responses. Always check if you have previous conversations with a user before responding. Be proactive about saving important information for future conversations.""" prompt = ChatPromptTemplate.from_messages([ ("system", system_message), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ]) return prompt def chat(self, user_id: str, message: str, conversation_id: Optional[str] = None) -> Dict[str, Any]: """ Process a chat message and persist to database. This method handles: 1. Creating or retrieving conversations 2. Saving user messages to database 3. Generating agent responses 4. Saving agent responses to database 5. Logging operations for monitoring Args: user_id: Unique identifier for the user message: User's message text conversation_id: Optional conversation ID to continue existing conversation Returns: dict: Contains conversation_id, response, and execution_time """ start_time = datetime.utcnow() try: # Get or create conversation session = self.db_manager.get_session() if conversation_id: conversation = session.query(Conversation).filter_by(conversation_id=conversation_id).first() else: # Create new conversation user = session.query(User).filter_by(user_id=user_id).first() if not user: user = User(user_id=user_id, name=user_id) session.add(user) session.commit() conversation = Conversation( conversation_id=f"conv_{user_id}_{datetime.utcnow().timestamp()}", user_id=user.id, title=message[:100] ) session.add(conversation) session.commit() # Save user message user_message = Message( conversation_id=conversation.id, role="user", content=message, model=self.model_name ) session.add(user_message) session.commit() # Get agent response response = self.agent_executor.invoke({ "input": f"[User ID: {user_id}] {message}" }) # Save assistant message assistant_message = Message( conversation_id=conversation.id, role="assistant", content=response['output'], model=self.model_name ) session.add(assistant_message) session.commit() # Log operation execution_time = (datetime.utcnow() - start_time).total_seconds() log_entry = AgentLog( conversation_id=conversation.conversation_id, level="INFO", operation="chat", message="Chat processed successfully", execution_time=execution_time ) session.add(log_entry) session.commit() # Extract conversation_id before closing session conversation_id_result = conversation.conversation_id session.close() logger.info(f"✓ Chat processed for user {user_id} in {execution_time:.2f}s") return { 'conversation_id': conversation_id_result, 'response': response['output'], 'execution_time': execution_time } except Exception as e: logger.error(f"❌ Error processing chat: {e}") # Log error session = self.db_manager.get_session() error_log = AgentLog( conversation_id=conversation_id or "unknown", level="ERROR", operation="chat", message=str(e), error_details={'exception_type': type(e).__name__} ) session.add(error_log) session.commit() session.close() raise ``` `DataPersistentAgent`はLangChainのファンクションコーリングエージェントをデータベースツールでラップします。`SaveUserInfo`ツールはUserレコードを作成または更新してユーザーデータを永続化します。`RetrieveHistory`ツールはコンテキストを提供するために過去の会話をクエリします。システムプロンプトはエージェントに対して情報の保存と履歴確認を積極的に行うよう指示します。`ConversationBufferMemory`はセッション内の短期コンテキストを維持します。データベースストレージはセッション間の長期永続化を提供します。 ![Data persistent AI agent output](https://media.brightdata.com/2025/12/rJfAh8sbWe.png)### ステップ3.5：データ収集モジュールの作成 会話からデータを抽出・構造化するツールを構築します。コレクターはLLMを使用してサマリーを生成し、設定を抽出し、エンティティを識別します。 ```none class DataCollector: """ Collects and structures data from agent conversations. This module: - Generates conversation summaries - Extracts user preferences from conversation history - Identifies and saves named entities """ def __init__(self, db_manager: DatabaseManager, llm: ChatOpenAI): """ Initialize data collector. Args: db_manager: Database manager instance llm: Language model for text analysis """ self.db_manager = db_manager self.llm = llm logger.info("✓ Data collector initialized") def extract_conversation_summary(self, conversation_id: str) -> str: """ Generate and save conversation summary using LLM. Args: conversation_id: ID of conversation to summarize Returns: str: Generated summary text """ try: session = self.db_manager.get_session() conversation = session.query(Conversation).filter_by(conversation_id=conversation_id).first() if not conversation: return "Conversation not found" messages = session.query(Message).filter_by(conversation_id=conversation.id).all() # Build conversation text conv_text = "\n".join([ f"{msg.role}: {msg.content}" for msg in messages ]) # Generate summary using LLM summary_prompt = f"""Summarize the following conversation in 2-3 sentences, capturing the main topics and outcomes: {conv_text} Summary:""" summary_response = self.llm.invoke([HumanMessage(content=summary_prompt)]) summary = summary_response.content # Update conversation with summary conversation.summary = summary session.commit() session.close() logger.info(f"✓ Generated summary for conversation {conversation_id}") return summary except Exception as e: logger.error(f"Failed to generate summary: {e}") return "" def extract_user_preferences(self, user_id: str) -> Dict[str, Any]: """ Extract and save user preferences from conversation history. Args: user_id: ID of user to analyze Returns: dict: Extracted preferences """ try: session = self.db_manager.get_session() user = session.query(User).filter_by(user_id=user_id).first() if not user: return {} # Get recent conversations conversations = session.query(Conversation).filter_by(user_id=user.id).order_by(Conversation.created_at.desc()).limit(10).all() all_messages = [] for conv in conversations: messages = session.query(Message).filter_by(conversation_id=conv.id).all() all_messages.extend([msg.content for msg in messages if msg.role == "user"]) if not all_messages: return {} # Analyze preferences using LLM analysis_prompt = f"""Analyze the following messages from a user and extract their preferences, interests, and communication style. Messages: {chr(10).join(all_messages[:20])} Return a JSON object with the following structure: {{ "interests": ["interest1", "interest2"], "communication_style": "description", "preferred_topics": ["topic1", "topic2"], "language_preference": "language" }}""" response = self.llm.invoke([HumanMessage(content=analysis_prompt)]) try: # Extract JSON from response content = response.content if '```json' in content: content = content.split('```json')[1].split('```')[0].strip() elif '```' in content: content = content.split('```')[1].split('```')[0].strip() preferences = json.loads(content) # Update user preferences user.preferences = preferences session.commit() logger.info(f"✓ Extracted preferences for user {user_id}") return preferences except json.JSONDecodeError: logger.warning("Failed to parse preferences JSON") return {} finally: session.close() except Exception as e: logger.error(f"Failed to extract preferences: {e}") return {} def extract_entities_with_llm(self, conversation_id: str) -> List[Dict[str, Any]]: """ Extract named entities using LLM. Args: conversation_id: ID of conversation to analyze Returns: list: List of extracted entities """ try: session = self.db_manager.get_session() conversation = session.query(Conversation).filter_by(conversation_id=conversation_id).first() if not conversation: return [] messages = session.query(Message).filter_by(conversation_id=conversation.id).all() conv_text = "\n".join([msg.content for msg in messages]) # Extract entities using LLM entity_prompt = f"""Extract named entities from the following conversation. Identify: - People (PERSON) - Organizations (ORG) - Locations (LOC) - Dates (DATE) - Products (PRODUCT) - Technologies (TECH) Conversation: {conv_text} Return a JSON array of entities with format: [ {{"type": "PERSON", "value": "John Doe", "context": "mentioned as team lead"}}, {{"type": "ORG", "value": "Acme Corp", "context": "customer company"}} ]""" response = self.llm.invoke([HumanMessage(content=entity_prompt)]) try: content = response.content if '```json' in content: content = content.split('```json')[1].split('```')[0].strip() elif '```' in content: content = content.split('```')[1].split('```')[0].strip() entities_data = json.loads(content) # Save entities to database saved_entities = [] for entity_data in entities_data: entity = Entity( conversation_id=conversation.id, entity_type=entity_data['type'], entity_value=entity_data['value'], context=entity_data.get('context', ''), confidence=0.9 # LLM extraction has high confidence ) session.add(entity) saved_entities.append(entity_data) session.commit() session.close() logger.info(f"✓ Extracted {len(saved_entities)} entities from conversation {conversation_id}") return saved_entities except json.JSONDecodeError: logger.warning("Failed to parse entities JSON") return [] except Exception as e: logger.error(f"Failed to extract entities: {e}") return [] ``` `DataCollector`はLLMを使用して会話を分析します。`extract_conversation_summary`メソッドは会話の簡潔なサマリーを作成します。`extract_user_preferences`メソッドはメッセージパターンを分析してユーザーの興味とコミュニケーションスタイルを識別します。`extract_entities_with_llm`メソッドは構造化されたプロンプトを使用して人物、組織、テクノロジーなどの名前付きエンティティを抽出します。抽出されたすべてのデータは将来の参照のためにデータベースに保存されます。 ### ステップ4：スマートデータ処理パイプラインの構築 エージェントをブロックせずにデータ収集を処理するバックグラウンド処理を実装します。パイプラインはワーカースレッドとキューを使用してサマリーとエンティティを処理します。 ```none class DataProcessingPipeline: """ Asynchronous data processing pipeline. This pipeline: - Processes conversations in the background - Generates summaries - Extracts entities without blocking main flow - Updates user preferences periodically """ def __init__(self, db_manager: DatabaseManager, collector: DataCollector, batch_size: int = 10): """ Initialize processing pipeline. Args: db_manager: Database manager instance collector: Data collector for processing operations batch_size: Number of items to process in each batch """ self.db_manager = db_manager self.collector = collector self.batch_size = batch_size # Processing queues self.summary_queue = Queue() self.entity_queue = Queue() self.preference_queue = Queue() # Worker threads self.workers = [] self.running = False logger.info("✓ Data processing pipeline initialized") def start(self): """Start background processing workers.""" self.running = True # Create worker threads summary_worker = Thread(target=self._process_summaries, daemon=True) entity_worker = Thread(target=self._process_entities, daemon=True) preference_worker = Thread(target=self._process_preferences, daemon=True) summary_worker.start() entity_worker.start() preference_worker.start() self.workers = [summary_worker, entity_worker, preference_worker] logger.info("✓ Started 3 background processing workers") def stop(self): """Stop background processing workers.""" self.running = False for worker in self.workers: worker.join(timeout=5) logger.info("✓ Stopped background processing workers") def queue_conversation_for_processing(self, conversation_id: str, user_id: str): """ Add conversation to processing queues. Args: conversation_id: ID of conversation to process user_id: ID of user for preference extraction """ self.summary_queue.put(conversation_id) self.entity_queue.put(conversation_id) self.preference_queue.put(user_id) logger.info(f"✓ Queued conversation {conversation_id} for processing") def _process_summaries(self): """Worker for processing conversation summaries.""" while self.running: try: if not self.summary_queue.empty(): conversation_id = self.summary_queue.get() self.collector.extract_conversation_summary(conversation_id) self.summary_queue.task_done() else: time.sleep(1) except Exception as e: logger.error(f"Error in summary worker: {e}") def _process_entities(self): """Worker for processing entity extraction.""" while self.running: try: if not self.entity_queue.empty(): conversation_id = self.entity_queue.get() self.collector.extract_entities_with_llm(conversation_id) self.entity_queue.task_done() else: time.sleep(1) except Exception as e: logger.error(f"Error in entity worker: {e}") def _process_preferences(self): """Worker for processing user preferences.""" while self.running: try: if not self.preference_queue.empty(): user_id = self.preference_queue.get() self.collector.extract_user_preferences(user_id) self.preference_queue.task_done() else: time.sleep(1) except Exception as e: logger.error(f"Error in preference worker: {e}") def get_queue_status(self) -> Dict[str, int]: """ Get current queue sizes. Returns: dict: Queue sizes for each processing type """ return { 'summary_queue': self.summary_queue.qsize(), 'entity_queue': self.entity_queue.qsize(), 'preference_queue': self.preference_queue.qsize() } ``` `ProcessingPipeline`はデータ収集をメッセージ処理から切り離します。会話が完了すると、すぐに処理されるのではなくキューに追加されます。別々のワーカースレッドがこれらのキューからアイテムを取り出し、バックグラウンドで処理します。これによりデータ収集がエージェントの応答をブロックするのを防ぎます。`daemon=True`設定はメインプログラムが終了するとワーカーも終了することを保証します。キューステータスの監視は処理バックログの追跡に役立ちます。 ![Data processing pipeline Agent](https://media.brightdata.com/2025/12/H1OcCUi-Zx.png)### ステップ5：リアルタイム監視とロギングの追加 エージェントのパフォーマンスを追跡し、エラーを検出し、レポートを生成する監視システムを作成します。モニターはログを分析して運用インサイトを提供します。 ```none class AgentMonitor: """ Real-time monitoring and metrics collection. This module: - Tracks performance metrics - Monitors system health - Generates analytics reports """ def __init__(self, db_manager: DatabaseManager): """ Initialize agent monitor. Args: db_manager: Database manager instance """ self.db_manager = db_manager logger.info("✓ Agent monitor initialized") def get_performance_metrics(self, hours: int = 24) -> Dict[str, Any]: """ Get performance metrics for the specified time period. Args: hours: Number of hours to look back Returns: dict: Performance metrics including operation counts and error rates """ try: session = self.db_manager.get_session() cutoff_time = datetime.utcnow() - timedelta(hours=hours) # Query logs logs = session.query(AgentLog).filter( AgentLog.created_at >= cutoff_time ).all() # Calculate metrics total_operations = len(logs) error_count = len([log for log in logs if log.level == "ERROR"]) avg_execution_time = sum([log.execution_time or 0 for log in logs]) / max(total_operations, 1) # Get conversation counts conversations = session.query(Conversation).filter( Conversation.created_at >= cutoff_time ).count() messages = session.query(Message).join(Conversation).filter( Message.created_at >= cutoff_time ).count() session.close() metrics = { 'time_period_hours': hours, 'total_operations': total_operations, 'error_count': error_count, 'error_rate': error_count / max(total_operations, 1), 'avg_execution_time': avg_execution_time, 'conversations_created': conversations, 'messages_processed': messages } logger.info(f"✓ Generated performance metrics for last {hours} hours") return metrics except Exception as e: logger.error(f"Failed to get performance metrics: {e}") return {} def health_check(self) -> Dict[str, Any]: """ Perform health check. Returns: dict: Health status including database connectivity and error rates """ try: # Check database connectivity db_healthy = self.db_manager.health_check() # Check recent error rate metrics = self.get_performance_metrics(hours=1) recent_errors = metrics.get('error_count', 0) # Determine overall health is_healthy = db_healthy and recent_errors < 10 health_status = { 'status': 'healthy' if is_healthy else 'degraded', 'database_connected': db_healthy, 'recent_errors': recent_errors, 'timestamp': datetime.utcnow().isoformat() } logger.info(f"✓ Health check: {health_status['status']}") return health_status except Exception as e: logger.error(f"Health check failed: {e}") return { 'status': 'unhealthy', 'error': str(e), 'timestamp': datetime.utcnow().isoformat() } ``` `AgentMonitor`はシステム操作への可視性を提供します。`AgentLog`テーブルをクエリすることで、総操作数、エラー率、平均実行時間などのメトリクスを追跡します。`get_metrics`メソッドは設定可能な時間ウィンドウ上の統計を計算します。`get_error_report`メソッドはデバッグのための詳細なエラー情報を取得します。この監視により問題の事前検出が可能になります。高いエラー率はユーザーへの影響が生じる前に調査を促します。 ### ステップ6：クエリインターフェースの構築 保存されたデータを取得・分析するクエリ機能を作成します。インターフェースは会話の検索、エンティティのトラッキング、分析の生成のためのメソッドを提供します。 ```none class DataQueryInterface: """ Interface for querying stored agent data. This module provides methods to: - Query user analytics - Retrieve conversation history - Search for specific information """ def __init__(self, db_manager: DatabaseManager): """ Initialize query interface. Args: db_manager: Database manager instance """ self.db_manager = db_manager logger.info("✓ Query interface initialized") def get_user_analytics(self, user_id: str) -> Dict[str, Any]: """ Get analytics for a specific user. Args: user_id: ID of user to analyze Returns: dict: User analytics including conversation counts and preferences """ try: session = self.db_manager.get_session() user = session.query(User).filter_by(user_id=user_id).first() if not user: return {} # Get conversation count conversation_count = session.query(Conversation).filter_by(user_id=user.id).count() # Get message count message_count = session.query(Message).join(Conversation).filter( Conversation.user_id == user.id ).count() # Get entity count entity_count = session.query(Entity).join(Conversation).filter( Conversation.user_id == user.id ).count() # Get time range first_conversation = session.query(Conversation).filter_by( user_id=user.id ).order_by(Conversation.created_at).first() last_conversation = session.query(Conversation).filter_by( user_id=user.id ).order_by(Conversation.created_at.desc()).first() session.close() analytics = { 'user_id': user_id, 'name': user.name, 'conversation_count': conversation_count, 'message_count': message_count, 'entity_count': entity_count, 'preferences': user.preferences, 'first_interaction': first_conversation.created_at.isoformat() if first_conversation else None, 'last_interaction': last_conversation.created_at.isoformat() if last_conversation else None, 'avg_messages_per_conversation': message_count / max(conversation_count, 1) } logger.info(f"✓ Generated analytics for user {user_id}") return analytics except Exception as e: logger.error(f"Failed to get user analytics: {e}") return {} ``` `QueryInterface`は保存されたデータにアクセスするメソッドを提供します。`get_user_conversations`メソッドはオプションのメッセージ含有付きで完全な会話履歴を取得します。`search_conversations`メソッドはSQLの`ILIKE`演算子を使用してメッセージコンテンツに対して全文検索を実行します。`get_entity_mentions`メソッドは特定のエンティティが言及されたすべての会話を検索します。`get_user_analytics`メソッドはユーザーアクティビティに関する統計を生成します。これらのクエリはダッシュボードの構築、レポートの生成、パーソナライズされた体験の作成を可能にします。 ### ステップ7：Bright Dataのリアルタイムウェブデータを使用したRAGの構築 [Bright Dataのリアルタイムウェブインテリジェンス](/ai/web-access)からRAG機能でデータベース接続型エージェントを強化します。この統合は会話履歴と最新のウェブデータを組み合わせてより良い応答を実現します。 ```none class BrightDataRAGEnhancer: """ Enhance data-persistent agent with Bright Data web intelligence. This module: - Fetches real-time web data from Bright Data - Ingests web data into vector store for RAG - Enhances agent with web-augmented knowledge """ def __init__(self, api_key: str, db_manager: DatabaseManager): """ Initialize RAG enhancer with Bright Data. Args: api_key: Bright Data API key db_manager: Database manager instance """ self.api_key = api_key self.db_manager = db_manager self.base_url = "https://api.brightdata.com" # Initialize vector store for RAG self.embeddings = OpenAIEmbeddings() self.vector_store = Chroma( embedding_function=self.embeddings, persist_directory="./chroma_db" ) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) logger.info("✓ Bright Data RAG enhancer initialized") def fetch_dataset_data( self, dataset_id: str, filters: Optional[Dict[str, Any]] = None, limit: int = 1000 ) -> List[Dict[str, Any]]: """ Fetch data from Bright Data Dataset Marketplace. Args: dataset_id: ID of dataset to fetch filters: Optional filters for data limit: Maximum number of records to fetch Returns: list: Retrieved dataset records """ headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } endpoint = f"{self.base_url}/datasets/v3/snapshot/{dataset_id}" params = { "format": "json", "limit": limit } if filters: params["filter"] = json.dumps(filters) try: response = requests.get(endpoint, headers=headers, params=params) response.raise_for_status() data = response.json() logger.info(f"✓ Retrieved {len(data)} records from Bright Data dataset {dataset_id}") return data except Exception as e: logger.error(f"Failed to fetch Bright Data dataset: {e}") return [] def ingest_web_data_to_rag( self, dataset_records: List[Dict[str, Any]], text_fields: List[str], metadata_fields: Optional[List[str]] = None ) -> int: """ Ingest web data into RAG vector store. Args: dataset_records: Records from Bright Data text_fields: Fields to use as text content metadata_fields: Fields to include in metadata Returns: int: Number of document chunks ingested """ try: documents = [] for record in dataset_records: # Combine text fields text_content = " ".join([ str(record.get(field, "")) for field in text_fields if record.get(field) ]) if not text_content.strip(): continue # Build metadata metadata = { "source": "bright_data", "record_id": record.get("id", "unknown"), "timestamp": datetime.utcnow().isoformat() } if metadata_fields: for field in metadata_fields: if field in record: metadata[field] = record[field] # Split text into chunks chunks = self.text_splitter.split_text(text_content) for chunk in chunks: documents.append({ "content": chunk, "metadata": metadata }) # Add to vector store if documents: texts = [doc["content"] for doc in documents] metadatas = [doc["metadata"] for doc in documents] self.vector_store.add_texts( texts=texts, metadatas=metadatas ) logger.info(f"✓ Ingested {len(documents)} document chunks into RAG") return len(documents) except Exception as e: logger.error(f"Failed to ingest web data to RAG: {e}") return 0 def create_rag_enhanced_agent( self, base_agent: DataPersistentAgent ) -> DataPersistentAgent: """ Enhance existing agent with RAG capabilities. Args: base_agent: Base agent to enhance Returns: DataPersistentAgent: Enhanced agent with RAG tool """ def rag_search(query: str) -> str: """Search both conversation history and web data.""" try: # Retrieve from conversation history session = self.db_manager.get_session() messages = session.query(Message).filter( Message.content.ilike(f'%{query}%') ).order_by(Message.created_at.desc()).limit(5).all() results = [] for msg in messages: results.append({ 'content': msg.content, 'source': 'conversation_history', 'relevance': 0.8 }) session.close() # Retrieve from vector store (web data) try: vector_results = self.vector_store.similarity_search_with_score(query, k=5) for doc, score in vector_results: results.append({ 'content': doc.page_content, 'source': 'web_data', 'relevance': 1 - score }) except Exception as e: logger.error(f"Failed to retrieve from vector store: {e}") if not results: return "No relevant information found." # Format context context_text = "\n\n".join([ f"[{item['source']}] {item['content'][:200]}..." for item in results[:5] ]) return f"Retrieved context:\n{context_text}" except Exception as e: logger.error(f"RAG search failed: {e}") return f"Error performing search: {str(e)}" # Add RAG tool to agent rag_tool = Tool( name="SearchKnowledgeBase", func=rag_search, description="Search both conversation history and real-time web data for relevant information. Input should be a search query." ) base_agent.tools.append(rag_tool) # Recreate agent with new tools base_agent.agent = create_openai_functions_agent( llm=base_agent.llm, tools=base_agent.tools, prompt=base_agent.prompt ) base_agent.agent_executor = AgentExecutor( agent=base_agent.agent, tools=base_agent.tools, memory=base_agent.memory, verbose=True, handle_parsing_errors=True, max_iterations=5 ) logger.info("✓ Enhanced agent with RAG capabilities") return base_agent ``` `BrightDataEnhancer`はリアルタイムウェブデータをエージェントに統合します。`fetch_dataset`メソッドはBright Dataのマーケットプレイスから構造化データを取得します。`ingest_to_rag`メソッドはこのデータを処理してチャンク化し、セマンティック検索のためにChromaベクターデータベースに保存します。`retrieve_context`メソッドはハイブリッド検索を実行し、データベース履歴とベクター類似検索を組み合わせます。`create_rag_tool`メソッドはこの機能をエージェントが使用するLangChainツールとしてパッケージ化します。`enhance_agent`メソッドはこのRAG機能を既存のエージェントに追加し、内部会話履歴と最新の外部データの両方を使用して質問に答えられるようにします。 ## 完全なデータ永続型エージェントシステムの実行 すべてのコンポーネントを組み合わせて機能的なシステムを作成します。 ```none def main(): """Main execution flow demonstrating all components working together.""" print("=" * 60) print("Data-Persistent AI Agent System - Initialization") print("=" * 60) # Step 1: Initialize database print("\n[Step 1] Setting up database connection...") db_manager = DatabaseManager( database_url=os.getenv("DATABASE_URL"), pool_size=5, max_retries=3 ) db_manager.initialize_database() # Step 2: Initialize core agent print("\n[Step 2] Building AI agent core...") agent = DataPersistentAgent( db_manager=db_manager, model_name=os.getenv("AGENT_MODEL", "gpt-4-turbo-preview") ) # Step 3: Initialize data collector print("\n[Step 3] Creating data collection module...") collector = DataCollector(db_manager, agent.llm) # Step 4: Initialize processing pipeline print("\n[Step 4] Implementing data processing pipeline...") pipeline = DataProcessingPipeline(db_manager, collector) pipeline.start() # Step 5: Initialize monitoring print("\n[Step 5] Adding monitoring and logging...") monitor = AgentMonitor(db_manager) # Step 6: Initialize query interface print("\n[Step 6] Building query interface...") query_interface = DataQueryInterface(db_manager) # Step 7: Optional Bright Data RAG Enhancement print("\n[Step 7] RAG Enhancement (Optional)...") bright_data_key = os.getenv("BRIGHT_DATA_API_KEY") if bright_data_key and bright_data_key != "your-bright-data-api-key": print("Fetching real-time web data from Bright Data...") enhancer = BrightDataRAGEnhancer(bright_data_key, db_manager) # Example: Fetch and ingest web data web_data = enhancer.fetch_dataset_data( dataset_id="example_dataset_id", limit=100 ) if web_data: enhancer.ingest_web_data_to_rag( dataset_records=web_data, text_fields=["title", "content", "description"], metadata_fields=["url", "published_date"] ) # Enhance agent with RAG agent = enhancer.create_rag_enhanced_agent(agent) print("✓ Agent enhanced with Bright Data RAG capabilities") else: print("⚠️ Bright Data API key not found - skipping web data integration") print("\n" + "=" * 60) print("Demo Conversations") print("=" * 60) # Demo user interactions test_user = "demo_user_001" # First conversation print("\n📝 Conversation 1:") response1 = agent.chat( user_id=test_user, message="Hi! I'm interested in learning about machine learning." ) print(f"Agent: {response1['response']}\n") # Queue for processing pipeline.queue_conversation_for_processing( response1['conversation_id'], test_user ) # Second conversation print("📝 Conversation 2:") response2 = agent.chat( user_id=test_user, message="Help me understand neural networks?", conversation_id=response1['conversation_id'] ) print(f"Agent: {response2['response']}\n") # Wait for background processing print("⏳ Processing data in background...") time.sleep(5) print("\n" + "=" * 60) print("Analytics & Monitoring") print("=" * 60) # Get performance metrics metrics = monitor.get_performance_metrics(hours=1) print(f"\n📊 Performance Metrics:") print(f" - Total operations: {metrics.get('total_operations', 0)}") print(f" - Error rate: {metrics.get('error_rate', 0):.2%}") print(f" - Avg execution time: {metrics.get('avg_execution_time', 0):.2f}s") print(f" - Conversations created: {metrics.get('conversations_created', 0)}") print(f" - Messages processed: {metrics.get('messages_processed', 0)}") # Get user analytics analytics = query_interface.get_user_analytics(test_user) print(f"\n👤 User Analytics:") print(f" - Conversation count: {analytics.get('conversation_count', 0)}") print(f" - Message count: {analytics.get('message_count', 0)}") print(f" - Entity count: {analytics.get('entity_count', 0)}") print(f" - Avg messages/conversation: {analytics.get('avg_messages_per_conversation', 0):.1f}") # Health check health = monitor.health_check() print(f"\n🏥 System Health: {health['status']}") # Queue status queue_status = pipeline.get_queue_status() print(f"\n📋 Processing Queues:") print(f" - Summary queue: {queue_status['summary_queue']}") print(f" - Entity queue: {queue_status['entity_queue']}") print(f" - Preference queue: {queue_status['preference_queue']}") # Stop pipeline pipeline.stop() print("\n" + "=" * 60) print("Data-Persistent Agent System - Complete") print("=" * 60) print("\n✓ All data persisted to database") print("✓ Background processing completed") print("✓ System ready for production use") if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\n\n⚠️ Shutting down gracefully...") except Exception as e: logger.error(f"System error: {e}") import traceback traceback.print_exc() ``` データベース接続型エージェントシステムを実行します： ```none python agent.py ``` システムは完全なワークフローを実行します。データベースを初期化してすべてのテーブルを作成します。データベースツールを持つLangChainエージェントをセットアップします。処理のためのバックグラウンドワーカーを起動します。デモ会話を処理してデータベースに保存します。バックグラウンドでエンティティを抽出してサマリーを生成します。リアルタイムの分析とメトリクスを表示します。 各コンポーネントが初期化されてデータを処理する際に詳細なロギングが表示されます。エージェントはすべてのメッセージを保存し、インサイトを抽出し、完全な会話コンテキストを維持します。 ![Building an AI Agent that Saves Data to Database Demo](https://media.brightdata.com/2025/12/cursorful-video-1764615164579-ezgif.com-video-to-gif-converter201.gif)## 実用的なユースケース ### 1. 完全な履歴を持つカスタマーサポート ```none # Agent retrieves past interactions support_agent = DataPersistentAgent(db_manager) response = support_agent.chat( user_id="customer_123", message="I'm still having that connection issue" ) # Agent sees previous conversations about connection problems ``` ### 2. 学習機能を持つパーソナルAIアシスタント ```none # Agent learns preferences over time query_interface = QueryInterface(db_manager) analytics = query_interface.get_user_analytics("user_456") # Shows interaction patterns, preferences, common topics ``` ### 3. ナレッジベースを持つリサーチアシスタント ```none # Combine conversation history with web data enhancer = BrightDataEnhancer(api_key, db_manager) enhancer.ingest_to_rag(research_data, ["title", "abstract", "content"]) agent = enhancer.enhance_agent(agent) # Agent references both past discussions and latest research ``` ## メリットのまとめ 機能データベースなしデータベース永続化ありメモリ再起動時に消失永続的なストレージパーソナライゼーションなし完全な履歴に基づく分析不可能完全なインタラクションデータエラーリカバリ手動介入自動リトライとロギングスケーラビリティ単一インスタンス共有状態のマルチインスタンスインサイトセッション後に消失抽出・追跡済み## まとめ これで、会話をデータベースに永続化する本番環境対応のAIエージェントシステムが完成しました。このシステムはすべてのインタラクションを保存し、エンティティとインサイトを抽出し、完全な会話履歴を維持し、自動エラーリカバリを備えた監視を提供します。 安全なアクセスのためのユーザー認証の追加、分析を可視化するダッシュボードの構築、セマンティック検索のためのエンベディングの実装、統合のためのAPIエンドポイントの作成、スケーラビリティのためのDockerでのデプロイなどで強化できます。モジュラー設計により、特定のニーズに合わせた簡単なカスタマイズが可能です。 [高度なAIエージェントパターン](/ai/agent-showcase)と[Bright Dataのウェブインテリジェンスインフラ](/ai)でさらなる機能を探索してください。 [無料アカウントを作成](/?hs_signup=1)して、記憶・学習するインテリジェントなシステムの構築を始めましょう。 お問い合わせ無料トライアル![google social icon](/wp-content/themes/brightdata/assets/images/ic_google.svg) 目次 日本企業向けデータソリューション 日本のトップ企業が信頼するスケーラブルなウェブデータプラットフォーム。日本語24時間サポート、GDPR・個人情報保護法準拠、日本企業20社以上の導入実績。 [詳細を見る](https://brightdata.jp/solutions/data-solutions-for-japanese-companies "詳細を見る") [ ](https://news.ycombinator.com/submitlink?t=%E3%83%87%E3%83%BC%E3%82%BF%E3%83%99%E3%83%BC%E3%82%B9%E3%81%AB%E3%83%87%E3%83%BC%E3%82%BF%E3%82%92%E4%BF%9D%E5%AD%98%E3%81%99%E3%82%8BAI%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E3%82%B9%E3%83%86%E3%83%83%E3%83%97%E3%81%94%E3%81%A8%E3%81%AB%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&u=https://brightdata.jp/blog/ai/ai-agent-with-a-database) [ ](https://www.linkedin.com/shareArticle?mini=true&title=%E3%83%87%E3%83%BC%E3%82%BF%E3%83%99%E3%83%BC%E3%82%B9%E3%81%AB%E3%83%87%E3%83%BC%E3%82%BF%E3%82%92%E4%BF%9D%E5%AD%98%E3%81%99%E3%82%8BAI%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E3%82%B9%E3%83%86%E3%83%83%E3%83%97%E3%81%94%E3%81%A8%E3%81%AB%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&url=https://brightdata.jp/blog/ai/ai-agent-with-a-database) [ ](http://www.reddit.com/submit?title=%E3%83%87%E3%83%BC%E3%82%BF%E3%83%99%E3%83%BC%E3%82%B9%E3%81%AB%E3%83%87%E3%83%BC%E3%82%BF%E3%82%92%E4%BF%9D%E5%AD%98%E3%81%99%E3%82%8BAI%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E3%82%B9%E3%83%86%E3%83%83%E3%83%97%E3%81%94%E3%81%A8%E3%81%AB%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&url=https://brightdata.jp/blog/ai/ai-agent-with-a-database) ## あなたは下記にもご興味がおありかもしれません [ ![OpenHuman with Bright Data](https://media.brightdata.jp/2026/09/OpenHuman-with-Bright-Data.png) ](https://brightdata.jp/blog/ai/openhuman-with-bright-data "Bright Data CLIを通じたOpenHumanにおける本番対応ウェブアクセス") [AI ![Antonello Zanini](https://media.brightdata.jp/2022/12/Antonello-Zanini-2-50x50.jpg) Antonello Zanini Technical Writer ### Bright Data CLIを通じたOpenHumanにおける本番対応ウェブアクセス Bright Data CLIをOpenHumanに統合して、AIエージェント向けの本番対応ウェブアクセスとデータ収集を実現します。 09-Sep-2026 3 分読 ](https://brightdata.jp/blog/ai/openhuman-with-bright-data) [ ![Multimodal Web Scraping with MiniMax](https://media.brightdata.jp/2026/09/Multimodal-Web-Scraping-with-MiniMax.png) ](https://brightdata.jp/blog/%e3%82%a6%e3%82%a7%e3%83%96%e3%83%87%e3%83%bc%e3%82%bf/multimodal-web-scraping-with-minimax "MiniMaxによるマルチモーダルウェブスクレイピング") [ウェブデータ ![Antonello Zanini](https://media.brightdata.jp/2022/12/Antonello-Zanini-2-50x50.jpg) Antonello Zanini Technical Writer ### MiniMaxによるマルチモーダルウェブスクレイピング Bright Data Web UnlockerとMiniMax M3ビジョンを組み合わせて、画像やウェブページのスクリーンショットから構造化データを抽出します。 09-Sep-2026 1 分読 ](https://brightdata.jp/blog/%e3%82%a6%e3%82%a7%e3%83%96%e3%83%87%e3%83%bc%e3%82%bf/multimodal-web-scraping-with-minimax) [ ![The 10 Best CLI Tools for Codex in 2026](https://media.brightdata.jp/2026/08/The-10-Best-CLI-Tools-for-Codex-in-2026.png) ](https://brightdata.jp/blog/ai/best-cli-tools-for-codex "2026年版 Codex向けCLIツール10選 – テスト＆ランキング") [AI ![Daniel Shashko](https://media.brightdata.jp/2022/04/Daniel-Shashko-2-50x50.png) Daniel Shashko Web Data & AI Expert ### 2026年版 Codex向けCLIツール10選 – テスト＆ランキング Codexをより速く、より有能にする10のCLIツール。サンドボックス内から実際のウェブアクセスを提供するBright Data CLIから始まります。 06-Sep-2026 4 分読 ](https://brightdata.jp/blog/ai/best-cli-tools-for-codex)