【智能体agent】入门之--4.1 autogen agentic rag

every blog every motto: You can do more than you think.
https://blog.csdn.net/weixin_39190382?type=blog

0. 前言

auto agentic rag使用

1. 正文

import os
import time
import asyncio
from typing import List, Dict
from dotenv import load_dotenvfrom autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_agentchat.messages import TextMessage
from azure.core.credentials import AzureKeyCredential
from autogen_ext.models.azure import AzureAIChatCompletionClientimport chromadbload_dotenv()

True

创建client

client = AzureAIChatCompletionClient(model="gpt-4o-mini",endpoint="https://models.inference.ai.azure.com",credential=AzureKeyCredential(os.getenv("GITHUB_TOKEN")),model_info={"json_output": True,"function_calling": True,"vision": True,"family": "unknown",},
)

向量数据库初始化

Initialize ChromaDB with persistent storage

chroma_client = chromadb.PersistentClient(path=“./chroma_db”)
collection = chroma_client.create_collection(
name=“travel_documents”,
metadata={“description”: “travel_service”},
get_or_create=True
)

Enhanced sample documents

documents = [
“Contoso Travel offers luxury vacation packages to exotic destinations worldwide.”,
“Our premium travel services include personalized itinerary planning and 24/7 concierge support.”,
“Contoso’s travel insurance covers medical emergencies, trip cancellations, and lost baggage.”,
“Popular destinations include the Maldives, Swiss Alps, and African safaris.”,
“Contoso Travel provides exclusive access to boutique hotels and private guided tours.”
]

Add documents with metadata

collection.add(
documents=documents,
ids=[f"doc_{i}" for i in range(len(documents))],
metadatas=[{“source”: “training”, “type”: “explanation”} for _ in documents]
)

上下文提供器实现

ContextProvider类负责从多个来源获取并整合上下文信息：

1. 向量数据库检索：使用ChromaDB对旅行文档进行语义搜索
2. 天气信息：维护一个包含主要城市天气数据的模拟数据库
3. 统一上下文：将文档数据和天气信息整合为全面的上下文

核心方法：

• get_retrieval_context()：根据查询检索相关文档
• get_weather_data()：提供指定地点的天气信息
• get_unified_context()：整合文档和天气上下文以生成增强型响应

（技术说明：该实现采用了混合数据源策略，通过语义检索和结构化数据查询相结合的方式提升上下文相关性）

class ContextProvider:def __init__(self, collection):self.collection = collection# Simulated weather databaseself.weather_database = {"new york": {"temperature": 72, "condition": "Partly Cloudy", "humidity": 65, "wind": "10 mph"},"london": {"temperature": 60, "condition": "Rainy", "humidity": 80, "wind": "15 mph"},"tokyo": {"temperature": 75, "condition": "Sunny", "humidity": 50, "wind": "5 mph"},"sydney": {"temperature": 80, "condition": "Clear", "humidity": 45, "wind": "12 mph"},"paris": {"temperature": 68, "condition": "Cloudy", "humidity": 70, "wind": "8 mph"},}def get_retrieval_context(self, query: str) -> str:"""Retrieves relevant documents from vector database based on query."""results = self.collection.query(query_texts=[query],include=["documents", "metadatas"],n_results=2)context_strings = []if results and results.get("documents") and len(results["documents"][0]) > 0:for doc, meta in zip(results["documents"][0], results["metadatas"][0]):context_strings.append(f"Document: {doc}\nMetadata: {meta}")return "\n\n".join(context_strings) if context_strings else "No relevant documents found"def get_weather_data(self, location: str) -> str:"""Simulates retrieving weather data for a given location."""if not location:return ""location_key = location.lower()if location_key in self.weather_database:data = self.weather_database[location_key]return f"Weather for {location.title()}:\n" \f"Temperature: {data['temperature']}°F\n" \f"Condition: {data['condition']}\n" \f"Humidity: {data['humidity']}%\n" \f"Wind: {data['wind']}"else:return f"No weather data available for {location}."def get_unified_context(self, query: str, location: str = None) -> str:"""Returns a unified context combining both document retrieval and weather data."""retrieval_context = self.get_retrieval_context(query)weather_context = ""if location:weather_context = self.get_weather_data(location)weather_intro = f"\nWeather Information for {location}:\n"else:weather_intro = ""return f"Retrieved Context:\n{retrieval_context}\n\n{weather_intro}{weather_context}"

agent configuration

Create agents with enhanced capabilities

assistant = AssistantAgent(
name=“assistant”,
model_client=client,
system_message=(
"You are a helpful AI assistant that provides answers using ONLY the provided context. "
“Do NOT include any external information. Base your answer entirely on the context given below.”
),
)

Query Processing with RAG

async def ask_rag_agent(query: str, context_provider: ContextProvider, location: str = None):"""Sends a query to the assistant agent with context from the provider.Args:query: The user's questioncontext_provider: The context provider instancelocation: Optional location for weather queries"""try:# Get unified contextcontext = context_provider.get_unified_context(query, location)# Augment the query with contextaugmented_query = (f"{context}\n\n"f"User Query: {query}\n\n""Based ONLY on the above context, please provide a helpful answer.")# Send the augmented query to the assistantstart_time = time.time()response = await assistant.on_messages([TextMessage(content=augmented_query, source="user")],cancellation_token=CancellationToken(),)processing_time = time.time() - start_timereturn {'query': query,'response': response.chat_message.content,'processing_time': processing_time,'location': location}except Exception as e:print(f"Error processing query: {e}")return None

example

async def main():# Initialize context providercontext_provider = ContextProvider(collection)# Example queriesqueries = [{"query": "What does Contoso's travel insurance cover?"},{"query": "What's the weather like in London?", "location": "london"},{"query": "What luxury destinations does Contoso offer and what's the weather in Paris?", "location": "paris"},]print("=== Autogen RAG Demo ===")for query_data in queries:query = query_data["query"]location = query_data.get("location")print(f"\n\nQuery: {query}")if location:print(f"Location: {location}")# Show the context being usedcontext = context_provider.get_unified_context(query, location)print("\n--- Context Used ---")print(context)print("-------------------")# Get response from the agentresult = await ask_rag_agent(query, context_provider, location)if result:print(f"\nResponse: {result['response']}")print("\n" + "="*50)

if __name__ == "__main__":if asyncio.get_event_loop().is_running():await main()else:asyncio.run(main())

=== Autogen RAG Demo ===

Query: What does Contoso’s travel insurance cover?

— Context Used —
Retrieved Context:
Document: Contoso’s travel insurance covers medical emergencies, trip cancellations, and lost baggage.
Metadata: {‘source’: ‘training’, ‘type’: ‘explanation’}

Document: Contoso Travel offers luxury vacation packages to exotic destinations worldwide.
Metadata: {‘source’: ‘training’, ‘type’: ‘explanation’}

---

Response: Contoso’s travel insurance covers medical emergencies, trip cancellations, and lost baggage.

==================================================

Query: What’s the weather like in London?
Location: london

— Context Used —
…

Response: Contoso Travel offers luxury vacation packages to exotic destinations worldwide and provides exclusive access to boutique hotels and private guided tours. The weather in Paris is cloudy with a temperature of 68°F, 70% humidity, and winds at 8 mph.

==================================================