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.")
aoai = AzureOpenAI( azure_endpoint="https://<your-foundry-resource>.openai.azure.com", api_key="<key>", api_version="2024-10-21", ) search_client = SearchClient( endpoint=endpoint, index_name="support-docs-index", credential=AzureKeyCredential("<admin-key>"), )
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
if msg.tool_calls: messages.append(msg) for call in msg.tool_calls: args = json.loads(call.function.arguments) if call.function.name == "search_docs": result = search_docs(args) elif call.function.name == "get_order": args["customer_id"] = customer_id # 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
return msg.content
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
Source
This article discusses content originally published by at Dev.to.
