TRAPO

Microsoft Agent Framework admite la adición de funcionalidades de generación aumentada de recuperación (RAG) a los agentes fácilmente mediante la adición de proveedores de contexto de IA al agente.

Para conocer los patrones de conversación y sesión junto con la recuperación, consulte Información general sobre conversaciones y memoria.

Uso de TextSearchProvider

La TextSearchProvider clase es una implementación integrada de un proveedor de contexto RAG.

Se puede asociar fácilmente a mediante ChatClientAgent la AIContextProviderFactory opción para proporcionar funcionalidades rag al agente.

El generador es una función asincrónica que recibe un objeto de contexto y un token de cancelación.

// Create the AI agent with the TextSearchProvider as the AI context provider.
AIAgent agent = azureOpenAIClient
    .GetChatClient(deploymentName)
    .AsAIAgent(new ChatClientAgentOptions
    {
        ChatOptions = new() { Instructions = "You are a helpful support specialist for Contoso Outdoors. Answer questions using the provided context and cite the source document when available." },
        AIContextProviderFactory = (ctx, ct) => new ValueTask<AIContextProvider>(
            new TextSearchProvider(SearchAdapter, ctx.SerializedState, ctx.JsonSerializerOptions, textSearchOptions))
    });

TextSearchProvider requiere una función que proporcione los resultados de la búsqueda dada una consulta. Esto se puede implementar mediante cualquier tecnología de búsqueda, por ejemplo, Azure AI Search o un motor de búsqueda web.

Este es un ejemplo de una función de búsqueda simulada que devuelve resultados predefinidos en función de la consulta. SourceName y SourceLink son opcionales, pero si el agente lo usará para citar el origen de la información al responder a la pregunta del usuario.

static Task<IEnumerable<TextSearchProvider.TextSearchResult>> SearchAdapter(string query, CancellationToken cancellationToken)
{
    // The mock search inspects the user's question and returns pre-defined snippets
    // that resemble documents stored in an external knowledge source.
    List<TextSearchProvider.TextSearchResult> results = new();

    if (query.Contains("return", StringComparison.OrdinalIgnoreCase) || query.Contains("refund", StringComparison.OrdinalIgnoreCase))
    {
        results.Add(new()
        {
            SourceName = "Contoso Outdoors Return Policy",
            SourceLink = "https://contoso.com/policies/returns",
            Text = "Customers may return any item within 30 days of delivery. Items should be unused and include original packaging. Refunds are issued to the original payment method within 5 business days of inspection."
        });
    }

    return Task.FromResult<IEnumerable<TextSearchProvider.TextSearchResult>>(results);
}

Opciones de TextSearchProvider

TextSearchProvider se puede personalizar a través de la TextSearchProviderOptions clase . Este es un ejemplo de creación de opciones para ejecutar la búsqueda antes de cada invocación de modelo y mantener una breve ventana gradual del contexto de conversación.

TextSearchProviderOptions textSearchOptions = new()
{
    // Run the search prior to every model invocation and keep a short rolling window of conversation context.
    SearchTime = TextSearchProviderOptions.TextSearchBehavior.BeforeAIInvoke,
    RecentMessageMemoryLimit = 6,
};

La TextSearchProvider clase admite las siguientes opciones a través de la TextSearchProviderOptions clase .

Opción	Tipo	Description	Predeterminado
SearchTime	`TextSearchProviderOptions.TextSearchBehavior`	Indica cuándo se debe ejecutar la búsqueda. Hay dos opciones, cada vez que se invoca el agente o a petición a través de una llamada a función.	`TextSearchProviderOptions.TextSearchBehavior.BeforeAIInvoke`
FunctionToolName	`string`	Nombre de la herramienta de búsqueda expuesta al operar en modo a petición.	"Buscar"
FunctionToolDescription	`string`	Descripción de la herramienta de búsqueda expuesta al operar en modo a petición.	"Permite buscar información adicional para ayudar a responder a la pregunta del usuario".
ContextPrompt	`string`	El símbolo del sistema de contexto prefijo a los resultados al operar en `BeforeAIInvoke` modo.	"## Contexto adicional\nTenga en cuenta la siguiente información de los documentos de origen al responder al usuario:
CitasPrompt	`string`	La instrucción anexada después de los resultados para solicitar citas al operar en `BeforeAIInvoke` modo.	"Incluya citas al documento de origen con el nombre del documento y vincule si el nombre del documento y el vínculo están disponibles".
ContextFormatter	`Func<IList<TextSearchProvider.TextSearchResult>, string>`	Delegado opcional para personalizar completamente el formato de la lista de resultados al operar en `BeforeAIInvoke` modo. Si se proporciona y `ContextPromptCitationsPrompt` se omiten.	`null`
RecentMessageMemoryLimit	`int`	El número de mensajes de conversación recientes (tanto el usuario como el asistente) que se conservan en la memoria e incluyen al construir la entrada de búsqueda para `BeforeAIInvoke` las búsquedas.	`0` (deshabilitado)
RecentMessageRolesIncluded	`List<ChatRole>`	Lista de `ChatRole` tipos a los que se van a filtrar los mensajes recientes a al decidir qué mensajes recientes se van a incluir al construir la entrada de búsqueda.	`ChatRole.User`

Sugerencia

Consulte los ejemplos de .NET para obtener ejemplos completos de ejecución.

Agent Framework admite el uso de colecciones VectorStore del kernel semántico para proporcionar funcionalidades rag a los agentes. Esto se logra a través de la funcionalidad de puente que convierte las funciones de búsqueda semántica del kernel en herramientas de Agent Framework.

Creación de una herramienta de búsqueda desde VectorStore

El create_search_function método de una colección VectorStore de kernel semántico devuelve un KernelFunction objeto que se puede convertir en una herramienta de Agent Framework mediante .as_agent_framework_tool(). Use la documentación de conectores de almacén de vectores para aprender a configurar diferentes colecciones de almacenes vectoriales.

from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding
from semantic_kernel.connectors.azure_ai_search import AzureAISearchCollection
from semantic_kernel.functions import KernelParameterMetadata
from agent_framework.openai import OpenAIResponsesClient

# Define your data model
class SupportArticle:
    article_id: str
    title: str
    content: str
    category: str
    # ... other fields

# Create an Azure AI Search collection
collection = AzureAISearchCollection[str, SupportArticle](
    record_type=SupportArticle,
    embedding_generator=OpenAITextEmbedding()
)

async with collection:
    await collection.ensure_collection_exists()
    # Load your knowledge base articles into the collection
    # await collection.upsert(articles)

    # Create a search function from the collection
    search_function = collection.create_search_function(
        function_name="search_knowledge_base",
        description="Search the knowledge base for support articles and product information.",
        search_type="keyword_hybrid",
        parameters=[
            KernelParameterMetadata(
                name="query",
                description="The search query to find relevant information.",
                type="str",
                is_required=True,
                type_object=str,
            ),
            KernelParameterMetadata(
                name="top",
                description="Number of results to return.",
                type="int",
                default_value=3,
                type_object=int,
            ),
        ],
        string_mapper=lambda x: f"[{x.record.category}] {x.record.title}: {x.record.content}",
    )

    # Convert the search function to an Agent Framework tool
    search_tool = search_function.as_agent_framework_tool()

    # Create an agent with the search tool
    agent = OpenAIResponsesClient(model_id="gpt-4o").as_agent(
        instructions="You are a helpful support specialist. Use the search tool to find relevant information before answering questions. Always cite your sources.",
        tools=search_tool
    )

    # Use the agent with RAG capabilities
    response = await agent.run("How do I return a product?")
    print(response.text)

Importante

Esta característica requiere semantic-kernel la versión 1.38 o posterior.

Personalización del comportamiento de búsqueda

Puede personalizar la función de búsqueda con varias opciones:

# Create a search function with filtering and custom formatting
search_function = collection.create_search_function(
    function_name="search_support_articles",
    description="Search for support articles in specific categories.",
    search_type="keyword_hybrid",
    # Apply filters to restrict search scope
    filter=lambda x: x.is_published == True,
    parameters=[
        KernelParameterMetadata(
            name="query",
            description="What to search for in the knowledge base.",
            type="str",
            is_required=True,
            type_object=str,
        ),
        KernelParameterMetadata(
            name="category",
            description="Filter by category: returns, shipping, products, or billing.",
            type="str",
            type_object=str,
        ),
        KernelParameterMetadata(
            name="top",
            description="Maximum number of results to return.",
            type="int",
            default_value=5,
            type_object=int,
        ),
    ],
    # Customize how results are formatted for the agent
    string_mapper=lambda x: f"Article: {x.record.title}\nCategory: {x.record.category}\nContent: {x.record.content}\nSource: {x.record.article_id}",
)

Para obtener los detalles completos sobre los parámetros disponibles para create_search_function, consulte la documentación del kernel semántico.

Uso de varias funciones de búsqueda

Puede proporcionar varias herramientas de búsqueda a un agente para distintos dominios de conocimiento:

# Create search functions for different knowledge bases
product_search = product_collection.create_search_function(
    function_name="search_products",
    description="Search for product information and specifications.",
    search_type="semantic_hybrid",
    string_mapper=lambda x: f"{x.record.name}: {x.record.description}",
).as_agent_framework_tool()

policy_search = policy_collection.create_search_function(
    function_name="search_policies",
    description="Search for company policies and procedures.",
    search_type="keyword_hybrid",
    string_mapper=lambda x: f"Policy: {x.record.title}\n{x.record.content}",
).as_agent_framework_tool()

# Create an agent with multiple search tools
agent = chat_client.as_agent(
    instructions="You are a support agent. Use the appropriate search tool to find information before answering. Cite your sources.",
    tools=[product_search, policy_search]
)

También puede crear varias funciones de búsqueda desde la misma colección con diferentes descripciones y parámetros para proporcionar funcionalidades de búsqueda especializadas:

# Create multiple search functions from the same collection
# Generic search for broad queries
general_search = support_collection.create_search_function(
    function_name="search_all_articles",
    description="Search all support articles for general information.",
    search_type="semantic_hybrid",
    parameters=[
        KernelParameterMetadata(
            name="query",
            description="The search query.",
            type="str",
            is_required=True,
            type_object=str,
        ),
    ],
    string_mapper=lambda x: f"{x.record.title}: {x.record.content}",
).as_agent_framework_tool()

# Detailed lookup for specific article IDs
detail_lookup = support_collection.create_search_function(
    function_name="get_article_details",
    description="Get detailed information for a specific article by its ID.",
    search_type="keyword",
    top=1,
    parameters=[
        KernelParameterMetadata(
            name="article_id",
            description="The specific article ID to retrieve.",
            type="str",
            is_required=True,
            type_object=str,
        ),
    ],
    string_mapper=lambda x: f"Title: {x.record.title}\nFull Content: {x.record.content}\nLast Updated: {x.record.updated_date}",
).as_agent_framework_tool()

# Create an agent with both search functions
agent = chat_client.as_agent(
    instructions="You are a support agent. Use search_all_articles for general queries and get_article_details when you need full details about a specific article.",
    tools=[general_search, detail_lookup]
)

Este enfoque permite al agente elegir la estrategia de búsqueda más adecuada en función de la consulta del usuario.

Ejemplo completo

# Copyright (c) Microsoft. All rights reserved.

import asyncio
from collections.abc import MutableSequence, Sequence
from typing import Any

from agent_framework import Agent, BaseContextProvider, Context, Message, SupportsChatGetResponse
from agent_framework.azure import AzureAIClient
from azure.identity.aio import AzureCliCredential
from pydantic import BaseModel


class UserInfo(BaseModel):
    name: str | None = None
    age: int | None = None


class UserInfoMemory(BaseContextProvider):
    def __init__(self, client: SupportsChatGetResponse, user_info: UserInfo | None = None, **kwargs: Any):
        """Create the memory.

        If you pass in kwargs, they will be attempted to be used to create a UserInfo object.
        """

        self._chat_client = client
        if user_info:
            self.user_info = user_info
        elif kwargs:
            self.user_info = UserInfo.model_validate(kwargs)
        else:
            self.user_info = UserInfo()

    async def invoked(
        self,
        request_messages: Message | Sequence[Message],
        response_messages: Message | Sequence[Message] | None = None,
        invoke_exception: Exception | None = None,
        **kwargs: Any,
    ) -> None:
        """Extract user information from messages after each agent call."""
        # Check if we need to extract user info from user messages
        user_messages = [msg for msg in request_messages if hasattr(msg, "role") and msg.role == "user"]  # type: ignore

        if (self.user_info.name is None or self.user_info.age is None) and user_messages:
            try:
                # Use the chat client to extract structured information
                result = await self._chat_client.get_response(
                    messages=request_messages,  # type: ignore
                    instructions="Extract the user's name and age from the message if present. "
                    "If not present return nulls.",
                    options={"response_format": UserInfo},
                )

                # Update user info with extracted data
                try:
                    extracted = result.value
                    if self.user_info.name is None and extracted.name:
                        self.user_info.name = extracted.name
                    if self.user_info.age is None and extracted.age:
                        self.user_info.age = extracted.age
                except Exception:
                    pass  # Failed to extract, continue without updating

            except Exception:
                pass  # Failed to extract, continue without updating

    async def invoking(self, messages: Message | MutableSequence[Message], **kwargs: Any) -> Context:
        """Provide user information context before each agent call."""
        instructions: list[str] = []

        if self.user_info.name is None:
            instructions.append(
                "Ask the user for their name and politely decline to answer any questions until they provide it."
            )
        else:
            instructions.append(f"The user's name is {self.user_info.name}.")

        if self.user_info.age is None:
            instructions.append(
                "Ask the user for their age and politely decline to answer any questions until they provide it."
            )
        else:
            instructions.append(f"The user's age is {self.user_info.age}.")

        # Return context with additional instructions
        return Context(instructions=" ".join(instructions))

    def serialize(self) -> str:
        """Serialize the user info for thread persistence."""
        return self.user_info.model_dump_json()


async def main():
    async with AzureCliCredential() as credential:
        client = AzureAIClient(credential=credential)

        # Create the memory provider
        memory_provider = UserInfoMemory(client)

        # Create the agent with memory
        async with Agent(
            client=client,
            instructions="You are a friendly assistant. Always address the user by their name.",
            context_providers=[memory_provider],
        ) as agent:
            # Create a new thread for the conversation
            thread = agent.create_session()

            print(await agent.run("Hello, what is the square root of 9?", session=thread))
            print(await agent.run("My name is Ruaidhrí", session=thread))
            print(await agent.run("I am 20 years old", session=thread))

            # Access the memory component and inspect the memories
            user_info_memory = memory_provider
            if user_info_memory:
                print()
                print(f"MEMORY - User Name: {user_info_memory.user_info.name}")  # type: ignore
                print(f"MEMORY - User Age: {user_info_memory.user_info.age}")  # type: ignore


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

# Copyright (c) Microsoft. All rights reserved.

import asyncio
from collections.abc import MutableSequence, Sequence
from typing import Any

from agent_framework import Agent, BaseContextProvider, Context, Message, SupportsChatGetResponse
from agent_framework.azure import AzureAIClient
from azure.identity.aio import AzureCliCredential
from pydantic import BaseModel


class UserInfo(BaseModel):
    name: str | None = None
    age: int | None = None


class UserInfoMemory(BaseContextProvider):
    def __init__(self, client: SupportsChatGetResponse, user_info: UserInfo | None = None, **kwargs: Any):
        """Create the memory.

        If you pass in kwargs, they will be attempted to be used to create a UserInfo object.
        """

        self._chat_client = client
        if user_info:
            self.user_info = user_info
        elif kwargs:
            self.user_info = UserInfo.model_validate(kwargs)
        else:
            self.user_info = UserInfo()

    async def invoked(
        self,
        request_messages: Message | Sequence[Message],
        response_messages: Message | Sequence[Message] | None = None,
        invoke_exception: Exception | None = None,
        **kwargs: Any,
    ) -> None:
        """Extract user information from messages after each agent call."""
        # Check if we need to extract user info from user messages
        user_messages = [msg for msg in request_messages if hasattr(msg, "role") and msg.role == "user"]  # type: ignore

        if (self.user_info.name is None or self.user_info.age is None) and user_messages:
            try:
                # Use the chat client to extract structured information
                result = await self._chat_client.get_response(
                    messages=request_messages,  # type: ignore
                    instructions="Extract the user's name and age from the message if present. "
                    "If not present return nulls.",
                    options={"response_format": UserInfo},
                )

                # Update user info with extracted data
                try:
                    extracted = result.value
                    if self.user_info.name is None and extracted.name:
                        self.user_info.name = extracted.name
                    if self.user_info.age is None and extracted.age:
                        self.user_info.age = extracted.age
                except Exception:
                    pass  # Failed to extract, continue without updating

            except Exception:
                pass  # Failed to extract, continue without updating

    async def invoking(self, messages: Message | MutableSequence[Message], **kwargs: Any) -> Context:
        """Provide user information context before each agent call."""
        instructions: list[str] = []

        if self.user_info.name is None:
            instructions.append(
                "Ask the user for their name and politely decline to answer any questions until they provide it."
            )
        else:
            instructions.append(f"The user's name is {self.user_info.name}.")

        if self.user_info.age is None:
            instructions.append(
                "Ask the user for their age and politely decline to answer any questions until they provide it."
            )
        else:
            instructions.append(f"The user's age is {self.user_info.age}.")

        # Return context with additional instructions
        return Context(instructions=" ".join(instructions))

    def serialize(self) -> str:
        """Serialize the user info for thread persistence."""
        return self.user_info.model_dump_json()


async def main():
    async with AzureCliCredential() as credential:
        client = AzureAIClient(credential=credential)

        # Create the memory provider
        memory_provider = UserInfoMemory(client)

        # Create the agent with memory
        async with Agent(
            client=client,
            instructions="You are a friendly assistant. Always address the user by their name.",
            context_providers=[memory_provider],
        ) as agent:
            # Create a new thread for the conversation
            thread = agent.create_session()

            print(await agent.run("Hello, what is the square root of 9?", session=thread))
            print(await agent.run("My name is Ruaidhrí", session=thread))
            print(await agent.run("I am 20 years old", session=thread))

            # Access the memory component and inspect the memories
            user_info_memory = memory_provider
            if user_info_memory:
                print()
                print(f"MEMORY - User Name: {user_info_memory.user_info.name}")  # type: ignore
                print(f"MEMORY - User Age: {user_info_memory.user_info.age}")  # type: ignore


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

Conectores vectorstore admitidos

Este patrón funciona con cualquier conector de VectorStore de kernel semántico, entre los que se incluyen:

Azure AI Search (AzureAISearchCollection)
Qdrant (QdrantCollection)
Pinecone (PineconeCollection)
Redis (RedisCollection)
Weaviate (WeaviateCollection)
In-Memory (InMemoryVectorStoreCollection)
Y más

Cada conector proporciona el mismo create_search_function método que se puede puenter a las herramientas de Agent Framework, lo que le permite elegir la base de datos vectorial que mejor se adapte a sus necesidades. Consulte la lista completa aquí.

Pasos siguientes

Agentes declarativos

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-02-13