Awesome LLM Apps: Kho tàng ứng dụng AI thực tế cho developers#

1
# Example: AI Travel Agent
2
from phi.agent import Agent
3
from phi.model.openai import OpenAIChat
4

5
travel_agent = Agent(
6
    model=OpenAIChat(id="gpt-4"),
7
    tools=[weather_tool, booking_tool, maps_tool],
8
    instructions="You are a travel planning expert..."
9
)
10

11
response = travel_agent.run("Plan a 5-day trip to Tokyo")

1
# Example: AI Investment Agent
2
class InvestmentAgent:
3
    def __init__(self):
4
        self.market_analyzer = MarketAnalyzer()
5
        self.risk_assessor = RiskAssessor()
6
        self.portfolio_optimizer = PortfolioOptimizer()
7

8
    def analyze_investment(self, ticker):
9
        # Comprehensive investment analysis
10
        market_data = self.market_analyzer.get_data(ticker)
11
        risk_profile = self.risk_assessor.assess(ticker)
12
        recommendation = self.generate_recommendation(market_data, risk_profile)
13
        return recommendation

1
# Example: AI Finance Agent Team
2
from phi.agent import Agent
3
from phi.tools.yfinance import YFinanceTools
4

5
# Specialized agents
6
market_analyst = Agent(
7
    name="Market Analyst",
8
    model=OpenAIChat(id="gpt-4"),
9
    tools=[YFinanceTools()],
10
    instructions="Analyze market trends and data"
11
)
12

13
risk_manager = Agent(
14
    name="Risk Manager",
15
    model=OpenAIChat(id="gpt-4"),
16
    instructions="Assess investment risks"
17
)
18

19
portfolio_manager = Agent(
20
    name="Portfolio Manager",
21
    model=OpenAIChat(id="gpt-4"),
22
    instructions="Optimize portfolio allocation"
23
)
24

25
# Coordinated team workflow
26
team = [market_analyst, risk_manager, portfolio_manager]

1
# Example: Voice RAG Agent
2
import speech_recognition as sr
3
import pyttsx3
4

5
class VoiceRAGAgent:
6
    def __init__(self):
7
        self.recognizer = sr.Recognizer()
8
        self.tts_engine = pyttsx3.init()
9
        self.rag_system = RAGSystem()
10

11
    def listen_and_respond(self):
12
        with sr.Microphone() as source:
13
            audio = self.recognizer.listen(source)
14
            query = self.recognizer.recognize_google(audio)
15

16
        response = self.rag_system.query(query)
17
        self.tts_engine.say(response)
18
        self.tts_engine.runAndWait()

1
# Advanced RAG Implementation
2
from langchain.vectorstores import Chroma
3
from langchain.embeddings import OpenAIEmbeddings
4
from langchain.llms import OpenAI
5

6
class AdvancedRAG:
7
    def __init__(self):
8
        self.embeddings = OpenAIEmbeddings()
9
        self.vectorstore = Chroma(embedding_function=self.embeddings)
10
        self.llm = OpenAI()
11
        self.retriever = self.vectorstore.as_retriever(
12
            search_type="mmr",  # Maximum Marginal Relevance
13
            search_kwargs={"k": 6, "fetch_k": 20}
14
        )
15

16
    def query_with_sources(self, question):
17
        # Retrieve relevant documents
18
        docs = self.retriever.get_relevant_documents(question)
19

20
        # Generate response với context
21
        context = "\n".join([doc.page_content for doc in docs])
22
        response = self.llm(f"Context: {context}\n\nQuestion: {question}")
23

24
        return {
25
            "answer": response,
26
            "sources": [doc.metadata for doc in docs]
27
        }

1
# LLM App với Personalized Memory
2
from phi.memory import AssistantMemory
3
from phi.storage.assistant.postgres import PgAssistantStorage
4

5
class PersonalizedAgent:
6
    def __init__(self, user_id):
7
        self.memory = AssistantMemory(
8
            storage=PgAssistantStorage(
9
                table_name="agent_memory",
10
                db_url="postgresql://user:pass@localhost/db"
11
            ),
12
            create_user_memories=True,
13
            create_session_summary=True
14
        )
15
        self.user_id = user_id
16

17
    def chat(self, message):
18
        # Retrieve user context
19
        user_context = self.memory.get_user_memories(self.user_id)
20

21
        # Generate personalized response
22
        response = self.generate_response(message, user_context)
23

24
        # Update memory
25
        self.memory.add_chat_message(self.user_id, message, response)
26

27
        return response

1
# MCP Agent Example
2
from mcp import MCPAgent, MCPServer
3

4
class GitHubMCPAgent(MCPAgent):
5
    def __init__(self):
6
        super().__init__()
7
        self.github_server = MCPServer("github")
8
        self.tools = [
9
            "create_issue", "list_repos",
10
            "get_commits", "create_pr"
11
        ]
12

13
    async def handle_request(self, request):
14
        if request.method == "github/create_issue":
15
            return await self.create_github_issue(request.params)
16
        elif request.method == "github/list_repos":
17
            return await self.list_repositories(request.params)

1
# AI Chess Agent
2
import chess
3
import chess.engine
4

5
class AIChessAgent:
6
    def __init__(self):
7
        self.board = chess.Board()
8
        self.engine = chess.engine.SimpleEngine.popen_uci("stockfish")
9

10
    def make_move(self):
11
        # AI decision making
12
        result = self.engine.play(self.board, chess.engine.Limit(time=2.0))
13
        move = result.move
14

15
        self.board.push(move)
16
        return move
17

18
    def evaluate_position(self):
19
        # Position evaluation using neural networks
20
        info = self.engine.analyse(self.board, chess.engine.Limit(depth=20))
21
        return info["score"].white().score()

1
# Chat với PDF Application
2
from langchain.document_loaders import PyPDFLoader
3
from langchain.text_splitter import RecursiveCharacterTextSplitter
4

5
class PDFChatAgent:
6
    def __init__(self, pdf_path):
7
        # Load và process PDF
8
        loader = PyPDFLoader(pdf_path)
9
        documents = loader.load()
10

11
        # Split into chunks
12
        text_splitter = RecursiveCharacterTextSplitter(
13
            chunk_size=1000,
14
            chunk_overlap=200
15
        )
16
        texts = text_splitter.split_documents(documents)
17

18
        # Create vector store
19
        self.vectorstore = Chroma.from_documents(texts, OpenAIEmbeddings())
20
        self.qa_chain = RetrievalQA.from_chain_type(
21
            llm=OpenAI(),
22
            chain_type="stuff",
23
            retriever=self.vectorstore.as_retriever()
24
        )
25

26
    def ask_question(self, question):
27
        return self.qa_chain.run(question)

1
# Llama 3.2 Fine-tuning Example
2
from transformers import (
3
    AutoTokenizer, AutoModelForCausalLM,
4
    TrainingArguments, Trainer
5
)
6
from datasets import Dataset
7

8
class LlamaFineTuner:
9
    def __init__(self, model_name="meta-llama/Llama-3.2-1B"):
10
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
11
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
12

13
    def prepare_dataset(self, texts, labels):
14
        def tokenize_function(examples):
15
            return self.tokenizer(
16
                examples['text'],
17
                truncation=True,
18
                padding=True,
19
                max_length=512
20
            )
21

22
        dataset = Dataset.from_dict({'text': texts, 'labels': labels})
23
        tokenized_dataset = dataset.map(tokenize_function, batched=True)
24
        return tokenized_dataset
25

26
    def fine_tune(self, train_dataset, eval_dataset):
27
        training_args = TrainingArguments(
28
            output_dir="./llama-finetuned",
29
            num_train_epochs=3,
30
            per_device_train_batch_size=4,
31
            per_device_eval_batch_size=4,
32
            warmup_steps=500,
33
            weight_decay=0.01,
34
            logging_dir="./logs",
35
        )
36

37
        trainer = Trainer(
38
            model=self.model,
39
            args=training_args,
40
            train_dataset=train_dataset,
41
            eval_dataset=eval_dataset,
42
        )
43

44
        trainer.train()
45
        trainer.save_model()

1
# Google ADK Crash Course Example
2
from google.adk import Agent, Tool
3

4
def weather_tool(location: str) -> str:
5
    # Weather API integration
6
    return f"Weather in {location}: Sunny, 25°C"
7

8
agent = Agent(
9
    name="WeatherAgent",
10
    instructions="Provide weather information for any location",
11
    tools=[weather_tool],
12
    model="gemini-pro"
13
)
14

15
response = agent.run("What's the weather like in Tokyo?")

1
# OpenAI SDK Crash Course
2
from openai import OpenAI
3

4
class OpenAIAgent:
5
    def __init__(self):
6
        self.client = OpenAI()
7
        self.tools = [
8
            {
9
                "type": "function",
10
                "function": {
11
                    "name": "get_weather",
12
                    "description": "Get current weather",
13
                    "parameters": {
14
                        "type": "object",
15
                        "properties": {
16
                            "location": {"type": "string"}
17
                        }
18
                    }
19
                }
20
            }
21
        ]
22

23
    def chat_completion(self, message):
24
        return self.client.chat.completions.create(
25
            model="gpt-4",
26
            messages=[{"role": "user", "content": message}],
27
            tools=self.tools
28
        )

1
# Multi-Agent Load Balancing
2
from concurrent.futures import ThreadPoolExecutor
3
import queue
4

5
class AgentPool:
6
    def __init__(self, agent_class, pool_size=10):
7
        self.agents = [agent_class() for _ in range(pool_size)]
8
        self.queue = queue.Queue()
9

10
        # Initialize agent pool
11
        for agent in self.agents:
12
            self.queue.put(agent)
13

14
    def process_request(self, request):
15
        agent = self.queue.get()
16
        try:
17
            result = agent.process(request)
18
            return result
19
        finally:
20
            self.queue.put(agent)
21

22
    def batch_process(self, requests):
23
        with ThreadPoolExecutor(max_workers=len(self.agents)) as executor:
24
            futures = [
25
                executor.submit(self.process_request, req)
26
                for req in requests
27
            ]
28
            return [f.result() for f in futures]

1
# Agent Performance Monitoring
2
import logging
3
import time
4
from functools import wraps
5

6
def monitor_agent_performance(func):
7
    @wraps(func)
8
    def wrapper(self, *args, **kwargs):
9
        start_time = time.time()
10

11
        try:
12
            result = func(self, *args, **kwargs)
13
            execution_time = time.time() - start_time
14

15
            logging.info(f"Agent {self.__class__.__name__} executed in {execution_time:.2f}s")
16
            return result
17

18
        except Exception as e:
19
            execution_time = time.time() - start_time
20
            logging.error(f"Agent {self.__class__.__name__} failed after {execution_time:.2f}s: {str(e)}")
21
            raise
22

23
    return wrapper

1
# Single Responsibility Agent
2
class SpecializedAgent:
3
    def __init__(self, domain):
4
        self.domain = domain
5
        self.tools = self.load_domain_tools()
6
        self.model = self.configure_model()
7

8
    def process(self, request):
9
        # Single, well-defined responsibility
10
        return self.execute_specialized_task(request)
11

12
# Coordination Pattern
13
class CoordinatorAgent:
14
    def __init__(self, specialist_agents):
15
        self.specialists = specialist_agents
16
        self.router = RequestRouter()
17

18
    def delegate(self, request):
19
        specialist = self.router.route(request)
20
        return specialist.process(request)

1
# Robust Agent Implementation
2
from tenacity import retry, stop_after_attempt, wait_exponential
3

4
class ResilientAgent:
5
    @retry(
6
        stop=stop_after_attempt(3),
7
        wait=wait_exponential(multiplier=1, min=4, max=10)
8
    )
9
    def robust_process(self, request):
10
        try:
11
            return self.model.generate(request)
12
        except RateLimitError:
13
            # Handle rate limiting
14
            time.sleep(60)
15
            raise
16
        except APIError as e:
17
            # Handle API errors
18
            self.logger.error(f"API Error: {e}")
19
            raise

1
# Future Agent Architecture
2
class NextGenAgent:
3
    def __init__(self):
4
        self.multimodal_processor = MultiModalProcessor()
5
        self.continuous_learner = ContinuousLearner()
6
        self.federated_network = FederatedNetwork()
7

8
    async def process_multimodal(self, inputs):
9
        # Handle text, image, audio simultaneously
10
        processed_inputs = await self.multimodal_processor.process(inputs)
11

12
        # Learn from interaction
13
        self.continuous_learner.update(processed_inputs)
14

15
        # Coordinate với network
16
        network_insights = await self.federated_network.query(processed_inputs)
17

18
        return self.synthesize_response(processed_inputs, network_insights)