import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
# Dummy dataset and LangGraph model for demonstration
class DummyDataset(torch.utils.data.Dataset):
def __init__(self, size=1000):
self.size = size
self.data = torch.randn(size, 10)
self.labels = (torch.sum(self.data, dim=1) > 0).long()
def __len__(self):
return self.size
def __getitem__(self, idx):
return self.data[idx], self.labels[idx]
class LangGraphStatefulAgent(nn.Module):
def __init__(self, input_dim=10, hidden_dim=32, output_dim=2, dropout=0.3):
super().__init__()
self.rnn = nn.GRU(input_dim, hidden_dim, batch_first=True)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, x, state=None):
# x shape: (batch, seq_len=1, features)
out, state = self.rnn(x, state)
out = self.dropout(out[:, -1, :])
out = self.fc(out)
return out, state
# Training loop with early stopping
def train_agent():
dataset = DummyDataset()
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
val_dataset = DummyDataset(size=200)
val_loader = DataLoader(val_dataset, batch_size=32)
model = LangGraphStatefulAgent()
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
best_val_acc = 0
patience = 5
patience_counter = 0
for epoch in range(50):
model.train()
total_loss = 0
for data, labels in dataloader:
data = data.unsqueeze(1) # seq_len=1
optimizer.zero_grad()
outputs, _ = model(data)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
total_loss += loss.item() * data.size(0)
avg_loss = total_loss / len(dataloader.dataset)
# Validation
model.eval()
correct = 0
total = 0
with torch.no_grad():
for val_data, val_labels in val_loader:
val_data = val_data.unsqueeze(1)
outputs, _ = model(val_data)
preds = outputs.argmax(dim=1)
correct += (preds == val_labels).sum().item()
total += val_labels.size(0)
val_acc = correct / total * 100
if val_acc > best_val_acc:
best_val_acc = val_acc
patience_counter = 0
else:
patience_counter += 1
if patience_counter >= patience:
break
return avg_loss, best_val_acc
train_loss, val_accuracy = train_agent()
print(f"Training loss: {train_loss:.3f}, Validation accuracy: {val_accuracy:.1f}%")