add azure-ai-agentserver-langgraph package

Author: lusu-msft

sdk/ai/azure-ai-agentserver-langgraph/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Release History
+
+## 1.0.0
+
+### Features Added
+
+First version

sdk/ai/azure-ai-agentserver-langgraph/README.md

@@ -0,0 +1,29 @@
+# Azure AI Agent Server adapter for LangGraph
+
+## Install
+
+In current folder, run:
+```bash
+pip install -e .
+```
+
+## Usage
+
+```python
+# your existing agent
+from my_langgraph_agent import my_awesome_agent
+
+# langgraph utils
+from azure.ai.agentserver.langgraph import from_langgraph
+
+if __name__ == "__main__":
+    # with this simple line, your agent will be hosted on http://localhost:8088
+    from_langgraph(my_awesome_agent).run()
+
+```
+
+**Note**
+If your langgraph agent was not using langgraph's builtin [MessageState](https://langchain-ai.github.io/langgraph/concepts/low_level/?h=messagesstate#messagesstate), you should implement your own `LanggraphStateConverter` and provide to `from_langgraph`.
+
+Reference this [example](https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/ai/azure-ai-agentserver-langgraph/samples/custom_state/main.py) for more details.
+

sdk/ai/azure-ai-agentserver-langgraph/azure/__init__.py

@@ -0,0 +1 @@
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/__init__.py

@@ -0,0 +1 @@
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/__init__.py

@@ -0,0 +1 @@
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/__init__.py

@@ -0,0 +1,21 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)
+
+from typing import TYPE_CHECKING, Optional
+
+from ._version import VERSION
+
+if TYPE_CHECKING:  # pragma: no cover
+    from . import models
+
+
+def from_langgraph(agent, state_converter: Optional["models.LanggraphStateConverter"] = None):
+    from .langgraph import LangGraphAdapter
+
+    return LangGraphAdapter(agent, state_converter=state_converter)
+
+
+__all__ = ["from_langgraph"]
+__version__ = VERSION

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/_version.py

@@ -0,0 +1,9 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) Python Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+VERSION = "1.0.0a1"

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/langgraph.py

@@ -0,0 +1,153 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+import os
+import re
+from typing import Optional
+
+from azure.ai.agentserver.core.constants import Constants
+from azure.ai.agentserver.core.logger import get_logger
+from azure.ai.agentserver.core.server.base import FoundryCBAgent
+from azure.ai.agentserver.core.server.common.agent_run_context import AgentRunContext
+from langchain_core.runnables import RunnableConfig
+
+from langgraph.graph.state import CompiledStateGraph
+
+from .models import (
+    LanggraphMessageStateConverter,
+    LanggraphStateConverter,
+)
+from .models.utils import is_state_schema_valid
+
+logger = get_logger()
+
+
+class LangGraphAdapter(FoundryCBAgent):
+    """
+    Adapter for LangGraph Agent.
+    """
+
+    def __init__(self, graph: CompiledStateGraph, state_converter: Optional[LanggraphStateConverter] = None):
+        """
+        Initialize the LangGraphAdapter with a CompiledStateGraph.
+
+        Args:
+            graph (StateGraph): The LangGraph StateGraph to adapt.
+            state_converter: custom state converter. Required if graph state is not MessagesState.
+        """
+        super().__init__()
+        self.graph = graph
+        self.azure_ai_tracer = None
+        if not state_converter:
+            if is_state_schema_valid(self.graph.builder.state_schema):
+                self.state_converter = LanggraphMessageStateConverter()
+            else:
+                raise ValueError("state_converter is required for non-MessagesState graph.")
+        else:
+            self.state_converter = state_converter
+
+    async def agent_run(self, context: AgentRunContext):
+        input_data = self.state_converter.request_to_state(context)
+        logger.debug(f"Converted input data: {input_data}")
+        if not context.stream:
+            response = await self.agent_run_non_stream(input_data, context)
+            return response
+        else:
+            return self.agent_run_astream(input_data, context)
+
+    def init_tracing_internal(self, exporter_endpoint=None, app_insights_conn_str=None):
+        # set env vars for langsmith
+        os.environ["LANGSMITH_OTEL_ENABLED"] = "true"
+        os.environ["LANGSMITH_TRACING"] = "true"
+        os.environ["LANGSMITH_OTEL_ONLY"] = "true"
+        if app_insights_conn_str:
+            # setup azure ai telemetry callbacks
+            try:
+                from langchain_azure_ai.callbacks.tracers import AzureAIOpenTelemetryTracer
+
+                self.azure_ai_tracer = AzureAIOpenTelemetryTracer(
+                    connection_string=app_insights_conn_str,
+                    enable_content_recording=True,
+                    name=self.get_agent_identifier(),
+                )
+                logger.info("AzureAIOpenTelemetryTracer initialized successfully.")
+            except Exception as e:
+                logger.error(f"Failed to import AzureAIOpenTelemetryTracer, ignore: {e}")
+
+    def setup_otlp_exporter(self, endpoint, provider):
+        endpoint = self.format_otlp_endpoint(endpoint)
+        return super().setup_otlp_exporter(endpoint, provider)
+
+    def get_trace_attributes(self):
+        attrs = super().get_trace_attributes()
+        attrs["service.namespace"] = "azure.ai.agentshosting.langgraph"
+        return attrs
+
+    async def agent_run_non_stream(self, input_data: dict, context: AgentRunContext):
+        """
+        Run the agent with non-streaming response.
+
+        Args:
+            context (AgentRunContext): The context for the agent run.
+
+        Returns:
+            RunObject: The result of the agent run.
+        """
+
+        try:
+            config = self.create_runnable_config(context)
+            stream_mode = self.state_converter.get_stream_mode(context)
+            result = await self.graph.ainvoke(input_data, config=config, stream_mode=stream_mode)
+            output = self.state_converter.state_to_response(result, context)
+            return output
+        except Exception as e:
+            logger.error(f"Error during agent run: {e}")
+            raise e
+
+    async def agent_run_astream(self, input_data: dict, context: AgentRunContext):
+        """
+        Run the agent with streaming response.
+
+        Args:
+            request_body (CreateResponse): The request body to run the agent with.
+
+        Returns:
+            StreamingResponse: The streaming response of the agent run.
+        """
+        try:
+            logger.info(f"Starting streaming agent run {context.response_id}")
+            config = self.create_runnable_config(context)
+            stream_mode = self.state_converter.get_stream_mode(context)
+            stream = self.graph.astream(input=input_data, config=config, stream_mode=stream_mode)
+            async for result in self.state_converter.state_to_response_stream(stream, context):
+                yield result
+        except Exception as e:
+            logger.error(f"Error during streaming agent run: {e}")
+            raise e
+
+    def create_runnable_config(self, context: AgentRunContext) -> RunnableConfig:
+        """
+        Create a RunnableConfig from the converted request data.
+        """
+        config = RunnableConfig(
+            configurable={
+                "thread_id": context.conversation_id,
+            },
+            callbacks=[self.azure_ai_tracer] if self.azure_ai_tracer else None,
+        )
+        return config
+
+    def format_otlp_endpoint(self, endpoint: str) -> str:
+        m = re.match(r"^(https?://[^/]+)", endpoint)
+        if m:
+            return f"{m.group(1)}/v1/traces"
+        return endpoint
+
+    def get_agent_identifier(self) -> str:
+        agent_name = os.getenv(Constants.AGENT_NAME)
+        if agent_name:
+            return agent_name
+        agent_id = os.getenv(Constants.AGENT_ID)
+        if agent_id:
+            return agent_id
+        return "AgentsHosting-LangGraph"

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/__init__.py

@@ -0,0 +1,15 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+from .langgraph_request_converter import LangGraphRequestConverter
+from .langgraph_response_converter import LangGraphResponseConverter
+from .langgraph_state_converter import LanggraphMessageStateConverter, LanggraphStateConverter
+from .langgraph_stream_response_converter import LangGraphStreamResponseConverter
+
+__all__ = [
+    "LangGraphRequestConverter",
+    "LangGraphResponseConverter",
+    "LangGraphStreamResponseConverter",
+    "LanggraphStateConverter",
+    "LanggraphMessageStateConverter",
+]

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/langgraph_request_converter.py

@@ -0,0 +1,130 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+import json
+from typing import Dict, List
+
+from azure.ai.agentserver.core.logger import get_logger
+from azure.ai.agentserver.core.models import CreateResponse, openai as openai_models, projects as project_models
+from langchain_core.messages import (
+    AIMessage,
+    AnyMessage,
+    HumanMessage,
+    SystemMessage,
+    ToolMessage,
+)
+from langchain_core.messages.tool import ToolCall
+
+logger = get_logger()
+
+role_mapping = {
+    project_models.ResponsesMessageRole.USER: HumanMessage,
+    project_models.ResponsesMessageRole.SYSTEM: SystemMessage,
+    project_models.ResponsesMessageRole.ASSISTANT: AIMessage,
+    # project_models.ResponsesMessageRole.DEVELOPER: ToolMessage,
+}
+
+item_content_type_mapping = {
+    project_models.ItemContentType.INPUT_TEXT: "text",
+    project_models.ItemContentType.INPUT_AUDIO: "audio",
+    project_models.ItemContentType.INPUT_IMAGE: "image",
+    project_models.ItemContentType.INPUT_FILE: "file",
+    project_models.ItemContentType.OUTPUT_TEXT: "text",
+    project_models.ItemContentType.OUTPUT_AUDIO: "audio",
+    # project_models.ItemContentType.REFUSAL: "refusal",
+}
+
+
+class LangGraphRequestConverter:
+    def __init__(self, data: CreateResponse):
+        self.data: CreateResponse = data
+
+    def convert(self) -> dict:
+        # Convert the CreateRunRequest input to a format suitable for LangGraph
+        langgraph_input = {"messages": []}
+
+        instructions = self.data.get("instructions")
+        if instructions and isinstance(instructions, str):
+            langgraph_input["messages"].append(SystemMessage(content=instructions))
+
+        input = self.data.get("input")
+        if isinstance(input, str):
+            langgraph_input["messages"].append(HumanMessage(content=input))
+        elif isinstance(input, List):
+            for inner in input:
+                message = self.convert_input(inner)
+                langgraph_input["messages"].append(message)
+        else:
+            raise ValueError(f"Unsupported input type: {type(input)}, {input}")
+        return langgraph_input
+
+    def convert_input(self, item: openai_models.ResponseInputItemParam) -> AnyMessage:
+        """
+        Convert ResponseInputItemParam to a LangGraph message
+        """
+        item_type = item.get("type", project_models.ItemType.MESSAGE)
+        if item_type == project_models.ItemType.MESSAGE:
+            # this is a message
+            return self.convert_message(item)
+        elif item_type == project_models.ItemType.FUNCTION_CALL:
+            return self.convert_function_call(item)
+        elif item_type == project_models.ItemType.FUNCTION_CALL_OUTPUT:
+            return self.convert_function_call_output(item)
+        else:
+            raise ValueError(f"Unsupported OpenAIItemParam type: {item_type}, {item}")
+
+    def convert_message(self, message: dict) -> AnyMessage:
+        """
+        Convert a message dict to a LangGraph message
+        """
+        content = message.get("content")
+        role = message.get("role", project_models.ResponsesMessageRole.USER)
+        if not content:
+            raise ValueError(f"Message missing content: {message}")
+        if isinstance(content, str):
+            return role_mapping[role](content=content)
+        elif isinstance(content, list):
+            return role_mapping[role](content=self.convert_OpenAIItemContentList(content))
+        raise ValueError(f"Unsupported ResponseMessagesItemParam content type: {type(content)}, {content}")
+
+    def convert_function_call(self, item: dict) -> AnyMessage:
+        try:
+            item = openai_models.ResponseFunctionToolCallParam(**item)
+            argument = item.get("arguments", None)
+            args = json.loads(argument) if argument else {}
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Invalid JSON in function call arguments: {argument}") from e
+        except Exception as e:
+            raise ValueError(f"Invalid function call item: {item}") from e
+        return AIMessage(tool_calls=[ToolCall(id=item.get("call_id"), name=item.get("name"), args=args)], content="")
+
+    def convert_function_call_output(self, item: dict) -> ToolMessage:
+        try:
+            item = openai_models.response_input_item_param.FunctionCallOutput(**item)
+        except Exception as e:
+            raise ValueError(f"Invalid function call output item: {item}") from e
+
+        output = item.get("output", None)
+        if isinstance(output, str):
+            return ToolMessage(content=output, tool_call_id=item.get("call_id"))
+        elif isinstance(output, list):
+            return ToolMessage(content=self.convert_OpenAIItemContentList(output), tool_call_id=item.get("call_id"))
+        raise ValueError(f"Unsupported function call output type: {type(output)}, {output}")
+
+    def convert_OpenAIItemContentList(self, content: List[Dict]) -> List:
+        """
+        Convert ItemContent to a list format
+        """
+        result = []
+        for item in content:
+            result.append(self.convert_OpenAIItemContent(item))
+        return result
+
+    def convert_OpenAIItemContent(self, content: Dict) -> Dict:
+        """
+        Convert ItemContent to a dict format
+        """
+        res = content.copy()
+        content_type = content.get("type")
+        res["type"] = item_content_type_mapping.get(content_type, content_type)
+        return res