إمكانية المراقبة على مستوى الإنتاج لوكلاء الذكاء الاصطناعي: الحد الأدنى من التعليمات البرمجية، نهج التكوين أولاً

# --------------------------------------------------
# 0. Load .env
# --------------------------------------------------
from dotenv import load_dotenv
load_dotenv(override=True)

# --------------------------------------------------
# 1. Imports
# --------------------------------------------------
import os
from typing import TypedDict

from langgraph.graph import StateGraph, END
from langchain_openai import AzureChatOpenAI

from langfuse import Langfuse
from langfuse.langchain import CallbackHandler

# --------------------------------------------------
# 2. Langfuse Client (WORKING CONFIG)
# --------------------------------------------------
langfuse = Langfuse(
    host="https://cloud.langfuse.com",
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"] , 
    secret_key=os.environ["LANGFUSE_SECRET_KEY"]  
)
langfuse_callback = CallbackHandler()
os.environ["LANGGRAPH_TRACING"] = "false"


# --------------------------------------------------
# 3. Azure OpenAI Setup
# --------------------------------------------------
llm = AzureChatOpenAI(
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2025-01-01-preview"),
    temperature=0.2,
    callbacks=[langfuse_callback],  # 🔑 enables token usage
)

# --------------------------------------------------
# 4. Shared State
# --------------------------------------------------
class AgentState(TypedDict, total=False):
    ticket: str
    category: str
    technical_response: str
    billing_response: str
    final_response: str

# --------------------------------------------------
# 5. Agent Definitions
# --------------------------------------------------

def triage_agent(state: dict) -> dict:
    with langfuse.start_as_current_observation(
        as_type="span",
        name="triage_agent",
        input={"ticket": state["ticket"]},
    ) as span:
        span.update_trace(name="Customer Service Query - LangGraph Demo") 

        response = llm.invoke([
            {
                "role": "system",
                "content": (
                    "Classify the query as one of: "
                    "Technical, Billing, Both. "
                    "Respond with only the label."
                ),
            },
            {"role": "user", "content": state["ticket"]},
        ])

        raw = response.content.strip().lower()

        if "both" in raw:
            category = "Both"
        elif "technical" in raw:
            category = "Technical"
        elif "billing" in raw:
            category = "Billing"
        else:
            category = "Technical"  # ✅ safe fallback

        span.update(output={"raw": raw, "category": category})

        return {"category": category}



def technical_support_agent(state: dict) -> dict:
    with langfuse.start_as_current_observation(
        as_type="span",
        name="technical_support_agent",
        input={
            "ticket": state["ticket"],
            "category": state.get("category"),
        },
    ) as span:

        response = llm.invoke([
            {
                "role": "system",
                "content": (
                    "You are a technical support specialist. "
                    "Provide a clear, step-by-step solution."
                ),
            },
            {"role": "user", "content": state["ticket"]},
        ])

        answer = response.content

        span.update(output={"technical_response": answer})

        return {"technical_response": answer}


def billing_support_agent(state: dict) -> dict:
    with langfuse.start_as_current_observation(
        as_type="span",
        name="billing_support_agent",
        input={
            "ticket": state["ticket"],
            "category": state.get("category"),
        },
    ) as span:

        response = llm.invoke([
            {
                "role": "system",
                "content": (
                    "You are a billing support specialist. "
                    "Answer clearly about payments, invoices, or accounts."
                ),
            },
            {"role": "user", "content": state["ticket"]},
        ])

        answer = response.content

        span.update(output={"billing_response": answer})

        return {"billing_response": answer}

def finalizer_agent(state: dict) -> dict:
    with langfuse.start_as_current_observation(
        as_type="span",
        name="finalizer_agent",
        input={
            "ticket": state["ticket"],
            "technical": state.get("technical_response"),
            "billing": state.get("billing_response"),
        },
    ) as span:

        parts = [
            f"Technical:\n{state['technical_response']}"
            for k in ["technical_response"]
            if state.get(k)
        ] + [
            f"Billing:\n{state['billing_response']}"
            for k in ["billing_response"]
            if state.get(k)
        ]

        if not parts:
            final = "Error: No agent responses available."
        else:
            response = llm.invoke([
                {
                    "role": "system",
                    "content": (
                        "Combine the following agent responses into ONE clear, professional, "
                        "customer-facing answer. Do not mention agents or internal labels. "
                        f"Answer the user's query: '{state['ticket']}'."
                    ),
                },
                {"role": "user", "content": "\n\n".join(parts)},
            ])
            final = response.content

        span.update(output={"final_response": final})
        return {"final_response": final}


# --------------------------------------------------
# 6. LangGraph Construction 
# --------------------------------------------------
builder = StateGraph(AgentState)

builder.add_node("triage", triage_agent)
builder.add_node("technical", technical_support_agent)
builder.add_node("billing", billing_support_agent)
builder.add_node("finalizer", finalizer_agent)

builder.set_entry_point("triage")

# Conditional routing
builder.add_conditional_edges(
    "triage",
    lambda state: state["category"],
    {
        "Technical": "technical",
        "Billing": "billing",
        "Both": "technical",
        "__default__": "technical",  # ✅ never dead-end
    },
)

# Sequential resolution
builder.add_conditional_edges(
    "technical",
    lambda state: state["category"],
    {
        "Both": "billing",         # Proceed to billing if Both
        "__default__": "finalizer",
    },
)
builder.add_edge("billing", "finalizer")
builder.add_edge("finalizer", END)

graph = builder.compile()


# --------------------------------------------------
# 9. Main
# --------------------------------------------------
if __name__ == "__main__":

    print("===============================================")
    print(" Conditional Multi-Agent Support System (Ready)")
    print("===============================================")
    print("Enter 'exit' or 'quit' to stop the program.\n")
    
    while True:
        # Get user input for the ticket
        ticket = input("Enter your support query (ticket): ")

        # Check for exit command
        if ticket.lower() in ["exit", "quit"]:
            print("\nExiting the support system. Goodbye!")
            break

        if not ticket.strip():
            print("Please enter a non-empty query.")
            continue
            
        try:                
                # --- Run the graph with the user's ticket ---
             result = graph.invoke(
                {"ticket": ticket},
                config={"callbacks": [langfuse_callback]},
            )
        
            # --- Print Results ---
            category = result.get('category', 'N/A')
            print(f"\n✅ Triage Classification: **{category}**")
            
            # Check which agents were executed based on the presence of a response
            executed_agents = []
            if result.get("technical_response"):
                executed_agents.append("Technical")
            if result.get("billing_response"):
                executed_agents.append("Billing")
            
            
            print(f"🛠️ Agents Executed: {', '.join(executed_agents) if executed_agents else 'None (Triage Failed)'}")

            print("\n================ FINAL RESPONSE ================\n")
            print(result["final_response"])
            print("\n" + "="*60 + "\n")

        except Exception as e:
            # This is important for debugging: print the exception type and message
            print(f"\nAn error occurred during processing ({type(e).__name__}): {e}")
            print("\nPlease try another query.")
            print("\n" + "="*60 + "\n")

from langfuse import get_client
from langfuse.openai import OpenAI
from langchain_openai import AzureChatOpenAI
from langfuse import Langfuse
import os
# Initialize client
from dotenv import load_dotenv
load_dotenv(override=True)

langfuse = Langfuse(
    host="https://cloud.langfuse.com",
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"] , 
    secret_key=os.environ["LANGFUSE_SECRET_KEY"]  
)

llm = AzureChatOpenAI(
    azure_deployment=os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME"),
    api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2025-01-01-preview"),
    temperature=0.2,
)

# Define your task function
def my_task(*, item, **kwargs):
    question = item.input['ticket'] 
    response = llm.invoke([{"role": "user", "content": question}])

    raw = response.content.strip().lower()
 
    return raw  
 
# Get dataset from Langfuse
dataset = langfuse.get_dataset("Regression")
 
# Run experiment directly on the dataset
result = dataset.run_experiment(
    name="Production Model Test",
    description="Monthly evaluation of our production model",
    task=my_task # see above for the task definition
)
 
# Use format method to display results
print(result.format())