Tutorial 07: Building AI Agents with Semantic Kernel¶

import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

def create_kernel() -> Kernel:
    kernel = Kernel()
    kernel.add_service(
        AzureChatCompletion(
            service_id="gpt54",
            deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT"],
            endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
            api_key=os.environ["AZURE_OPENAI_API_KEY"],
        )
    )
    return kernel

if __name__ == "__main__":
    k = create_kernel()
    print(f"Kernel created with services: {list(k.services.keys())}")

from semantic_kernel.functions import kernel_function
from azure.identity import DefaultAzureCredential
from azure.storage.filedatalake import DataLakeServiceClient
import os

class DataQueryPlugin:
    def __init__(self):
        self.credential = DefaultAzureCredential()
        self.account = os.environ["STORAGE_ACCOUNT_NAME"]

    @kernel_function(
        name="list_gold_tables",
        description="List all available tables in the Gold layer of the data lake."
    )
    def list_gold_tables(self) -> str:
        service = DataLakeServiceClient(
            account_url=f"https://{self.account}.dfs.core.windows.net",
            credential=self.credential,
        )
        fs = service.get_file_system_client("gold")
        tables = [p.name for p in fs.get_paths() if p.is_directory]
        return f"Available Gold tables: {', '.join(tables)}"

    @kernel_function(
        name="get_table_schema",
        description="Get the schema (column names) of a Gold layer table."
    )
    def get_table_schema(self, table_name: str) -> str:
        service = DataLakeServiceClient(
            account_url=f"https://{self.account}.dfs.core.windows.net",
            credential=self.credential,
        )
        fs = service.get_file_system_client("gold")
        paths = list(fs.get_paths(path=table_name, max_results=5))
        files = [p.name for p in paths]
        return f"Table '{table_name}' contains files: {files}"

    @kernel_function(
        name="suggest_sql_query",
        description="Suggest a SQL query for a given analytical question."
    )
    def suggest_sql_query(self, question: str, table_name: str) -> str:
        return (
            f"Suggested query for '{question}':\n"
            f"SELECT * FROM usda_gold.{table_name} "
            f"WHERE ... -- customize based on your question\n"
            f"LIMIT 100;"
        )

from plugins.data_query_plugin import DataQueryPlugin

kernel = create_kernel()
kernel.add_plugin(DataQueryPlugin(), plugin_name="DataQuery")
print(f"Plugins: {list(kernel.plugins.keys())}")

import asyncio
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.contents import ChatHistory
from kernel_setup import create_kernel
from plugins.data_query_plugin import DataQueryPlugin

async def main():
    kernel = create_kernel()
    kernel.add_plugin(DataQueryPlugin(), plugin_name="DataQuery")

    agent = ChatCompletionAgent(
        kernel=kernel,
        service_id="gpt54",
        name="DataAnalyst",
        instructions=(
            "You are a data analyst for the CSA-in-a-Box platform. "
            "Use the DataQuery plugin to explore Gold layer tables "
            "and help users understand their data."
        ),
    )

    chat = ChatHistory()
    print("DataAnalyst Agent (type 'quit' to exit)")
    print("-" * 50)

    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
        chat.add_user_message(user_input)
        async for response in agent.invoke(chat):
            print(f"\nDataAnalyst: {response.content}")

asyncio.run(main())

import asyncio
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.agents.group_chat import (
    GroupChatOrchestration,
    RoundRobinGroupChatManager,
)
from semantic_kernel.agents.runtime import InProcessRuntime
from kernel_setup import create_kernel
from plugins.data_query_plugin import DataQueryPlugin

async def main():
    kernel = create_kernel()
    kernel.add_plugin(DataQueryPlugin(), plugin_name="DataQuery")

    # Agent 1: DataAnalyst
    analyst = ChatCompletionAgent(
        kernel=kernel,
        service_id="gpt54",
        name="DataAnalyst",
        instructions=(
            "You analyze data in the Gold layer. Use DataQuery tools "
            "to explore tables, suggest queries, and summarize findings. "
            "Present results clearly with numbers and tables."
        ),
    )

    # Agent 2: QualityReviewer
    reviewer = ChatCompletionAgent(
        kernel=kernel,
        service_id="gpt54",
        name="QualityReviewer",
        instructions=(
            "You review data quality. Check for completeness, accuracy, "
            "and consistency. Flag missing values, outliers, and schema "
            "issues. Provide a quality score (1-10) with justification."
        ),
    )

    # Agent 3: GovernanceOfficer
    officer = ChatCompletionAgent(
        kernel=kernel,
        service_id="gpt54",
        name="GovernanceOfficer",
        instructions=(
            "You enforce data governance policies. Check that data "
            "products comply with naming conventions, access controls, "
            "and classification rules. Flag PII or sensitive data. "
            "Approve or reject data for production use."
        ),
    )

    print("Multi-Agent Governance Review")
    print("=" * 50)

    # Set up orchestration
    orchestration = GroupChatOrchestration(
        members=[analyst, reviewer, officer],
        manager=RoundRobinGroupChatManager(max_rounds=6),
    )

    runtime = InProcessRuntime()
    runtime.start()

    result = await orchestration.invoke(
        task=(
            "Review the crop production data product in the Gold layer. "
            "The DataAnalyst should explore the data, the QualityReviewer "
            "should assess quality, and the GovernanceOfficer should "
            "verify compliance. Provide a final recommendation."
        ),
        runtime=runtime,
    )

    print("\n--- Final Result ---")
    print(result.value)

    await runtime.stop_when_idle()

asyncio.run(main())

from semantic_kernel.agents.group_chat import RoundRobinGroupChatManager

manager = RoundRobinGroupChatManager(
    max_rounds=10,
)

# The orchestration stops when max_rounds is reached
# or when an agent includes "APPROVED" or "REJECTED" in their response

orchestration = GroupChatOrchestration(
    members=[analyst, reviewer, officer],
    manager=RoundRobinGroupChatManager(max_rounds=3),  # One round each
)

import asyncio
from semantic_kernel.connectors.mcp import MCPStdioPlugin

async def create_mcp_plugin():
    plugin = MCPStdioPlugin(
        name="DataCatalog",
        command="python",
        args=["mcp_server.py"],
        env={"STORAGE_ACCOUNT": "csadlsdev"},
    )
    await plugin.connect()
    return plugin

async def main():
    kernel = create_kernel()

    mcp_plugin = await create_mcp_plugin()
    kernel.add_plugin(mcp_plugin, plugin_name="DataCatalog")

    agent = ChatCompletionAgent(
        kernel=kernel,
        service_id="gpt54",
        name="CatalogAgent",
        instructions="Use DataCatalog tools to browse and search the data catalog.",
    )

    # ... use agent as before

import asyncio
from kernel_setup import create_kernel
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.agents.group_chat import (
    GroupChatOrchestration,
    RoundRobinGroupChatManager,
)
from semantic_kernel.agents.runtime import InProcessRuntime
from plugins.data_query_plugin import DataQueryPlugin

async def governance_review(data_product: str):
    kernel = create_kernel()
    kernel.add_plugin(DataQueryPlugin(), plugin_name="DataQuery")

    agents = [
        ChatCompletionAgent(
            kernel=kernel, service_id="gpt54", name="DataAnalyst",
            instructions="Analyze the data product. List tables, row counts, and key metrics.",
        ),
        ChatCompletionAgent(
            kernel=kernel, service_id="gpt54", name="QualityReviewer",
            instructions="Review data quality. Score 1-10. Flag issues.",
        ),
        ChatCompletionAgent(
            kernel=kernel, service_id="gpt54", name="GovernanceOfficer",
            instructions="Check compliance. Approve or reject with reasons.",
        ),
    ]

    orchestration = GroupChatOrchestration(
        members=agents,
        manager=RoundRobinGroupChatManager(max_rounds=6),
    )

    runtime = InProcessRuntime()
    runtime.start()

    result = await orchestration.invoke(
        task=f"Perform a governance review of the '{data_product}' data product.",
        runtime=runtime,
    )

    await runtime.stop_when_idle()
    return result.value

if __name__ == "__main__":
    result = asyncio.run(governance_review("fct_crop_production"))
    print("\n" + "=" * 50)
    print("GOVERNANCE REVIEW COMPLETE")
    print("=" * 50)
    print(result)