TRAPO

O Microsoft Agent Framework dá suporte à adição de recursos de RAG (Geração Aumentada de Recuperação) aos agentes facilmente adicionando provedores de contexto de IA ao agente.

Para obter padrões de conversa/sessão juntamente com a recuperação, consulte a visão geral de Conversas & Memória.

Usando TextSearchProvider

A TextSearchProvider classe é uma implementação pronta para uso de um provedor de contexto RAG.

Ele pode ser facilmente anexado a um ChatClientAgent uso da opção AIContextProviderFactory para fornecer recursos RAG ao agente.

A fábrica é uma função assíncrona que recebe um objeto de contexto e um token de cancelamento.

// 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))
    });

A TextSearchProvider função requer uma função que fornece os resultados da pesquisa dada uma consulta. Isso pode ser implementado usando qualquer tecnologia de pesquisa, por exemplo, o Azure AI Search ou um mecanismo de pesquisa na Web.

Aqui está um exemplo de uma função de pesquisa simulada que retorna resultados predefinidos com base na consulta. SourceName e SourceLink são opcionais, mas se fornecidos serão usados pelo agente para citar a origem das informações ao responder à pergunta do usuário.

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);
}

Opções textSearchProvider

Pode TextSearchProvider ser personalizado por meio da TextSearchProviderOptions classe. Aqui está um exemplo de como criar opções para executar a pesquisa antes de cada invocação de modelo e manter uma breve janela sem interrupção do contexto de conversa.

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,
};

A TextSearchProvider classe dá suporte às seguintes opções por meio da TextSearchProviderOptions classe.

Opção	Tipo	DESCRIÇÃO	Padrão
SearchTime	`TextSearchProviderOptions.TextSearchBehavior`	Indica quando a pesquisa deve ser executada. Há duas opções, cada vez que o agente é invocado ou sob demanda por meio de chamada de função.	`TextSearchProviderOptions.TextSearchBehavior.BeforeAIInvoke`
FunctionToolName	`string`	O nome da ferramenta de pesquisa exposta ao operar no modo sob demanda.	"Pesquisar"
FunctionToolDescription	`string`	A descrição da ferramenta de pesquisa exposta ao operar no modo sob demanda.	"Permite pesquisar informações adicionais para ajudar a responder à pergunta do usuário."
ContextPrompt	`string`	O prompt de contexto prefixado nos resultados ao operar no `BeforeAIInvoke` modo.	"## Contexto Adicional\nConsidere as seguintes informações de documentos de origem ao responder ao usuário:"
CitationsPrompt	`string`	A instrução acrescentada após os resultados para solicitar citações ao operar no `BeforeAIInvoke` modo.	"Inclua citações ao documento de origem com o nome do documento e o link se o nome e o link do documento estiverem disponíveis."
ContextFormatter	`Func<IList<TextSearchProvider.TextSearchResult>, string>`	Delegado opcional para personalizar totalmente a formatação da lista de resultados ao operar no `BeforeAIInvoke` modo. Se fornecido e `ContextPromptCitationsPrompt` ignorado.	`null`
RecentMessageMemoryLimit	`int`	O número de mensagens de conversa recentes (usuário e assistente) para manter na memória e incluir ao construir a entrada de pesquisa para `BeforeAIInvoke` pesquisas.	`0` (desabilitado)
RecentMessageRolesIncluded	`List<ChatRole>`	A lista de `ChatRole` tipos para os quais filtrar mensagens recentes ao decidir quais mensagens recentes incluir ao construir a entrada de pesquisa.	`ChatRole.User`

Dica

Consulte os exemplos do .NET para obter exemplos executáveis completos.

O Agent Framework dá suporte ao uso das coleções VectorStore do Kernel Semântico para fornecer recursos de RAG aos agentes. Isso é obtido por meio da funcionalidade de ponte que converte funções de pesquisa de Kernel Semântico em ferramentas do Agent Framework.

Criando uma ferramenta de pesquisa do VectorStore

O create_search_function método de uma coleção Semantic Kernel VectorStore retorna um KernelFunction que pode ser convertido em uma ferramenta do Agent Framework usando .as_agent_framework_tool(). Use a documentação de conectores do repositório de vetores para saber como configurar diferentes coleções de repositórios de vetores.

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

Esse recurso requer a semantic-kernel versão 1.38 ou superior.

Personalizando o comportamento da pesquisa

Você pode personalizar a função de pesquisa com várias opções:

# 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 obter os detalhes completos sobre os parâmetros disponíveis, create_search_functionconsulte a documentação do Kernel Semântico.

Usando várias funções de pesquisa

Você pode fornecer várias ferramentas de pesquisa a um agente para domínios de conhecimento diferentes:

# 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]
)

Você também pode criar várias funções de pesquisa da mesma coleção com diferentes descrições e parâmetros para fornecer recursos de pesquisa especializados:

# 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]
)

Essa abordagem permite que o agente escolha a estratégia de pesquisa mais apropriada com base na consulta do usuário.

Exemplo 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 com suporte

Esse padrão funciona com qualquer conector Semântico do Kernel VectorStore, incluindo:

Azure AI Search (AzureAISearchCollection)
Qdrant (QdrantCollection)
Pinecone (PineconeCollection)
Redis (RedisCollection)
Weaviate (WeaviateCollection)
In-Memory (InMemoryVectorStoreCollection)
E muito mais

Cada conector fornece o mesmo create_search_function método que pode ser conectado às ferramentas do Agent Framework, permitindo que você escolha o banco de dados vetor que melhor atenda às suas necessidades. Veja a lista completa aqui.

Próximas etapas

Agentes Declarativos

Comentários

Esta página foi útil?

Last updated on 2026-02-13