用 Microsoft Fabric 和 Azure 数据库搭建 Agentic 应用：统一数据与 AI 的实战路径

# 安装依赖
pip install semantic-kernel azure-cosmosdb-python openai

# 设置环境变量（按你的实际资源替换）
export AZURE_OPENAI_ENDPOINT="https://your-openai.openai.azure.com/"
export AZURE_OPENAI_API_KEY="your-key"
export AZURE_OPENAI_DEPLOYMENT="gpt-4.1"
export COSMOS_ENDPOINT="https://your-cosmos.documents.azure.com:443/"
export COSMOS_KEY="your-cosmos-key"
export COSMOS_DATABASE="AgentStateDB"
export COSMOS_CONTAINER="Sessions"

import os
import json
import uuid
from datetime import datetime

from azure.cosmos import CosmosClient, PartitionKey
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.functions import kernel_function


# ── 1. Cosmos DB 会话状态存储 ──────────────────────────────
class SessionStore:
    """把 Agent 的每轮对话和工具调用记录写入 Cosmos DB"""

    def __init__(self):
        client = CosmosClient(
            os.environ["COSMOS_ENDPOINT"], os.environ["COSMOS_KEY"]
        )
        db = client.get_database_client(os.environ["COSMOS_DATABASE"])
        # 如果容器不存在就创建
        try:
            db.create_container(
                id=os.environ["COSMOS_CONTAINER"],
                partition_key=PartitionKey(path="/user_id"),
            )
        except Exception:
            pass  # 容器已存在
        self.container = db.get_container_client(os.environ["COSMOS_CONTAINER"])

    def save(self, user_id: str, role: str, content: str, metadata: dict = None):
        doc = {
            "id": str(uuid.uuid4()),
            "user_id": user_id,
            "role": role,
            "content": content,
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": metadata or {},
        }
        self.container.upsert_item(doc)
        return doc["id"]

    def get_recent(self, user_id: str, limit: int = 10):
        query = f"""
        SELECT TOP {limit} c.role, c.content, c.timestamp
        FROM c WHERE c.user_id = '{user_id}'
        ORDER BY c.timestamp DESC
        """
        return list(self.container.query_items(query, enable_cross_partition_query=False))


# ── 2. 模拟 Fabric 数据查询的插件 ────────────────────────────
class FabricDataPlugin:
    """实际生产中，这里会调用 Fabric REST API 或 OneLake Delta 表读取。
    示例用模拟数据演示 Agent 的调用链路。"""

    @kernel_function(
        description="查询最近 N 天的销售汇总数据，返回按日期聚合的营收和订单数"
    )
    def get_sales_summary(self, days: int = 7) -> str:
        # 生产替换：用 pyarrow / delta-python 读 OneLake 中的 Delta 表
        # from deltalake import DeltaTable
        # table = DeltaTable("abfss://workspace@onelake.dfs.core.windows.net/sales/lakehouse/Tables/sales_fact")
        # df = table.to_pandas().tail(days * 24) ...
        mock_data = [
            {"date": "2026-05-19", "revenue": 128400, "orders": 342},
            {"date": "2026-05-18", "revenue": 115200, "orders": 310},
            {"date": "2026-05-17", "revenue": 98700,  "orders": 278},
        ]
        return json.dumps(mock_data, ensure_ascii=False)

    @kernel_function(
        description="查询指定 SKU 的库存余量（从 Azure SQL 业务库读取）"
    )
    def get_inventory(self, sku: str) -> str:
        # 生产替换：用 pyodbc 连 Azure SQL 执行 SELECT
        mock = {"sku": sku, "qty_on_hand": 520, "warehouse": "SH-03"}
        return json.dumps(mock, ensure_ascii=False)


# ── 3. 组装 Agent ────────────────────────────────────────────
def build_agent() -> ChatCompletionAgent:
    kernel = Kernel()
    kernel.add_service(
        AzureChatCompletion(
            service_id="azure-openai",
            deployment_name=os.environ["AZURE_OPENAI_DEPLOYMENT"],
            endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
            api_key=os.environ["AZURE_OPENAI_API_KEY"],
        )
    )
    kernel.add_plugin(FabricDataPlugin(), plugin_name="fabric_data")

    agent = ChatCompletionAgent(
        kernel=kernel,
        name="SalesAnalyst",
        instructions="""你是一个销售分析 Agent。
- 用户问销售趋势时，调用 get_sales_summary 获取数据再分析。
- 用户问库存时，调用 get_inventory 查询具体 SKU。
- 回答要包含数据引用，不要凭空编数字。
- 如果数据不足，明确告诉用户并建议扩大查询范围。""",
    )
    return agent


# ── 4. 运行 ──────────────────────────────────────────────────
async def main():
    store = SessionStore()
    agent = build_agent()
    user_id = "user-001"

    # 加载历史会话作为上下文
    history = store.get_recent(user_id, limit=6)
    context_msgs = [f"{h['role']}: {h['content']}" for h in history]

    questions = [
        "最近 3 天销售情况怎么样？",
        "SKU-A1001 还有多少库存？",
    ]

    for q in questions:
        store.save(user_id, "user", q)
        full_prompt = "\n".join(context_msgs) + f"\nuser: {q}"
        response = await agent.invoke(full_prompt)
        store.save(user_id, "assistant", str(response), metadata={"source": "agent"})
        print(f"Q: {q}\nA: {response}\n---")
        context_msgs.append(f"user: {q}")
        context_msgs.append(f"assistant: {response}")


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