Natuurlijk! Hieronder geef ik je een stapsgewijze gids en codevoorbeelden voor het integreren van een aandelenmarktdata-API in jouw LangChain-agent, inclusief tooldefinities, communicatie binnen ketens, en prestatie- en foutafhandelingsrichtlijnen. --- ### 1. Toolinterface definiëren Definieer eerst een toolinterface voor de API, inclusief methoden voor het ophalen van realtime aandelenkoersen en het uitvoeren van portfolio-analyse. ```python from typing import Dict, Any from langchain.tools import BaseTool class StockMarketAPI(BaseTool): name = "StockMarketAPI" description = "Haalt realtime aandelenkoersen op en voert portfolioanalyses uit." def __init__(self, api_key: str): self.api_key = api_key super().__init__() def _call(self, prompt: str) -> str: # Implementatie hangt af van de specifieke API pass def get_realtime_price(self, ticker: str) -> float: # API-aanroep voor realtime prijs # Voorbeeld: gebruik requests om API te benaderen import requests url = f"https://api.aandelenmarkt.com/v1/price/{ticker}" headers = {"Authorization": f"Bearer {self.api_key}"} response = requests.get(url, headers=headers) if response.status_code == 200: data = response.json() return float(data['price']) else: raise Exception(f"API fout bij ophalen prijs voor {ticker}") def analyze_portfolio(self, holdings: Dict[str, float]) -> Dict[str, Any]: # Portfolioanalyse: bereken totale waarde en verdeling total_value = 0.0 detail = {} for ticker, amount in holdings.items(): price = self.get_realtime_price(ticker) value = price * amount total_value += value detail[ticker] = { "amount": amount, "price": price, "value": value } return { "total_value": total_value, "holdings": detail } ``` --- ### 2. Tool in LangChain integreren Maak een instance van deze tool en registreer deze in de agent. ```python from langchain.agents import initialize_agent, AgentType from langchain.chat_models import ChatOpenAI # API sleutel (vervang door je eigen API-sleutel) api_key = "JOUW_API_SLEUTEL" # Instantiateer de tool stock_api_tool = StockMarketAPI(api_key=api_key) # Configureer de agent llm = ChatOpenAI(model="gpt-4", temperature=0) tools = [stock_api_tool] agent = initialize_agent( tools=tools, llm=llm, agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True ) ``` --- ### 3. Tool aanroepen binnen ketens Gebruik de `tools` parameter en de `_call` method binnen je prompt of logica. Bijvoorbeeld: ```python # Voorbeeld: portfolioanalyse opvragen holdings = {"AAPL": 50, "GOOGL": 20} try: portfolio_info = stock_api_tool.analyze_portfolio(holdings) print(f"Totale portefeuillewaarde: {portfolio_info['total_value']}") except Exception as e: print(f"Fout bij ophalen portefeuillegegevens: {e}") ``` Je kunt ook een prompt formuleren die de tool activeert: ```python prompt = ( "Geef een overzicht van mijn portefeuille en de realtime koersen van mijn aandelen." ) # De agent kan de tool aanroepen via de prompt, mits correct geconfigureerd. ``` --- ### 4. Outputbeheer en foutafhandeling Voor effectieve output- en foutafhandeling: - **Fallbacks**: implementeer retries en tijdslimieten. - **Logging**: registreer API-fouten. - **Timeouts**: stel timeouts in requests in. Voorbeeld met retries: ```python import time import requests def get_realtime_price_with_retry(ticker, retries=3, delay=2): for attempt in range(retries): try: url = f"https://api.aandelenmarkt.com/v1/price/{ticker}" response = requests.get(url, headers={"Authorization": f"Bearer {self.api_key}"}, timeout=5) if response.status_code == 200: data = response.json() return float(data['price']) else: raise Exception(f"API fout: status {response.status_code}") except Exception as e: if attempt < retries - 1: time.sleep(delay) else: # Log en fallback print(f"Fout bij ophalen van {ticker}: {e}") return None ``` --- ### 5. Prestatie-eisen (1000 verzoeken/minuut) - **API rate limiting**: controleer de API-limieten en implementeer een rate limiter (bijvoorbeeld via `ratelimit` library). - **Asynchrone verwerking**: gebruik `asyncio` en `aiohttp` voor parallelle API-aanroepen. Voorbeeld met `asyncio` en `aiohttp`: ```python import asyncio import aiohttp async def fetch_price(session, ticker, api_key): url = f"https://api.aandelenmarkt.com/v1/price/{ticker}" headers = {"Authorization": f"Bearer {api_key}"} try: async with session.get(url, headers=headers, timeout=5) as resp: if resp.status == 200: data = await resp.json() return ticker, float(data['price']) else: return ticker, None except Exception: return ticker, None async def get_prices_async(tickers, api_key): semaphore = asyncio.Semaphore(10) # Beperk tot 10 gelijktijdige verzoeken async with aiohttp.ClientSession() as session: tasks = [fetch_price(session, t, api_key) for t in tickers] results = await asyncio.gather(*tasks) return dict(results) # Gebruik: tickers = ["AAPL", "GOOGL", "MSFT", ...] prices = asyncio.run(get_prices_async(tickers, api_key)) ``` --- ### Samenvatting - **Tooldefinitie**: maak duidelijke methoden voor API-interacties. - **Integratie**: registreer tools in de LangChain-agent. - **Aanroepen binnen ketens**: gebruik methoden of prompts om tools te activeren. - **Foutafhandeling**: retries, timeouts, logging. - **Prestaties**: async verwerking, rate limiting, parallelisme. --- Heeft u verdere vragen of wilt u dat ik een complete, werkende codebase voor uw situatie opstel?

Hier is een gedetailleerde implementatiegids voor uw LangChain agent met aandelenmarktintegratie: ## 1. Tool Interface Definitie ```python from langchain.tools import BaseTool from typing import Optional, Type from pydantic import BaseModel, Field import requests import pandas as pd import numpy as np from datetime import datetime class StockDataInput(BaseModel): symbol: str = Field(description="Aandelen symbool (bijv. AAPL, MSFT)") timeframe: Optional[str] = Field("1d", description="Tijdsframe voor data (1d, 1h, 1m)") class PortfolioAnalysisInput(BaseModel): symbols: list = Field(description="Lijst van aandelen symbolen") weights: Optional[list] = Field(None, description="Gewichten voor portfolio") ``` ## 2. Aangepaste Tools Implementatie ```python class StockDataTool(BaseTool): name = "stock_data_fetcher" description = "Haalt realtime aandelenkoersen en historische data op" args_schema: Type[BaseModel] = StockDataInput def _run(self, symbol: str, timeframe: str = "1d"): try: # API endpoint (vervang met uw provider) url = f"https://api.marketdata.com/stock/{symbol}?timeframe={timeframe}" response = requests.get(url, timeout=5) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return f"Fout bij ophalen data: {str(e)}" class PortfolioAnalysisTool(BaseTool): name = "portfolio_analyzer" description = "Voert geavanceerde portfolioanalyse uit" args_schema: Type[BaseModel] = PortfolioAnalysisInput def _run(self, symbols: list, weights: Optional[list] = None): try: portfolio_data = {} for symbol in symbols: data = self.get_stock_data(symbol) portfolio_data[symbol] = data analysis = self.calculate_metrics(portfolio_data, weights) return analysis except Exception as e: return f"Portfolio analyse fout: {str(e)}" def calculate_metrics(self, data, weights): # Implementeer uw analyse logica metrics = { "total_return": self.calculate_total_return(data), "volatility": self.calculate_volatility(data, weights), "sharpe_ratio": self.calculate_sharpe_ratio(data, weights) } return metrics ``` ## 3. Agent Configuratie met Rate Limiting ```python from langchain.agents import AgentType, initialize_agent from langchain.chat_models import ChatOpenAI from ratelimit import limits, sleep_and_retry # Rate limiting decorator @sleep_and_retry @limits(calls=1000, period=60) def limited_api_call(): pass class RateLimitedStockTool(StockDataTool): def _run(self, *args, **kwargs): limited_api_call() return super()._run(*args, **kwargs) # Agent initialisatie llm = ChatOpenAI(temperature=0, model="gpt-4") tools = [RateLimitedStockTool(), PortfolioAnalysisTool()] agent = initialize_agent( tools, llm, agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verbose=True, max_iterations=10 ) ``` ## 4. Geavanceerde Foutafhandeling ```python from tenacity import retry, stop_after_attempt, wait_exponential class RobustStockTool(StockDataTool): @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) def _run_with_retry(self, *args, **kwargs): return super()._run(*args, **kwargs) def _run(self, *args, **kwargs): try: return self._run_with_retry(*args, **kwargs) except Exception as e: return self.fallback_handler(*args, **kwargs, error=e) def fallback_handler(self, symbol, timeframe, error): # Fallback naar cached data of alternatieve bron return f"Fallback data voor {symbol} - Laatste bekende koers: $150.25" ``` ## 5. Prestatie Monitoring ```python from prometheus_client import Counter, Histogram import time # Metrics REQUEST_COUNT = Counter('stock_requests_total', 'Total API requests') REQUEST_LATENCY = Histogram('request_latency_seconds', 'Request latency') class MonitoredStockTool(StockDataTool): def _run(self, *args, **kwargs): start_time = time.time() REQUEST_COUNT.inc() try: result = super()._run(*args, **kwargs) latency = time.time() - start_time REQUEST_LATENCY.observe(latency) return result except Exception as e: latency = time.time() - start_time REQUEST_LATENCY.observe(latency) raise e ``` ## 6. Configuratie Bestanden ```yaml # config/api_config.yaml market_data_api: base_url: "https://api.marketdata.com" api_key: "${API_KEY}" rate_limit: 1000 timeout: 5 retry_attempts: 3 cache: enabled: true ttl: 300 # 5 minuten ``` ## 7. Gebruiksvoorbeeld ```python # Agent gebruik result = agent.run( "Wat is de huidige koers van AAPL en bereken de Sharpe ratio voor mijn portfolio met AAPL, MSFT en GOOGL?" ) # Direct tool gebruik stock_tool = StockDataTool() portfolio_tool = PortfolioAnalysisTool() aapl_data = stock_tool.run({"symbol": "AAPL", "timeframe": "1d"}) portfolio_analysis = portfolio_tool.run({ "symbols": ["AAPL", "MSFT", "GOOGL"], "weights": [0.4, 0.3, 0.3] }) ``` ## Beperkingen en Prestatie-eisen: 1. **Rate Limiting**: Maximaal 1000 requests per minuut 2. **Timeout**: 5 seconden per API call 3. **Retry Mechanism**: 3 pogingen met exponentieel backoff 4. **Cache**: 5 minuten caching voor identieke requests 5. **Fallback**: Gebruik cached data bij API failures ## Benodigde Dependencies: ```bash pip install langchain openai requests ratelimit tenacity prometheus-client pandas numpy ``` Deze implementatie biedt robuuste, schaalbare toolintegratie met geavanceerde foutafhandeling en prestatiemonitoring, specifiek afgestemd op uw aandelenmarkt use case.