Semantic Web Revival: RDF and Linked Data in the AI Era
The Semantic Web vision of the early 2000s—machine-readable, interconnected data—found renewed purpose with the rise of AI agents. While JSON-LD and Schema.org became the de facto standard for basic structured data, the full vision of RDF (Resource Description Framework) and Linked Data is experiencing a renaissance as agents demand richer semantic relationships.
Why Agents Need More Than Schema.org
Schema.org provides essential structure but has limitations:
{
"@context": "https://schema.org",
"@type": "Product",
"name": "Wireless Headphones",
"offers": {
"@type": "Offer",
"price": "299.99",
"seller": {
"@type": "Organization",
"name": "AudioStore"
}
}
}
This works for basic information, but agents need richer semantic relationships:
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
@prefix schema: <https://schema.org/> .
@prefix gr: <http://purl.org/goodrelations/v1#> .
@prefix foaf: <http://xmlns.com/foaf/0.1/> .
@prefix agent: <http://example.com/ns/agent#> .
:product123 a schema:Product ;
rdfs:label "Wireless Headphones" ;
schema:manufacturer :manufacturerABC ;
schema:offers :offer456 ;
agent:compatibleWith [
agent:agentCapability "audio_output" ;
agent:requiresProtocol "bluetooth_5_3"
] ;
agent:recommendedFor [
a agent:AgentUseCase ;
rdfs:label "Music Listening" ;
agent:confidence 0.95
] .
:manufacturerABC a schema:Organization ;
rdfs:label "Acme Audio" ;
foaf:homepage <https://acme-audio.com> ;
agent:agentAPI <https://api.acme-audio.com/agents> .
:offer456 a gr:Offering ;
gr:hasPriceSpecification [
a gr:UnitPriceSpecification ;
gr:hasCurrency "USD" ;
gr:hasCurrencyValue "299.99"^^xsd:decimal ;
gr:validThrough "2026-12-31"^^xsd:date
] ;
agent:realTimePricing <https://api.audiostore.com/price/product123> ;
agent:inventoryStatus <https://api.audiostore.com/inventory/product123> .
Implementing Linked Data for Agents
"""
Linked Data server for AI agent consumption
"""
from rdflib import Graph, URIRef, Literal, Namespace
from rdflib.namespace import RDF, RDFS, XSD
from flask import Flask, jsonify, request
import json
app = Flask(__name__)
# Define namespaces
SCHEMA = Namespace("https://schema.org/")
AGENT = Namespace("http://example.com/ns/agent/")
GR = Namespace("http://purl.org/goodrelations/v1#")
class LinkedDataServer:
"""
Serve Linked Data optimized for AI agent consumption
"""
def __init__(self):
self.graph = Graph()
self.bind_namespaces()
def bind_namespaces(self):
"""Bind commonly used namespaces"""
self.graph.bind("schema", SCHEMA)
self.graph.bind("agent", AGENT)
self.graph.bind("gr", GR)
self.graph.bind("rdfs", RDFS)
def add_product(self, product_data: dict):
"""
Add product with rich semantic annotations
"""
product_uri = URIRef(f"https://example.com/product/{product_data['id']}")
# Basic product info
self.graph.add((product_uri, RDF.type, SCHEMA.Product))
self.graph.add((product_uri, RDFS.label, Literal(product_data['name'])))
self.graph.add((product_uri, SCHEMA.description, Literal(product_data['description'])))
# Agent-specific annotations
self.graph.add((product_uri, AGENT.agentDiscoverable, Literal(True)))
self.graph.add((product_uri, AGENT.agentActionable, Literal(True)))
# Link to manufacturer
manufacturer_uri = URIRef(product_data['manufacturer_url'])
self.graph.add((product_uri, SCHEMA.manufacturer, manufacturer_uri))
# Add pricing with GoodRelations
offer_uri = URIRef(f"{product_uri}/offer")
self.graph.add((offer_uri, RDF.type, GR.Offering))
self.graph.add((offer_uri, GR.hasPriceSpecification, self._create_price_spec(product_data['price'], offer_uri)))
# Add real-time endpoints for agents
self.graph.add((offer_uri, AGENT.realTimePrice, URIRef(product_data['price_api'])))
self.graph.add((product_uri, AGENT.realTimeInventory, URIRef(product_data['inventory_api'])))
return product_uri
def _create_price_spec(self, price, offer_uri):
"""Create price specification"""
price_uri = URIRef(f"{offer_uri}/price")
self.graph.add((price_uri, RDF.type, GR.UnitPriceSpecification))
self.graph.add((price_uri, GR.hasCurrencyValue, Literal(price, datatype=XSD.decimal)))
self.graph.add((price_uri, GR.hasCurrency, Literal("USD")))
return price_uri
def get_linked_data(self, resource_uri: str, format: str = "json-ld"):
"""
Get Linked Data in specified format
"""
if format == "json-ld":
return self._serialize_jsonld(resource_uri)
elif format == "turtle":
return self.graph.serialize(format="turtle")
elif format == "ntriples":
return self.graph.serialize(format="nt")
def _serialize_jsonld(self, resource_uri: str):
"""Serialize as JSON-LD with context"""
# Create a subgraph for the resource
relevant_triples = self._get_resource_subgraph(resource_uri)
# Convert to JSON-LD with appropriate context
context = {
"schema": "https://schema.org/",
"agent": "http://example.com/ns/agent/",
"gr": "http://purl.org/goodrelations/v1#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#"
}
return {
"@context": context,
"@graph": self._triples_to_jsonld(relevant_triples)
}
def _get_resource_subgraph(self, uri: str, depth: int = 2):
"""Get relevant triples for a resource (neighbors within depth)"""
# Implementation would traverse the graph
pass
def _triples_to_jsonld(self, triples):
"""Convert RDF triples to JSON-LD structure"""
# Implementation would convert triples
pass
@app.route/.product/<product_id>')
def product_ld(product_id):
"""
Serve product as Linked Data with content negotiation
"""
server = LinkedDataServer()
# Content negotiation based on Accept header
accept_header = request.headers.get('Accept', 'application/ld+json')
if 'application/ld+json' in accept_header or 'application/json' in accept_header:
data = server.get_linked_data(f"https://example.com/product/{product_id}", "json-ld")
return jsonify(data), 200, {'Content-Type': 'application/ld+json'}
elif 'text/turtle' in accept_header:
data = server.get_linked_data(f"https://example.com/product/{product_id}", "turtle")
return data, 200, {'Content-Type': 'text/turtle'}
else:
# Default to JSON-LD
data = server.get_linked_data(f"https://example.com/product/{product_id}", "json-ld")
return jsonify(data)
@app.route('/.well-known/webid')
def webid_profile():
"""
WebID profile for agent discovery
"""
return jsonify({
"@context": "https://www.w3.org/ns/solid/contexts/webid",
"@id": "https://example.com/.well-known/webid",
"agent": {
"endpoints": {
"products": "https://api.example.com/agents/products",
"pricing": "https://api.example.com/agents/pricing",
"inventory": "https://api.example.com/agents/inventory"
},
"capabilities": [
"product_discovery",
"price_lookup",
"inventory_check",
"order_placement"
],
"authentication": "https://example.com/.well-known/oauth-authorization-server"
}
})
Knowledge Graph Integration for Agent Discovery
"""
Knowledge Graph builder for agent-accessible content
"""
from rdflib import Graph, Namespace, Literal, URIRef
from rdfalchemy import rdfSubject, rdfSingle
# Define your organization's knowledge graph
ORG = Namespace("https://example.com/ontology/")
class AgentAccessibleEntity(rdfSubject):
"""
Base class for entities accessible to agents
"""
rdf_type = ORG.AgentResource
agent_endpoint = rdfSingle(ORG.agentEndpoint)
update_frequency = rdfSingle(ORG.updateFrequency)
required_auth = rdfSingle(ORG.requiredAuth)
def to_agent_dict(self):
"""
Convert to agent-friendly representation
"""
return {
"id": str(self.resUri),
"endpoint": self.agent_endpoint,
"auth_requirements": self.required_auth,
"freshness_guarantee": self.update_frequency
}
class ProductKnowledgeGraph:
"""
Build and maintain product knowledge graph
"""
def __init__(self):
self.graph = Graph()
def add_product_with_relationships(self, product_data):
"""
Add product with semantic relationships
"""
product_uri = URIRef(f"https://example.com/product/{product_data['id']}")
# Add product type
self.graph.add((product_uri, RDF.type, SCHEMA.Product))
# Add category relationships (hierarchical)
for category in product_data['categories']:
category_uri = URIRef(f"https://example.com/category/{category}")
self.graph.add((product_uri, SCHEMA.category, category_uri))
# Add complementary products
for related_id in product_data['complementary_products']:
related_uri = URIRef(f"https://example.com/product/{related_id}")
self.graph.add((product_uri, AGENT.complements, related_uri))
# Add compatible products
for compatible_id in product_data['compatible_products']:
compatible_uri = URIRef(f"https://example.com/product/{compatible_id}")
self.graph.add((product_uri, AGENT.isCompatibleWith, compatible_uri))
def query_for_agent(self, sparql_query):
"""
Execute SPARQL query for agent
"""
results = self.graph.query(sparql_query)
return self._format_for_agent(results)
def _format_for_agent(self, results):
"""
Format SPARQL results for agent consumption
"""
formatted = []
for result in results:
formatted.append({var: str(result[var]) for var in result.vars})
return formatted