mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
[BREAKING] Python: Schema changes for azure functions package (#2151)
* Python: Add Scaffolding for Durable AzureFunctions package to Agent Framework (#1823) * Add scafolding * update readme * add code owners and label * update owners * .NET: Durable extension: initial src and unit tests (#1900) * Python: Add Durable Agent Wrapper code (#1913) * add initial changes * Move code and add single sample * Update logger * Remove unused code * address PR comments * cleanup code and address comments --------- Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com> * Azure Functions .NET samples (#1939) * Python: Add Unit tests for Azurefunctions package (#1976) * Add Unit tests for Azurefunctions * remove duplicate import * .NET: [Feature Branch] Migrate state schema updates and support for agents as MCP tools (#1979) * Python: Add more samples for Azure Functions (#1980) * Move all samples * fix comments * remove dead lines * Make samples simpler * .NET: [Feature Branch] Durable Task extension integration tests (#2017) * .NET: [Feature Branch] Update OpenAI config for integration tests (#2063) * Python: Add Integration tests for AzureFunctions (#2020) * Add Integration tests * Remove DTS extension * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Add pyi file for type safety * Add samples in readme * Updated all readme instructions * Address comments * Update readmes * Fix requirements * Address comments --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * .NET: [Feature Branch] Update dotnet-build-and-test.yml to support integration tests (#2070) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Fix DTS startup issue and improve logging (#2103) * .NET: [Feature Branch] Introduce Azure OpenAI config for .NET pipeline (#2106) Also fixes an issue where we were trying to start docker containers for integration tests on Windows, which doesn't work. Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Fix uv.lock after merge * Python: Add README for Azure Functions samples setup (#2100) * Add README for Azure Functions samples setup Added setup instructions for Azure Functions samples, including environment setup, virtual environment creation, and running samples. * Update python/samples/getting_started/azure_functions/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestion from @Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Laveesh Rohra <larohra@microsoft.com> * Fix or remove broken markdown file links (#2115) * .NET: [Feature Branch] Update HTTP API to be consistent across languages (#2118) * Python: Fix AzureFunctions Integration Tests (#2116) * Add Identity Auth to samples * Update python/samples/getting_started/azure_functions/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/samples/getting_started/azure_functions/01_single_agent/function_app.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/samples/getting_started/azure_functions/02_multi_agent/function_app.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/samples/getting_started/azure_functions/06_multi_agent_orchestration_conditionals/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Python: Fix Http Schema (#2112) * Rename to threadid * Respond in plain text * Make snake-case * Add http prefix * rename to wait-for-response * Add query param check * address comments * .NET: Remove IsPackable=false in preparation for nuget release (#2142) * Python: Move `azurefunctions` to `azure` for import (#2141) * Move import to Azure * fix mypy * Update python/packages/azurefunctions/README.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Add missing types * Address comments --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/packages/azurefunctions/pyproject.toml Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/packages/azurefunctions/agent_framework_azurefunctions/__init__.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Fix imports * Address PR feedback from westey-m (#2150) - Adds a link from the /dotnet/samples/README.md to /dotnet/samples/AzureFunctions - Make DurableAgentThread deserialization internal for future-proofing - Update JSON serialization logic to address recently discovered issues with source generator serialization * Schema changes for azure functions * Fixed serialization bug * update to camel case * Adding logs * merge with main * sync uv.lock * Updated schema * Fixed tests * Addressed comments * Fixed mypy errors * Fixed bug in responsetype and authorName * Addressed feedback * Addressed more feedback * Python: Addressing comments for #2151 (#2315) * Initial fixes * Address more comments * Address remaining comments * Fixed remaining snake_case properties * Fixed remaining snake_case properties * Fixed mypy errors * Minor changes * revert tool names * Fixed mypy errors --------- Co-authored-by: Laveesh Rohra <larohra@microsoft.com> Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com> Co-authored-by: Chris Gillum <cgillum@microsoft.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Anirudh Garg <anirudhg@microsoft.com> Co-authored-by: Victoria Hall <victoriahall@microsoft.com>
This commit is contained in:
Gavin Aguiar
committed by
GitHub
Unverified
GitHub
parent
039e49f353
commit
6ae32f007d
@@ -16,27 +16,29 @@ import azure.functions as func
|
||||
from agent_framework import AgentProtocol, get_logger
|
||||
|
||||
from ._callbacks import AgentResponseCallbackProtocol
|
||||
from ._constants import (
|
||||
DEFAULT_MAX_POLL_RETRIES,
|
||||
DEFAULT_POLL_INTERVAL_SECONDS,
|
||||
MIMETYPE_APPLICATION_JSON,
|
||||
MIMETYPE_TEXT_PLAIN,
|
||||
REQUEST_RESPONSE_FORMAT_JSON,
|
||||
REQUEST_RESPONSE_FORMAT_TEXT,
|
||||
THREAD_ID_FIELD,
|
||||
THREAD_ID_HEADER,
|
||||
WAIT_FOR_RESPONSE_FIELD,
|
||||
WAIT_FOR_RESPONSE_HEADER,
|
||||
)
|
||||
from ._durable_agent_state import DurableAgentState
|
||||
from ._entities import create_agent_entity
|
||||
from ._errors import IncomingRequestError
|
||||
from ._models import AgentSessionId, RunRequest
|
||||
from ._orchestration import AgentOrchestrationContextType, DurableAIAgent
|
||||
from ._state import AgentState
|
||||
|
||||
logger = get_logger("agent_framework.azurefunctions")
|
||||
|
||||
THREAD_ID_FIELD: str = "thread_id"
|
||||
RESPONSE_FORMAT_JSON: str = "json"
|
||||
RESPONSE_FORMAT_TEXT: str = "text"
|
||||
WAIT_FOR_RESPONSE_FIELD: str = "wait_for_response"
|
||||
WAIT_FOR_RESPONSE_HEADER: str = "x-ms-wait-for-response"
|
||||
|
||||
|
||||
EntityHandler = Callable[[df.DurableEntityContext], None]
|
||||
HandlerT = TypeVar("HandlerT", bound=Callable[..., Any])
|
||||
|
||||
DEFAULT_MAX_POLL_RETRIES: int = 30
|
||||
DEFAULT_POLL_INTERVAL_SECONDS: float = 1.0
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
||||
class DFAppBase:
|
||||
@@ -317,11 +319,11 @@ class AgentFunctionApp(DFAppBase):
|
||||
"""
|
||||
logger.debug(f"[HTTP Trigger] Received request on route: /api/agents/{agent_name}/run")
|
||||
|
||||
response_format: str = RESPONSE_FORMAT_JSON
|
||||
request_response_format: str = REQUEST_RESPONSE_FORMAT_JSON
|
||||
thread_id: str | None = None
|
||||
|
||||
try:
|
||||
req_body, message, response_format = self._parse_incoming_request(req)
|
||||
req_body, message, request_response_format = self._parse_incoming_request(req)
|
||||
thread_id = self._resolve_thread_id(req=req, req_body=req_body)
|
||||
wait_for_response = self._should_wait_for_response(req=req, req_body=req_body)
|
||||
|
||||
@@ -334,7 +336,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return self._create_http_response(
|
||||
payload={"error": "Message is required"},
|
||||
status_code=400,
|
||||
response_format=response_format,
|
||||
request_response_format=request_response_format,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
|
||||
@@ -351,6 +353,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
message,
|
||||
thread_id,
|
||||
correlation_id,
|
||||
request_response_format,
|
||||
)
|
||||
logger.debug("Signalling entity %s with request: %s", entity_instance_id, run_request)
|
||||
await client.signal_entity(entity_instance_id, "run_agent", run_request)
|
||||
@@ -370,7 +373,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return self._create_http_response(
|
||||
payload=result,
|
||||
status_code=200 if result.get("status") == "success" else 500,
|
||||
response_format=response_format,
|
||||
request_response_format=request_response_format,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
|
||||
@@ -383,7 +386,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return self._create_http_response(
|
||||
payload=accepted_response,
|
||||
status_code=202,
|
||||
response_format=response_format,
|
||||
request_response_format=request_response_format,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
|
||||
@@ -392,7 +395,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return self._create_http_response(
|
||||
payload={"error": str(exc)},
|
||||
status_code=exc.status_code,
|
||||
response_format=response_format,
|
||||
request_response_format=request_response_format,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
except ValueError as exc:
|
||||
@@ -400,7 +403,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return self._create_http_response(
|
||||
payload={"error": "Invalid JSON"},
|
||||
status_code=400,
|
||||
response_format=response_format,
|
||||
request_response_format=request_response_format,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
except Exception as exc:
|
||||
@@ -408,7 +411,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return self._create_http_response(
|
||||
payload={"error": str(exc)},
|
||||
status_code=500,
|
||||
response_format=response_format,
|
||||
request_response_format=request_response_format,
|
||||
thread_id=thread_id,
|
||||
)
|
||||
|
||||
@@ -466,7 +469,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return func.HttpResponse(
|
||||
json.dumps({"status": "healthy", "agents": agent_info, "agent_count": len(self.agents)}),
|
||||
status_code=200,
|
||||
mimetype="application/json",
|
||||
mimetype=MIMETYPE_APPLICATION_JSON,
|
||||
)
|
||||
|
||||
_ = health_check
|
||||
@@ -491,7 +494,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
self,
|
||||
client: df.DurableOrchestrationClient,
|
||||
entity_instance_id: df.EntityId,
|
||||
) -> AgentState | None:
|
||||
) -> DurableAgentState | None:
|
||||
state_response = await client.read_entity_state(entity_instance_id)
|
||||
if not state_response or not state_response.entity_exists:
|
||||
return None
|
||||
@@ -502,9 +505,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
|
||||
typed_state_payload = cast(dict[str, Any], state_payload)
|
||||
|
||||
agent_state = AgentState()
|
||||
agent_state.restore_state(typed_state_payload)
|
||||
return agent_state
|
||||
return DurableAgentState.from_dict(typed_state_payload)
|
||||
|
||||
async def _get_response_from_entity(
|
||||
self,
|
||||
@@ -580,31 +581,58 @@ class AgentFunctionApp(DFAppBase):
|
||||
|
||||
return result
|
||||
|
||||
def _build_response_payload(
|
||||
self,
|
||||
*,
|
||||
response: str | None,
|
||||
message: str,
|
||||
thread_id: str,
|
||||
status: str,
|
||||
correlation_id: str,
|
||||
extra_fields: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a consistent response structure and allow optional extra fields."""
|
||||
payload = {
|
||||
"response": response,
|
||||
"message": message,
|
||||
THREAD_ID_FIELD: thread_id,
|
||||
"status": status,
|
||||
"correlation_id": correlation_id,
|
||||
}
|
||||
if extra_fields:
|
||||
payload.update(extra_fields)
|
||||
return payload
|
||||
|
||||
async def _build_timeout_result(self, message: str, thread_id: str, correlation_id: str) -> dict[str, Any]:
|
||||
"""Create the timeout response."""
|
||||
return {
|
||||
"response": "Agent is still processing or timed out...",
|
||||
"message": message,
|
||||
THREAD_ID_FIELD: thread_id,
|
||||
"status": "timeout",
|
||||
"correlation_id": correlation_id,
|
||||
}
|
||||
return self._build_response_payload(
|
||||
response="Agent is still processing or timed out...",
|
||||
message=message,
|
||||
thread_id=thread_id,
|
||||
status="timeout",
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
|
||||
def _build_success_result(
|
||||
self, response_data: dict[str, Any], message: str, thread_id: str, correlation_id: str, state: AgentState
|
||||
self, response_data: dict[str, Any], message: str, thread_id: str, correlation_id: str, state: DurableAgentState
|
||||
) -> dict[str, Any]:
|
||||
"""Build the success result returned to the HTTP caller."""
|
||||
return {
|
||||
"response": response_data.get("content"),
|
||||
"message": message,
|
||||
THREAD_ID_FIELD: thread_id,
|
||||
"status": "success",
|
||||
"message_count": response_data.get("message_count", state.message_count),
|
||||
"correlation_id": correlation_id,
|
||||
}
|
||||
return self._build_response_payload(
|
||||
response=response_data.get("content"),
|
||||
message=message,
|
||||
thread_id=thread_id,
|
||||
status="success",
|
||||
correlation_id=correlation_id,
|
||||
extra_fields={"message_count": response_data.get("message_count", state.message_count)},
|
||||
)
|
||||
|
||||
def _build_request_data(
|
||||
self, req_body: dict[str, Any], message: str, thread_id: str, correlation_id: str
|
||||
self,
|
||||
req_body: dict[str, Any],
|
||||
message: str,
|
||||
thread_id: str,
|
||||
correlation_id: str,
|
||||
request_response_format: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Create the durable entity request payload."""
|
||||
enable_tool_calls_value = req_body.get("enable_tool_calls")
|
||||
@@ -613,6 +641,7 @@ class AgentFunctionApp(DFAppBase):
|
||||
return RunRequest(
|
||||
message=message,
|
||||
role=req_body.get("role"),
|
||||
request_response_format=request_response_format,
|
||||
response_format=req_body.get("response_format"),
|
||||
enable_tool_calls=enable_tool_calls,
|
||||
thread_id=thread_id,
|
||||
@@ -621,23 +650,23 @@ class AgentFunctionApp(DFAppBase):
|
||||
|
||||
def _build_accepted_response(self, message: str, thread_id: str, correlation_id: str) -> dict[str, Any]:
|
||||
"""Build the response returned when not waiting for completion."""
|
||||
return {
|
||||
"response": "Agent request accepted",
|
||||
"message": message,
|
||||
THREAD_ID_FIELD: thread_id,
|
||||
"status": "accepted",
|
||||
"correlation_id": correlation_id,
|
||||
}
|
||||
return self._build_response_payload(
|
||||
response="Agent request accepted",
|
||||
message=message,
|
||||
thread_id=thread_id,
|
||||
status="accepted",
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
|
||||
def _create_http_response(
|
||||
self,
|
||||
payload: dict[str, Any] | str,
|
||||
status_code: int,
|
||||
response_format: str,
|
||||
request_response_format: str,
|
||||
thread_id: str | None,
|
||||
) -> func.HttpResponse:
|
||||
"""Create the HTTP response using helper serializers for clarity."""
|
||||
if response_format == RESPONSE_FORMAT_TEXT:
|
||||
if request_response_format == REQUEST_RESPONSE_FORMAT_TEXT:
|
||||
return self._build_plain_text_response(payload=payload, status_code=status_code, thread_id=thread_id)
|
||||
|
||||
return self._build_json_response(payload=payload, status_code=status_code)
|
||||
@@ -650,13 +679,13 @@ class AgentFunctionApp(DFAppBase):
|
||||
) -> func.HttpResponse:
|
||||
"""Return a plain-text response with optional thread identifier header."""
|
||||
body_text = payload if isinstance(payload, str) else self._convert_payload_to_text(payload)
|
||||
headers = {"x-ms-thread-id": thread_id} if thread_id is not None else None
|
||||
return func.HttpResponse(body_text, status_code=status_code, mimetype="text/plain", headers=headers)
|
||||
headers = {THREAD_ID_HEADER: thread_id} if thread_id is not None else None
|
||||
return func.HttpResponse(body_text, status_code=status_code, mimetype=MIMETYPE_TEXT_PLAIN, headers=headers)
|
||||
|
||||
def _build_json_response(self, payload: dict[str, Any] | str, status_code: int) -> func.HttpResponse:
|
||||
"""Return the JSON response, serializing dictionaries as needed."""
|
||||
body_json = payload if isinstance(payload, str) else json.dumps(payload)
|
||||
return func.HttpResponse(body_json, status_code=status_code, mimetype="application/json")
|
||||
return func.HttpResponse(body_json, status_code=status_code, mimetype=MIMETYPE_APPLICATION_JSON)
|
||||
|
||||
def _convert_payload_to_text(self, payload: dict[str, Any]) -> str:
|
||||
"""Convert a structured payload into a human-readable text response."""
|
||||
@@ -702,18 +731,19 @@ class AgentFunctionApp(DFAppBase):
|
||||
normalized_content_type = self._extract_content_type(headers)
|
||||
body_parser, body_format = self._select_body_parser(normalized_content_type)
|
||||
prefers_json = self._accepts_json_response(headers)
|
||||
response_format = self._select_response_format(body_format=body_format, prefers_json=prefers_json)
|
||||
request_response_format = self._select_request_response_format(
|
||||
body_format=body_format, prefers_json=prefers_json
|
||||
)
|
||||
|
||||
req_body, message = body_parser(req)
|
||||
return req_body, message, response_format
|
||||
return req_body, message, request_response_format
|
||||
|
||||
def _extract_normalized_headers(self, req: func.HttpRequest) -> dict[str, str]:
|
||||
"""Create a lowercase header mapping from the incoming request."""
|
||||
headers: dict[str, str] = {}
|
||||
raw_headers = req.headers
|
||||
if isinstance(raw_headers, Mapping):
|
||||
header_mapping: Mapping[str, Any] = cast(Mapping[str, Any], raw_headers)
|
||||
for key, value in header_mapping.items():
|
||||
for key, value in raw_headers.items():
|
||||
if value is not None:
|
||||
headers[str(key).lower()] = str(value)
|
||||
return headers
|
||||
@@ -729,9 +759,9 @@ class AgentFunctionApp(DFAppBase):
|
||||
normalized_content_type: str,
|
||||
) -> tuple[Callable[[func.HttpRequest], tuple[dict[str, Any], str]], str]:
|
||||
"""Choose the body parser and declared body format."""
|
||||
if normalized_content_type in {"application/json"} or normalized_content_type.endswith("+json"):
|
||||
return self._parse_json_body, RESPONSE_FORMAT_JSON
|
||||
return self._parse_text_body, RESPONSE_FORMAT_TEXT
|
||||
if normalized_content_type in {MIMETYPE_APPLICATION_JSON} or normalized_content_type.endswith("+json"):
|
||||
return self._parse_json_body, REQUEST_RESPONSE_FORMAT_JSON
|
||||
return self._parse_text_body, REQUEST_RESPONSE_FORMAT_TEXT
|
||||
|
||||
@staticmethod
|
||||
def _accepts_json_response(headers: dict[str, str]) -> bool:
|
||||
@@ -742,16 +772,16 @@ class AgentFunctionApp(DFAppBase):
|
||||
|
||||
for value in accept_header.split(","):
|
||||
media_type = value.split(";")[0].strip().lower()
|
||||
if media_type == "application/json":
|
||||
if media_type == MIMETYPE_APPLICATION_JSON:
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _select_response_format(body_format: str, prefers_json: bool) -> str:
|
||||
def _select_request_response_format(body_format: str, prefers_json: bool) -> str:
|
||||
"""Combine body format and accept preference to determine response format."""
|
||||
if body_format == RESPONSE_FORMAT_JSON or prefers_json:
|
||||
return RESPONSE_FORMAT_JSON
|
||||
return RESPONSE_FORMAT_TEXT
|
||||
if body_format == REQUEST_RESPONSE_FORMAT_JSON or prefers_json:
|
||||
return REQUEST_RESPONSE_FORMAT_JSON
|
||||
return REQUEST_RESPONSE_FORMAT_TEXT
|
||||
|
||||
@staticmethod
|
||||
def _parse_json_body(req: func.HttpRequest) -> tuple[dict[str, Any], str]:
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Constants for Azure Functions Agent Framework integration."""
|
||||
|
||||
# Supported request/response formats and MIME types
|
||||
REQUEST_RESPONSE_FORMAT_JSON: str = "json"
|
||||
REQUEST_RESPONSE_FORMAT_TEXT: str = "text"
|
||||
MIMETYPE_APPLICATION_JSON: str = "application/json"
|
||||
MIMETYPE_TEXT_PLAIN: str = "text/plain"
|
||||
|
||||
# Field and header names
|
||||
THREAD_ID_FIELD: str = "thread_id"
|
||||
THREAD_ID_HEADER: str = "x-ms-thread-id"
|
||||
WAIT_FOR_RESPONSE_FIELD: str = "wait_for_response"
|
||||
WAIT_FOR_RESPONSE_HEADER: str = "x-ms-wait-for-response"
|
||||
|
||||
# Polling configuration
|
||||
DEFAULT_MAX_POLL_RETRIES: int = 30
|
||||
DEFAULT_POLL_INTERVAL_SECONDS: float = 1.0
|
||||
+1192
File diff suppressed because it is too large
Load Diff
@@ -11,14 +11,30 @@ import asyncio
|
||||
import inspect
|
||||
import json
|
||||
from collections.abc import AsyncIterable, Callable
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, cast
|
||||
|
||||
import azure.durable_functions as df
|
||||
from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, Role, get_logger
|
||||
from agent_framework import (
|
||||
AgentProtocol,
|
||||
AgentRunResponse,
|
||||
AgentRunResponseUpdate,
|
||||
ChatMessage,
|
||||
ErrorContent,
|
||||
Role,
|
||||
get_logger,
|
||||
)
|
||||
|
||||
from ._callbacks import AgentCallbackContext, AgentResponseCallbackProtocol
|
||||
from ._durable_agent_state import (
|
||||
DurableAgentState,
|
||||
DurableAgentStateData,
|
||||
DurableAgentStateEntry,
|
||||
DurableAgentStateMessage,
|
||||
DurableAgentStateRequest,
|
||||
DurableAgentStateResponse,
|
||||
)
|
||||
from ._models import AgentResponse, RunRequest
|
||||
from ._state import AgentState
|
||||
|
||||
logger = get_logger("agent_framework.azurefunctions.entities")
|
||||
|
||||
@@ -38,11 +54,11 @@ class AgentEntity:
|
||||
|
||||
Attributes:
|
||||
agent: The AgentProtocol instance
|
||||
state: The AgentState managing conversation history
|
||||
state: The DurableAgentState managing conversation history
|
||||
"""
|
||||
|
||||
agent: AgentProtocol
|
||||
state: AgentState
|
||||
state: DurableAgentState
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -56,11 +72,27 @@ class AgentEntity:
|
||||
callback: Optional callback invoked during streaming updates and final responses
|
||||
"""
|
||||
self.agent = agent
|
||||
self.state = AgentState()
|
||||
self.state = DurableAgentState()
|
||||
self.callback = callback
|
||||
|
||||
logger.debug(f"[AgentEntity] Initialized with agent type: {type(agent).__name__}")
|
||||
|
||||
def _is_error_response(self, entry: DurableAgentStateEntry) -> bool:
|
||||
"""Check if a conversation history entry is an error response.
|
||||
|
||||
Error responses should be kept in history for tracking but not sent to the agent
|
||||
since Azure OpenAI doesn't support 'error' content type.
|
||||
|
||||
Args:
|
||||
entry: A conversation history entry (DurableAgentStateEntry or dict)
|
||||
|
||||
Returns:
|
||||
True if the entry is a response containing error content, False otherwise
|
||||
"""
|
||||
if isinstance(entry, DurableAgentStateResponse):
|
||||
return entry.is_error
|
||||
return False
|
||||
|
||||
async def run_agent(
|
||||
self,
|
||||
context: df.DurableEntityContext,
|
||||
@@ -94,26 +126,27 @@ class AgentEntity:
|
||||
raise ValueError("RunRequest must include a thread_id")
|
||||
if not correlation_id:
|
||||
raise ValueError("RunRequest must include a correlation_id")
|
||||
role = run_request.role or Role.USER
|
||||
response_format = run_request.response_format
|
||||
enable_tool_calls = run_request.enable_tool_calls
|
||||
|
||||
logger.debug(f"[AgentEntity.run_agent] Received message: {message}")
|
||||
logger.debug(f"[AgentEntity.run_agent] Thread ID: {thread_id}")
|
||||
logger.debug(f"[AgentEntity.run_agent] Correlation ID: {correlation_id}")
|
||||
logger.debug(f"[AgentEntity.run_agent] Role: {role.value}")
|
||||
logger.debug(f"[AgentEntity.run_agent] Enable tool calls: {enable_tool_calls}")
|
||||
logger.debug(f"[AgentEntity.run_agent] Response format: {'provided' if response_format else 'none'}")
|
||||
state_request = DurableAgentStateRequest.from_run_request(run_request)
|
||||
self.state.data.conversation_history.append(state_request)
|
||||
|
||||
# Store message in history with role
|
||||
self.state.add_user_message(message, role=role, correlation_id=correlation_id)
|
||||
|
||||
logger.debug("[AgentEntity.run_agent] Executing agent...")
|
||||
logger.debug(f"[AgentEntity.run_agent] Received Message: {state_request}")
|
||||
|
||||
try:
|
||||
logger.debug("[AgentEntity.run_agent] Starting agent invocation")
|
||||
|
||||
run_kwargs: dict[str, Any] = {"messages": self.state.get_chat_messages()}
|
||||
# Build messages from conversation history, excluding error responses
|
||||
# Error responses are kept in history for tracking but not sent to the agent
|
||||
chat_messages: list[ChatMessage] = [
|
||||
m.to_chat_message()
|
||||
for entry in self.state.data.conversation_history
|
||||
if not self._is_error_response(entry)
|
||||
for m in entry.messages
|
||||
]
|
||||
|
||||
run_kwargs: dict[str, Any] = {"messages": chat_messages}
|
||||
if not enable_tool_calls:
|
||||
run_kwargs["tools"] = None
|
||||
if response_format:
|
||||
@@ -133,8 +166,8 @@ class AgentEntity:
|
||||
|
||||
response_text = None
|
||||
structured_response = None
|
||||
|
||||
response_str: str | None = None
|
||||
|
||||
try:
|
||||
if response_format:
|
||||
try:
|
||||
@@ -156,18 +189,19 @@ class AgentEntity:
|
||||
)
|
||||
response_text = "Error extracting response"
|
||||
|
||||
state_response = DurableAgentStateResponse.from_run_response(correlation_id, agent_run_response)
|
||||
self.state.data.conversation_history.append(state_response)
|
||||
|
||||
agent_response = AgentResponse(
|
||||
response=response_text,
|
||||
message=str(message),
|
||||
thread_id=str(thread_id),
|
||||
status="success",
|
||||
message_count=self.state.message_count,
|
||||
message_count=len(self.state.data.conversation_history),
|
||||
structured_response=structured_response,
|
||||
)
|
||||
result = agent_response.to_dict()
|
||||
|
||||
content = json.dumps(structured_response) if structured_response else (response_text or "")
|
||||
self.state.add_assistant_message(content, agent_run_response, correlation_id)
|
||||
logger.debug("[AgentEntity.run_agent] AgentRunResponse stored in conversation history")
|
||||
|
||||
return result
|
||||
@@ -181,12 +215,28 @@ class AgentEntity:
|
||||
logger.error(f"Error type: {type(exc).__name__}")
|
||||
logger.error(f"Full traceback:\n{error_traceback}")
|
||||
|
||||
# Create error message
|
||||
error_message = DurableAgentStateMessage.from_chat_message(
|
||||
ChatMessage(
|
||||
role=Role.ASSISTANT, contents=[ErrorContent(message=str(exc), error_code=type(exc).__name__)]
|
||||
)
|
||||
)
|
||||
|
||||
# Create and store error response in conversation history
|
||||
error_state_response = DurableAgentStateResponse(
|
||||
correlation_id=correlation_id,
|
||||
created_at=datetime.now(tz=timezone.utc),
|
||||
messages=[error_message],
|
||||
is_error=True,
|
||||
)
|
||||
self.state.data.conversation_history.append(error_state_response)
|
||||
|
||||
error_response = AgentResponse(
|
||||
response=f"Error: {exc!s}",
|
||||
message=str(message),
|
||||
thread_id=str(thread_id),
|
||||
status="error",
|
||||
message_count=self.state.message_count,
|
||||
message_count=len(self.state.data.conversation_history),
|
||||
error=str(exc),
|
||||
error_type=type(exc).__name__,
|
||||
)
|
||||
@@ -333,7 +383,7 @@ class AgentEntity:
|
||||
def reset(self, context: df.DurableEntityContext) -> None:
|
||||
"""Reset the entity state (clear conversation history)."""
|
||||
logger.debug("[AgentEntity.reset] Resetting entity state")
|
||||
self.state.reset()
|
||||
self.state.data = DurableAgentStateData(conversation_history=[])
|
||||
logger.debug("[AgentEntity.reset] State reset complete")
|
||||
|
||||
|
||||
@@ -362,7 +412,7 @@ def create_agent_entity(
|
||||
entity = AgentEntity(agent, callback)
|
||||
|
||||
if current_state is not None:
|
||||
entity.state.restore_state(current_state)
|
||||
entity.state = DurableAgentState.from_dict(current_state)
|
||||
logger.debug(
|
||||
"[entity_function] Restored entity from state (message_count: %s)", entity.state.message_count
|
||||
)
|
||||
@@ -392,8 +442,9 @@ def create_agent_entity(
|
||||
logger.error("[entity_function] Unknown operation: %s", operation)
|
||||
context.set_result({"error": f"Unknown operation: {operation}"})
|
||||
|
||||
logger.debug("State dict: %s", entity.state.to_dict())
|
||||
context.set_state(entity.state.to_dict())
|
||||
logger.debug(f"[entity_function] Operation {operation} completed successfully")
|
||||
logger.info(f"[entity_function] Operation {operation} completed successfully")
|
||||
|
||||
except Exception as exc:
|
||||
import traceback
|
||||
|
||||
@@ -17,6 +17,8 @@ from typing import TYPE_CHECKING, Any, cast
|
||||
import azure.durable_functions as df
|
||||
from agent_framework import AgentThread, Role
|
||||
|
||||
from ._constants import REQUEST_RESPONSE_FORMAT_TEXT
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover - type checking imports only
|
||||
from pydantic import BaseModel
|
||||
|
||||
@@ -278,35 +280,43 @@ class RunRequest:
|
||||
|
||||
Attributes:
|
||||
message: The message to send to the agent
|
||||
request_response_format: The desired response format (e.g., "text" or "json")
|
||||
role: The role of the message sender (user, system, or assistant)
|
||||
response_format: Optional Pydantic BaseModel type describing the structured response format
|
||||
enable_tool_calls: Whether to enable tool calls for this request
|
||||
thread_id: Optional thread ID for tracking
|
||||
correlation_id: Optional correlation ID for tracking the response to this specific request
|
||||
created_at: Optional timestamp when the request was created
|
||||
"""
|
||||
|
||||
message: str
|
||||
request_response_format: str
|
||||
role: Role = Role.USER
|
||||
response_format: type[BaseModel] | None = None
|
||||
enable_tool_calls: bool = True
|
||||
thread_id: str | None = None
|
||||
correlation_id: str | None = None
|
||||
created_at: str | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
request_response_format: str = REQUEST_RESPONSE_FORMAT_TEXT,
|
||||
role: Role | str | None = Role.USER,
|
||||
response_format: type[BaseModel] | None = None,
|
||||
enable_tool_calls: bool = True,
|
||||
thread_id: str | None = None,
|
||||
correlation_id: str | None = None,
|
||||
created_at: str | None = None,
|
||||
) -> None:
|
||||
self.message = message
|
||||
self.role = self.coerce_role(role)
|
||||
self.response_format = response_format
|
||||
self.request_response_format = request_response_format
|
||||
self.enable_tool_calls = enable_tool_calls
|
||||
self.thread_id = thread_id
|
||||
self.correlation_id = correlation_id
|
||||
self.created_at = created_at
|
||||
|
||||
@staticmethod
|
||||
def coerce_role(value: Role | str | None) -> Role:
|
||||
@@ -326,13 +336,17 @@ class RunRequest:
|
||||
"message": self.message,
|
||||
"enable_tool_calls": self.enable_tool_calls,
|
||||
"role": self.role.value,
|
||||
"request_response_format": self.request_response_format,
|
||||
}
|
||||
if self.response_format:
|
||||
result["response_format"] = _serialize_response_format(self.response_format)
|
||||
if self.thread_id:
|
||||
result["thread_id"] = self.thread_id
|
||||
if self.correlation_id:
|
||||
result["correlation_id"] = self.correlation_id
|
||||
result["correlationId"] = self.correlation_id
|
||||
if self.created_at:
|
||||
result["created_at"] = self.created_at
|
||||
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
@@ -340,11 +354,13 @@ class RunRequest:
|
||||
"""Create RunRequest from dictionary."""
|
||||
return cls(
|
||||
message=data.get("message", ""),
|
||||
request_response_format=data.get("request_response_format", REQUEST_RESPONSE_FORMAT_TEXT),
|
||||
role=cls.coerce_role(data.get("role")),
|
||||
response_format=_deserialize_response_format(data.get("response_format")),
|
||||
enable_tool_calls=data.get("enable_tool_calls", True),
|
||||
thread_id=data.get("thread_id"),
|
||||
correlation_id=data.get("correlation_id"),
|
||||
correlation_id=data.get("correlationId"),
|
||||
created_at=data.get("created_at"),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -1,179 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Agent State Management.
|
||||
|
||||
This module defines the AgentState class for managing conversation state and
|
||||
serializing agent framework responses.
|
||||
"""
|
||||
|
||||
from collections.abc import MutableMapping
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, cast
|
||||
|
||||
from agent_framework import AgentRunResponse, ChatMessage, Role, get_logger
|
||||
|
||||
logger = get_logger("agent_framework.azurefunctions.state")
|
||||
|
||||
|
||||
class AgentState:
|
||||
"""Manages agent conversation state using agent_framework types (ChatMessage, AgentRunResponse).
|
||||
|
||||
This class handles:
|
||||
- Conversation history tracking using ChatMessage objects
|
||||
- Agent response storage using AgentRunResponse objects with correlation IDs
|
||||
- State persistence and restoration
|
||||
- Message counting
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize empty agent state."""
|
||||
self.conversation_history: list[ChatMessage] = []
|
||||
self.last_response: str | None = None
|
||||
self.message_count: int = 0
|
||||
|
||||
def _current_timestamp(self) -> str:
|
||||
"""Return an ISO 8601 UTC timestamp."""
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
def add_user_message(
|
||||
self,
|
||||
content: str,
|
||||
role: Role = Role.USER,
|
||||
correlation_id: str | None = None,
|
||||
) -> None:
|
||||
"""Add a user message to the conversation history as a ChatMessage object.
|
||||
|
||||
Args:
|
||||
content: The message content
|
||||
role: The message role (user, system, etc.)
|
||||
correlation_id: Optional correlation identifier associated with the user message
|
||||
"""
|
||||
self.message_count += 1
|
||||
timestamp = self._current_timestamp()
|
||||
additional_props: MutableMapping[str, Any] = {"timestamp": timestamp}
|
||||
if correlation_id is not None:
|
||||
additional_props["correlation_id"] = correlation_id
|
||||
chat_message = ChatMessage(role=role, text=content, additional_properties=additional_props)
|
||||
self.conversation_history.append(chat_message)
|
||||
logger.debug(f"Added {role} ChatMessage to history (message #{self.message_count})")
|
||||
|
||||
def add_assistant_message(
|
||||
self, content: str, agent_response: AgentRunResponse, correlation_id: str | None = None
|
||||
) -> None:
|
||||
"""Add an assistant message to the conversation history with full agent response.
|
||||
|
||||
Args:
|
||||
content: The text content of the response
|
||||
agent_response: The AgentRunResponse object from the agent framework
|
||||
correlation_id: Optional correlation ID for tracking this response
|
||||
"""
|
||||
self.last_response = content
|
||||
timestamp = self._current_timestamp()
|
||||
serialized_response = self.serialize_response(agent_response)
|
||||
|
||||
# Create a ChatMessage for the assistant response
|
||||
# The agent_response already contains messages, but we store it as a custom ChatMessage
|
||||
# with the agent_response stored in additional_properties for full metadata preservation
|
||||
additional_props: dict[str, Any] = {
|
||||
"agent_response": serialized_response,
|
||||
"correlation_id": correlation_id,
|
||||
"timestamp": timestamp,
|
||||
"message_count": self.message_count,
|
||||
}
|
||||
chat_message = ChatMessage(role="assistant", text=content, additional_properties=additional_props)
|
||||
|
||||
self.conversation_history.append(chat_message)
|
||||
|
||||
logger.debug(
|
||||
f"Added assistant ChatMessage to history with AgentRunResponse metadata (correlation_id: {correlation_id})"
|
||||
)
|
||||
|
||||
def get_chat_messages(self) -> list[ChatMessage]:
|
||||
"""Return a copy of the full conversation history."""
|
||||
return list(self.conversation_history)
|
||||
|
||||
def try_get_agent_response(self, correlation_id: str) -> dict[str, Any] | None:
|
||||
"""Get an agent response by correlation ID.
|
||||
|
||||
Args:
|
||||
correlation_id: The correlation ID to look up
|
||||
|
||||
Returns:
|
||||
The agent response data if found, None otherwise
|
||||
"""
|
||||
for message in reversed(self.conversation_history):
|
||||
metadata = getattr(message, "additional_properties", {}) or {}
|
||||
if metadata.get("correlation_id") == correlation_id:
|
||||
return self._build_agent_response_payload(message, metadata)
|
||||
|
||||
return None
|
||||
|
||||
def serialize_response(self, response: AgentRunResponse) -> dict[str, Any]:
|
||||
"""Serialize an ``AgentRunResponse`` to a dictionary.
|
||||
|
||||
Args:
|
||||
response: The agent framework response object
|
||||
|
||||
Returns:
|
||||
Dictionary containing all response fields
|
||||
"""
|
||||
try:
|
||||
return response.to_dict()
|
||||
except Exception as exc: # pragma: no cover - defensive logging path
|
||||
logger.warning(f"Error serializing response: {exc}")
|
||||
return {"response": str(response), "serialization_error": str(exc)}
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Get the current state as a dictionary for persistence.
|
||||
|
||||
Returns:
|
||||
Dictionary containing conversation_history (as serialized ChatMessages),
|
||||
last_response, and message_count
|
||||
"""
|
||||
return {
|
||||
"conversation_history": [msg.to_dict() for msg in self.conversation_history],
|
||||
"last_response": self.last_response,
|
||||
"message_count": self.message_count,
|
||||
}
|
||||
|
||||
def restore_state(self, state: dict[str, Any]) -> None:
|
||||
"""Restore state from a dictionary, reconstructing ChatMessage objects.
|
||||
|
||||
Args:
|
||||
state: Dictionary containing conversation_history, last_response, and message_count
|
||||
"""
|
||||
# Restore conversation history as ChatMessage objects
|
||||
history_data = state.get("conversation_history", [])
|
||||
restored_history: list[ChatMessage] = []
|
||||
for raw_message in history_data:
|
||||
if isinstance(raw_message, dict):
|
||||
restored_history.append(ChatMessage.from_dict(cast(dict[str, Any], raw_message)))
|
||||
else:
|
||||
restored_history.append(cast(ChatMessage, raw_message))
|
||||
|
||||
self.conversation_history = restored_history
|
||||
|
||||
self.last_response = state.get("last_response")
|
||||
self.message_count = state.get("message_count", 0)
|
||||
logger.debug("Restored state: %s ChatMessages in history", len(self.conversation_history))
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Reset the state to empty."""
|
||||
self.conversation_history = []
|
||||
self.last_response = None
|
||||
self.message_count = 0
|
||||
logger.debug("State reset to empty")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""String representation of the state."""
|
||||
return f"AgentState(messages={self.message_count}, history_length={len(self.conversation_history)})"
|
||||
|
||||
def _build_agent_response_payload(self, message: ChatMessage, metadata: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Construct the agent response payload returned to callers."""
|
||||
return {
|
||||
"content": message.text,
|
||||
"agent_response": metadata.get("agent_response"),
|
||||
"message_count": metadata.get("message_count", self.message_count),
|
||||
"timestamp": metadata.get("timestamp"),
|
||||
"correlation_id": metadata.get("correlation_id"),
|
||||
}
|
||||
Reference in New Issue
Block a user