LlamaIndex Agents

# LlamaIndex OpenAI Integration
from llama_index.llms.openai import OpenAI
from llama_index.core.tools import FunctionTool
from llama_index.core import VectorStoreIndex
import openai

# Configure LLM with OpenAI API key
llm = OpenAI(
    model="gpt-4",
    api_key="your_openai_api_key",
    temperature=0.1,
    max_tokens=1000
)

# Setup for function tool integration
def multiply_two_numbers(a: float, b: float) -> float:
    """Python functions for mathematical operations"""
    return a * b

def def_multiply(numbers: list) -> float:
    """Multiply multiple numbers using def multiply pattern"""
    result = 1
    for num in numbers:
        result *= num
    return result

# Create function tool from python functions
multiply_tool = FunctionTool.from_defaults(fn=multiply_two_numbers)
multi_tool = FunctionTool.from_defaults(fn=def_multiply)

# Tool calling with LlamaIndex
from llama_index.core.agent import ReActAgent

agent = ReActAgent.from_tools(
    [multiply_tool, multi_tool],
    llm=llm,
    verbose=True
)

# Execute with openai integration
response = agent.chat("Calculate 15 * 7 and then multiply result by 3")
print(f"LLM OpenAI response: {response}")

# Vector Database Integration for RAG System
from llama_index.vector_stores import ChromaVectorStore
from llama_index import VectorStoreIndex, ServiceContext
from llama_index.core.tools import QueryEngineTool
import chromadb

class VectorRAGSystem:
    def __init__(self):
        # Initialize Chroma for rag system
        self.chroma_client = chromadb.PersistentClient(path="./vector_db")
        self.collection = self.chroma_client.get_or_create_collection("documents")
        
        # Setup vector store for framework for building
        self.vector_store = ChromaVectorStore(chroma_collection=self.collection)
        
    def create_query_engine(self, documents: list):
        """Build query engine for document retrieval"""
        
        # Create index for rag system
        index = VectorStoreIndex.from_documents(
            documents,
            vector_store=self.vector_store
        )
        
        # Setup query engine with semantic search
        query_engine = index.as_query_engine(
            similarity_top_k=5,
            response_mode="tree_summarize"
        )
        
        return query_engine
    
    def setup_multi_index_system(self, document_sets: dict):
        """Build multi agent systems with different indexes"""
        
        engines = {}
        for name, docs in document_sets.items():
            # Create specialized query engine for each domain
            collection = self.chroma_client.get_or_create_collection(f"{name}_docs")
            vector_store = ChromaVectorStore(chroma_collection=collection)
            
            index = VectorStoreIndex.from_documents(docs, vector_store=vector_store)
            engines[name] = index.as_query_engine()
        
        return engines

# Usage example
rag_system = VectorRAGSystem()
query_engine = rag_system.create_query_engine(documents)

# Agentic Workflows Implementation
from llama_index.core.workflow import Workflow, StartEvent, StopEvent
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI
from llama_index.core.tools import FunctionTool

class AgenticWorkflowSystem(Workflow):
    """Build agentic systems with coordinated workflows"""
    
    def __init__(self):
        super().__init__()
        self.llm = OpenAI(model="gpt-4")
        
        # Setup agents for agentic workflows
        self.research_agent = self.create_research_agent()
        self.analysis_agent = self.create_analysis_agent()
        self.synthesis_agent = self.create_synthesis_agent()
    
    def create_research_agent(self):
        """Create agent for data gathering in agentic systems"""
        
        def research_data(topic: str) -> str:
            """Research function for data collection"""
            return f"Research data collected for: {topic}"
        
        research_tool = FunctionTool.from_defaults(fn=research_data)
        
        agent = ReActAgent.from_tools(
            [research_tool],
            llm=self.llm,
            system_prompt="You are a research agent in an agentic workflow"
        )
        
        return agent
    
    def create_analysis_agent(self):
        """Create analysis agent for agentic workflows"""
        
        def analyze_data(data: str) -> str:
            """Analysis function for data processing"""
            return f"Analysis completed for: {data}"
        
        analysis_tool = FunctionTool.from_defaults(fn=analyze_data)
        
        agent = ReActAgent.from_tools(
            [analysis_tool],
            llm=self.llm,
            system_prompt="You are an analysis agent in agentic systems"
        )
        
        return agent
    
    def create_synthesis_agent(self):
        """Create synthesis agent for final output"""
        
        def synthesize_results(research: str, analysis: str) -> str:
            """Synthesis function for combining results"""
            return f"Synthesis of {research} and {analysis}"
        
        synthesis_tool = FunctionTool.from_defaults(fn=synthesize_results)
        
        agent = ReActAgent.from_tools(
            [synthesis_tool],
            llm=self.llm,
            system_prompt="You are a synthesis agent completing agentic workflows"
        )
        
        return agent
    
    async def run_agentic_workflow(self, query: str):
        """Execute complete agentic systems workflow"""
        
        # Step 1: Research phase
        research_result = await self.research_agent.achat(
            f"Research information about: {query}"
        )
        
        # Step 2: Analysis phase  
        analysis_result = await self.analysis_agent.achat(
            f"Analyze the research: {research_result}"
        )
        
        # Step 3: Synthesis phase
        final_result = await self.synthesis_agent.achat(
            f"Synthesize research and analysis: {research_result}, {analysis_result}"
        )
        
        return {
            "research": research_result,
            "analysis": analysis_result,
            "synthesis": final_result,
            "workflow_status": "completed"
        }

# Execute agentic workflows
workflow = AgenticWorkflowSystem()
import asyncio
result = asyncio.run(workflow.run_agentic_workflow("Market analysis for AI tools"))