Building a Shopping Assistant with Microsoft Agent Framework

pip install neo4j-agent-memory[microsoft-agent,openai]

docker run -d \
  --name neo4j \
  -p 7474:7474 -p 7687:7687 \
  -e NEO4J_AUTH=neo4j/password \
  neo4j:5

"""Memory configuration for the shopping assistant."""

import os
from pydantic import SecretStr

from neo4j_agent_memory import MemoryClient, MemorySettings
from neo4j_agent_memory.integrations.microsoft_agent import (
    GDSConfig,
    GDSAlgorithm,
    Neo4jMicrosoftMemory,
)

def get_settings() -> MemorySettings:
    """Create memory settings from environment variables."""
    return MemorySettings(
        neo4j={
            "uri": os.getenv("NEO4J_URI", "bolt://localhost:7687"),
            "username": os.getenv("NEO4J_USERNAME", "neo4j"),
            "password": SecretStr(os.getenv("NEO4J_PASSWORD", "password")),
        },
        embedding={
            "provider": "openai",
            "model": "text-embedding-3-small",
            "api_key": SecretStr(os.getenv("OPENAI_API_KEY", "")),
        },
    )

def get_gds_config() -> GDSConfig:
    """Configure GDS algorithms with fallback."""
    return GDSConfig(
        enabled=True,
        use_pagerank_for_ranking=True,
        expose_as_tools=[
            GDSAlgorithm.SHORTEST_PATH,
            GDSAlgorithm.NODE_SIMILARITY,
        ],
        fallback_to_basic=True,  # Use Cypher if GDS not installed
        warn_on_fallback=True,
    )

async def create_memory(session_id: str, user_id: str | None = None) -> Neo4jMicrosoftMemory:
    """Create a memory instance for a session."""
    settings = get_settings()
    gds_config = get_gds_config()

    client = MemoryClient(settings)
    await client.connect()

    return Neo4jMicrosoftMemory(
        memory_client=client,
        session_id=session_id,
        user_id=user_id,
        include_short_term=True,
        include_long_term=True,
        include_reasoning=True,
        max_context_items=15,
        extract_entities=True,
        gds_config=gds_config,
    )

"""Shopping assistant agent."""

import json
from typing import Annotated

from agent_framework import Agent, FunctionTool, Message, tool
from agent_framework.azure import AzureOpenAIResponsesClient

from neo4j_agent_memory.integrations.microsoft_agent import (
    Neo4jMicrosoftMemory,
    create_memory_tools,
    record_agent_trace,
)

SYSTEM_PROMPT = """You are a helpful shopping assistant. Your role is to:

1. Help customers find products that match their needs
2. Learn and remember their preferences (brands, styles, budget)
3. Provide personalized recommendations based on their history
4. Explain how products relate to each other

When customers express preferences, use the remember_preference tool to save them.
When recommending products, explain why they match the customer's preferences.
"""


def get_product_tools(memory: Neo4jMicrosoftMemory) -> list[FunctionTool]:
    """Create callable product tools bound to a memory instance."""
    client = memory.memory_client

    @tool(name="search_products", description="Search for products matching a query")
    async def search_products(
        query: Annotated[str, "Search query describing the product"],
        category: Annotated[str | None, "Optional category filter"] = None,
        max_price: Annotated[float | None, "Optional maximum price"] = None,
    ) -> str:
        """Search products in the catalog."""
        conditions = []
        params: dict = {"query": query}

        if category:
            conditions.append("p.category = $category")
            params["category"] = category
        if max_price:
            conditions.append("p.price <= $max_price")
            params["max_price"] = max_price

        where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""

        cypher = f"""
        MATCH (p:Product)
        {where_clause}
        {"WHERE" if not where_clause else "AND"} (toLower(p.name) CONTAINS toLower($query)
             OR toLower(p.description) CONTAINS toLower($query))
        RETURN p.name as name, p.price as price, p.category as category, p.brand as brand
        LIMIT 5
        """

        result = await client.graph.execute_read(cypher, params)
        products = [dict(r) for r in result]
        return json.dumps({"products": products, "count": len(products)})

    return [search_products]


async def create_agent(memory: Neo4jMicrosoftMemory) -> Agent:
    """Create the shopping assistant agent."""
    chat_client = AzureOpenAIResponsesClient(model="gpt-4")

    # Create memory tools bound to the memory instance
    memory_tools = create_memory_tools(memory, include_gds_tools=bool(memory.gds))

    # Create product tools bound to the memory instance
    product_tools = get_product_tools(memory)

    # Create agent — framework auto-invokes all callable tools
    return chat_client.as_agent(
        name="ShoppingAssistant",
        instructions=SYSTEM_PROMPT,
        tools=memory_tools + product_tools,
        context_providers=[memory.context_provider],
    )

"""Main application entry point."""

import asyncio
import json

from agent_framework import Message

from memory_config import create_memory
from agent import create_agent
from neo4j_agent_memory.integrations.microsoft_agent import record_agent_trace


async def main():
    session_id = "demo-session-1"
    user_id = "demo-user"

    print("Shopping Assistant")
    print("Type 'quit' to exit\n")

    # Create memory and agent
    memory = await create_memory(session_id, user_id)
    agent = await create_agent(memory)

    try:
        while True:
            user_input = input("You: ").strip()
            if user_input.lower() in ("quit", "exit"):
                break
            if not user_input:
                continue

            # Save user message
            await memory.save_message("user", user_input)

            # Run agent — framework auto-invokes tools and returns final response
            response = await agent.run(Message("user", [user_input]))

            # Display response
            if response.text:
                print(f"Assistant: {response.text}\n")
                await memory.save_message("assistant", response.text)

            # Record trace for learning
            await record_agent_trace(
                memory=memory,
                messages=[
                    {"role": "user", "content": user_input},
                    {"role": "assistant", "content": response.text[:200] if response.text else ""},
                ],
                task=user_input,
                outcome="success",
                success=True,
            )

    finally:
        await memory.memory_client.close()


if __name__ == "__main__":
    asyncio.run(main())

"""Load sample product data."""

import asyncio
from neo4j import AsyncGraphDatabase

async def load_sample_data():
    driver = AsyncGraphDatabase.driver(
        "bolt://localhost:7687",
        auth=("neo4j", "password"),
    )

    async with driver.session() as session:
        # Create sample products
        await session.run("""
            UNWIND [
                {id: 'nike-pegasus', name: 'Nike Pegasus 40', price: 130,
                 category: 'Running Shoes', brand: 'Nike'},
                {id: 'adidas-ultraboost', name: 'Adidas Ultraboost 24', price: 190,
                 category: 'Running Shoes', brand: 'Adidas'},
                {id: 'nike-air-max', name: 'Nike Air Max 90', price: 130,
                 category: 'Casual Shoes', brand: 'Nike'},
                {id: 'nb-990', name: 'New Balance 990v6', price: 200,
                 category: 'Running Shoes', brand: 'New Balance'}
            ] AS product
            MERGE (p:Product {id: product.id})
            SET p += product
        """)

        # Create categories and relationships
        await session.run("""
            MATCH (p:Product)
            MERGE (c:Category {name: p.category})
            MERGE (b:Brand {name: p.brand})
            MERGE (p)-[:IN_CATEGORY]->(c)
            MERGE (p)-[:MADE_BY]->(b)
        """)

        # Create similarity relationships
        await session.run("""
            MATCH (p1:Product)-[:IN_CATEGORY]->(c)<-[:IN_CATEGORY]-(p2:Product)
            WHERE p1 <> p2
            MERGE (p1)-[:SIMILAR_TO]-(p2)
        """)

    await driver.close()
    print("Sample data loaded!")

if __name__ == "__main__":
    asyncio.run(load_sample_data())

# Set environment variables
export NEO4J_URI=bolt://localhost:7687
export NEO4J_USERNAME=neo4j
export NEO4J_PASSWORD=password
export OPENAI_API_KEY=sk-your-key

# Load sample data
python load_data.py

# Run the assistant
python main.py

Shopping Assistant
Type 'quit' to exit

You: I'm looking for running shoes
  [Using tool: call_abc123...]
Assistant: I found several running shoes for you:
- Nike Pegasus 40 ($130) - Great for everyday running
- Adidas Ultraboost 24 ($190) - Premium cushioning
- New Balance 990v6 ($200) - Made in USA, premium build

Would you like more details on any of these?

You: I prefer Nike brand
  [Using tool: call_def456...]
Assistant: I've noted your preference for Nike. Based on that, I'd recommend
the Nike Pegasus 40 at $130. It's a versatile running shoe that works great
for daily training. Would you like to know more about it?

You: My budget is under $150
  [Using tool: call_ghi789...]
Assistant: Perfect! With your Nike preference and $150 budget, the Nike
Pegasus 40 at $130 is an excellent choice. It leaves room in your budget
for accessories like socks or insoles. Would you like me to search for
Nike accessories as well?

You: What do you know about my preferences?
  [Using tool: call_jkl012...]
Assistant: Based on our conversation, I've learned:
- **Brand**: You prefer Nike
- **Budget**: Under $150
- **Category**: Running shoes

These preferences help me provide personalized recommendations. Is there
anything else you'd like me to remember?