mirror of
https://github.com/microsoft/semantic-kernel.git
synced 2026-10-12 11:19:56 +00:00
ふぁーandEvan Mattson
d3318adfc2
### Motivation and Context
I confirmed that Code Interpreter can run via Azure Responses API in
Semantic Kernel, but I found two issues: an error occurs when a file is
created inside the Code Interpreter, and the response lacks parameters
required to access the created file.
### Description
#### Bug 1: ValidationError for AnnotationContent
Semantic Kernel uses the following enum for
`AnnotationContent.content_type`:
```python
@experimental
class CitationType(str, Enum):
"""Citation type."""
URL_CITATION = "url_citation"
FILE_PATH = "file_path"
FILE_CITATION = "file_citation"
```
However, the Responses API returns file citations with a `type` like
this:
```json
{"container_id": "cntr_68a6c50135d88190b6fdc062051155b50847ddec73d3be1b", "end_index": 94, "file_id": "cfile_68a6c67a7fcc8190889d7e1789d485e2", "filename": "sample.txt", "start_index": 66, "type": "container_file_citation"}
```
There is no matching enum member on the Semantic Kernel side. The
`CitationType` should include `CONTAINER_FILE_CITATION`, for example:
```python
@experimental
class CitationType(str, Enum):
"""Citation type."""
URL_CITATION = "url_citation"
FILE_PATH = "file_path"
FILE_CITATION = "file_citation"
CONTAINER_FILE_CITATION = "container_file_citation"
```
#### Bug 2: Missing parameters to access generated files
The response returned by the agent shows a `file_id`, but lacks the
`container_id` and `filename` required to download the file (the
filename is needed to replace `sandbox:/mnt/data/sample.txt`):
Example call:
```python
r = await agent.get_response(cast(list, history.messages), thread=thread)
print(r.content.items)
```
Example output:
```python
[
TextContent(inner_content=None, ai_model_id=None, metadata={}, content_type='text', text='<user_prompt>', encoding=None),
TextContent(inner_content=None, ai_model_id=None, metadata={}, content_type='text', text='<assistant_message>\n\n[sample.txt](sandbox:/mnt/data/sample.txt)', encoding=None),
AnnotationContent(inner_content=None, ai_model_id=None, metadata={}, content_type='annotation', file_id='cfile_68a6c67a7fcc8190889d7e1789d485e2', quote=None, start_index=88, end_index=116, url=None, title=None, citation_type=<CitationType.CONTAINER_FILE_CITATION: 'container_file_citation'>)
]
```
To include these values in the parsed result, `AnnotationContent` in
`semantic_kernel/contents/annotation_content.py` should expose
`container_id` and `filename`, and map the incoming `type` to
`citation_type`. Example suggested structure:
```python
@experimental
class AnnotationContent(KernelContent):
"""Annotation content."""
content_type: Literal[ContentTypes.ANNOTATION_CONTENT] = Field(ANNOTATION_CONTENT_TAG, init=False) # type: ignore
tag: ClassVar[str] = ANNOTATION_CONTENT_TAG
file_id: str | None = None
quote: str | None = None
start_index: int | None = None
end_index: int | None = None
url: str | None = None
title: str | None = None
# added
container_id: str | None = None
filename: str | None = None
citation_type: CitationType | None = Field(None, alias="type")
model_config = ConfigDict(
extra="ignore",
populate_by_name=True,
)
def __str__(self) -> str:
...
def to_element(self) -> Element:
...
@classmethod
def from_element(cls: type[_T], element: Element) -> _T:
...
def to_dict(self) -> dict[str, Any]:
...
```
Summary: add the missing enum value `container_file_citation` and extend
`AnnotationContent` to include `container_id` and `filename` (and map
`type` to `citation_type`) so file references returned by the Azure
Responses API are fully usable.
#### Code
```python
client: AsyncAzureOpenAI = AsyncAzureOpenAI(
base_url=urljoin(server_settings.AZURE_OPENAI_COMPLETION_ENDPOINT.rstrip("/") + "/", "openai/v1/"),
api_key=server_settings.AZURE_OPENAI_COMPLETION_API_KEY,
api_version="preview",
)
async def run(history: ChatHistory) -> dict[str, Any]:
with open("sample.txt", "rb") as fp:
f = await client.files.create(file=fp, purpose="assistants")
agent = AzureResponsesAgent(
ai_model_id=server_settings.AZURE_OPENAI_COMPLETION_DEPLOYMENT_NAME,
client=client,
name="name",
function_choice_behavior=FunctionChoiceBehavior.Required(),
tools=[
CodeInterpreter(
type="code_interpreter",
container=CodeInterpreterContainerCodeInterpreterToolAuto(type="auto", file_ids=[f.id]),
)
],
)
thread = ResponsesAgentThread(client, history, previous_response_id=None, enable_store=True)
r = await agent.get_response(cast(list, history.messages), thread=thread)
print(thread.id, r.content.items)
annotation_content = [item for item in r.content.items if isinstance(item, AnnotationContent)]
for ac in annotation_content:
print(ac.file_id, ac.filename, ac.container_id)
```
### Contribution Checklist
<!-- Before submitting this PR, please make sure: -->
- [x] The code builds clean without any errors or warnings
- [x] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [x] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄
Co-authored-by: Evan Mattson <[email protected]>
94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
import logging
|
|
from enum import Enum
|
|
from typing import Any, ClassVar, Literal, TypeVar
|
|
from xml.etree.ElementTree import Element # nosec
|
|
|
|
from pydantic import ConfigDict, Field
|
|
|
|
from semantic_kernel.contents.const import ANNOTATION_CONTENT_TAG, ContentTypes
|
|
from semantic_kernel.contents.kernel_content import KernelContent
|
|
from semantic_kernel.utils.feature_stage_decorator import experimental
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_T = TypeVar("_T", bound="AnnotationContent")
|
|
|
|
|
|
@experimental
|
|
class CitationType(str, Enum):
|
|
"""Citation type."""
|
|
|
|
URL_CITATION = "url_citation"
|
|
FILE_PATH = "file_path"
|
|
FILE_CITATION = "file_citation"
|
|
CONTAINER_FILE_CITATION = "container_file_citation"
|
|
|
|
|
|
@experimental
|
|
class AnnotationContent(KernelContent):
|
|
"""Annotation content."""
|
|
|
|
content_type: Literal[ContentTypes.ANNOTATION_CONTENT] = Field(ANNOTATION_CONTENT_TAG, init=False) # type: ignore
|
|
tag: ClassVar[str] = ANNOTATION_CONTENT_TAG
|
|
file_id: str | None = None
|
|
quote: str | None = None
|
|
start_index: int | None = None
|
|
end_index: int | None = None
|
|
url: str | None = None
|
|
title: str | None = None
|
|
citation_type: CitationType | None = Field(None, alias="type")
|
|
container_id: str | None = None
|
|
filename: str | None = None
|
|
|
|
model_config = ConfigDict(
|
|
extra="ignore",
|
|
populate_by_name=True,
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
"""Return the string representation of the annotation content."""
|
|
ctype = self.citation_type.value if self.citation_type else None
|
|
return f"AnnotationContent(type={ctype}, file_id={self.file_id}, url={self.url}, quote={self.quote}, start_index={self.start_index}, end_index={self.end_index})" # noqa: E501
|
|
|
|
def to_element(self) -> Element:
|
|
"""Convert the annotation content to an Element."""
|
|
element = Element(self.tag)
|
|
if self.citation_type:
|
|
element.set("type", self.citation_type)
|
|
if self.file_id:
|
|
element.set("file_id", self.file_id)
|
|
if self.quote:
|
|
element.set("quote", self.quote)
|
|
if self.start_index is not None:
|
|
element.set("start_index", str(self.start_index))
|
|
if self.end_index is not None:
|
|
element.set("end_index", str(self.end_index))
|
|
if self.url is not None:
|
|
element.set("url", self.url)
|
|
if self.title is not None:
|
|
element.set("title", self.title)
|
|
return element
|
|
|
|
@classmethod
|
|
def from_element(cls: type[_T], element: Element) -> _T:
|
|
"""Create an instance from an Element."""
|
|
return cls(
|
|
type=element.get("type"),
|
|
file_id=element.get("file_id"),
|
|
quote=element.get("quote"),
|
|
start_index=int(element.get("start_index")) if element.get("start_index") else None, # type: ignore
|
|
end_index=int(element.get("end_index")) if element.get("end_index") else None, # type: ignore
|
|
url=element.get("url") if element.get("url") else None, # type: ignore
|
|
title=element.get("title") if element.get("title") else None, # type: ignore
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Convert the instance to a dictionary."""
|
|
ctype = self.citation_type.value if self.citation_type else None
|
|
return {
|
|
"type": "text",
|
|
"text": f"type={ctype}, {self.file_id or self.url} {self.quote} (Start Index={self.start_index}->End Index={self.end_index})", # noqa: E501
|
|
}
|