Grounded Search Agents: Microsoft Foundry + Azure AI Search + Cosmos DB + Function Calling

Summary

Why "search" isn't enough anymore

Architecture overview

Step 1 — Provision and index documents in Azure AI Search

endpoint = "https://<your-search-service>.search.windows.net" credential = AzureKeyCredential("<admin-key>") index_client = SearchIndexClient(endpoint, credential)

fields = [ SimpleField(name="id", type=SearchFieldDataType.String, key=True), SearchableField(name="content", type=SearchFieldDataType.String), SimpleField(name="source_url", type=SearchFieldDataType.String), SearchField( name="content_vector", type=SearchFieldDataType.Collection(SearchFieldDataType.Single), searchable=True, vector_search_dimensions=1536, vector_search_profile_name="default-profile", ), ]

vector_search = VectorSearch( profiles=[VectorSearchProfile(name="default-profile", algorithm_configuration_name="hnsw-config")], algorithms=[HnswAlgorithmConfiguration(name="hnsw-config")], )

semantic_config = SemanticConfiguration( name="default-semantic", prioritized_fields=SemanticPrioritizedFields( content_fields=[SemanticField(field_name="content")] ), )

index = SearchIndex( name="support-docs-index", fields=fields, vector_search=vector_search, semantic_search=SemanticSearch(configurations=[semantic_config]), )

index_client.create_or_update_index(index) print("Index created.")

def chunk_text(text: str, max_tokens: int = 512, overlap: int = 80): words = text.split() step = max_tokens - overlap return [" ".join(words[i:i + max_tokens]) for i in range(0, len(words), step)]

def embed(text: str) -> list[float]: resp = aoai.embeddings.create(model="text-embedding-3-small", input=text) return resp.data[0].embedding

def ingest_document(text: str, source_url: str): docs = [] for chunk in chunk_text(text): docs.append({ "id": str(uuid.uuid4()), "content": chunk, "source_url": source_url, "content_vector": embed(chunk), }) search_client.upload_documents(documents=docs)

with open("return_policy.txt") as f: ingest_document(f.read(), source_url="policies/return_policy.txt")

Step 2 — Structured lookups in Cosmos DB

client = CosmosClient("<cosmos-endpoint>", credential="<cosmos-key>") db = client.create_database_if_not_exists("retail") orders = db.create_container_if_not_exists( id="orders", partition_key=PartitionKey(path="/customer_id"), offer_throughput=400, )

orders.upsert_item({ "id": "4471", "customer_id": "cust_9001", "status": "delayed", "delay_reason": "carrier weather disruption", "eta": "2026-08-02", })

Step 3 — The agent loop with function calling

aoai = AzureOpenAI( azure_endpoint="https://<your-foundry-resource>.openai.azure.com", api_key="<key>", api_version="2024-10-21", ) search_client = SearchClient(endpoint, "support-docs-index", AzureKeyCredential("<admin-key>")) cosmos = CosmosClient("<cosmos-endpoint>", credential="<cosmos-key>") orders_container = cosmos.get_database_client("retail").get_container_client("orders")

tools = [ { "type": "function", "function": { "name": "search_docs", "description": "Semantic search over policy and support documents. Returns passages with citations.", "parameters": { "type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "get_order", "description": "Look up a customer order by ID for status, ETA, and delay reason.", "parameters": { "type": "object", "properties": { "order_id": {"type": "string"}, "customer_id": {"type": "string"}, }, "required": ["order_id", "customer_id"], }, }, }, ]

def search_docs(query: str): results = search_client.search( search_text=query, query_type="semantic", semantic_configuration_name="default-semantic", top=3, ) return [{"content": r["content"], "source": r["source_url"]} for r in results]

def get_order(order_id: str, customer_id: str): item = orders_container.read_item(item=order_id, partition_key=customer_id) return {"status": item["status"], "reason": item.get("delay_reason"), "eta": item.get("eta")}

def run_agent(user_message: str, customer_id: str, history: list): messages = history + [{"role": "user", "content": user_message}] response = aoai.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto", ) msg = response.choices[0].message

<span class="k">if</span> <span class="n">msg</span><span class="p">.</span><span class="n">tool_calls</span><span class="p">:</span>
    <span class="n">messages</span><span class="p">.</span><span class="nf">append</span><span class="p">(</span><span class="n">msg</span><span class="p">)</span>
    <span class="k">for</span> <span class="n">call</span> <span class="ow">in</span> <span class="n">msg</span><span class="p">.</span><span class="n">tool_calls</span><span class="p">:</span>
        <span class="n">args</span> <span class="o">=</span> <span class="n">json</span><span class="p">.</span><span class="nf">loads</span><span class="p">(</span><span class="n">call</span><span class="p">.</span><span class="n">function</span><span class="p">.</span><span class="n">arguments</span><span class="p">)</span>
        <span class="k">if</span> <span class="n">call</span><span class="p">.</span><span class="n">function</span><span class="p">.</span><span class="n">name</span> <span class="o">==</span> <span class="sh">"</span><span class="s">search_docs</span><span class="sh">"</span><span class="p">:</span>
            <span class="n">result</span> <span class="o">=</span> <span class="nf">search_docs</span><span class="p">(</span><span class="o">**</span><span class="n">args</span><span class="p">)</span>
        <span class="k">elif</span> <span class="n">call</span><span class="p">.</span><span class="n">function</span><span class="p">.</span><span class="n">name</span> <span class="o">==</span> <span class="sh">"</span><span class="s">get_order</span><span class="sh">"</span><span class="p">:</span>
            <span class="n">args</span><span class="p">[</span><span class="sh">"</span><span class="s">customer_id</span><span class="sh">"</span><span class="p">]</span> <span class="o">=</span> <span class="n">customer_id</span>  <span class="c1"># enforce, don't trust model-provided id

result = get_order(**args) messages.append({ "role": "tool", "tool_call_id": call.id, "content": json.dumps(result), }) final = aoai.chat.completions.create(model="gpt-4o", messages=messages) return final.choices[0].message.content

<span class="k">return</span> <span class="n">msg</span><span class="p">.</span><span class="n">content</span>

Step 4 — Session state in Cosmos DB

def save_history(session_id: str, messages: list): container = cosmos.get_database_client("retail").get_container_client("sessions") container.upsert_item({"id": session_id, "messages": messages})

Comparing retrieval strategies

Production considerations

References