Practical Application
5 minutes min de lecture
Kimi K2 Technical Team
Kimi-K2 en Action : Développement d'Agents et Exploration de Scénarios d'Application
Kimi-K2 en Action : Développement d'Agents et Exploration de Scénarios d'Application
Introduction
Avec le développement rapide de la technologie de l'intelligence artificielle, les agents sont devenus une direction importante pour les applications IA. Kimi-K2, avec son architecture MoE à un trillion de paramètres et son optimisation spécialisée pour les agents, offre aux développeurs une base puissante pour créer des applications d'agents efficaces. Cet article démontrera comment tirer parti des capacités fondamentales de Kimi-K2 pour développer des applications d'agents pratiques à travers des études de cas réelles.
Avantages des Agents Kimi-K2
1. Capacités d'Appel d'Outils Puissantes
Kimi-K2 a été spécifiquement optimisé pour la fonctionnalité d'appel d'outils lors de sa conception, lui permettant de comprendre des descriptions d'outils complexes et de faire des appels précis :
import json
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
# Définir les fonctions d'outils
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Obtenir des informations météorologiques pour une ville spécifiée",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "Nom de la ville"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "Effectuer des calculs mathématiques",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Expression mathématique"
}
},
"required": ["expression"]
}
}
}
]
# Exemple de conversation d'agent
def chat_with_tools(model, tokenizer, user_input, tools):
messages = [
{"role": "system", "content": "Vous êtes un assistant intelligent capable d'appeler des outils pour aider les utilisateurs à résoudre des problèmes."},
{"role": "user", "content": user_input}
]
# Ajouter des descriptions d'outils
tool_prompt = f"Outils disponibles : {json.dumps(tools, ensure_ascii=False, indent=2)}"
messages[0]["content"] += f"\n\n{tool_prompt}"
# Générer une réponse
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer(text, return_tensors="pt")
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=512,
temperature=0.7,
do_sample=True
)
response = tokenizer.decode(outputs[0][inputs.input_ids.shape[-1]:], skip_special_tokens=True)
return response
# Exemple d'utilisation
user_query = "Quel temps fait-il à Pékin aujourd'hui ? Si la probabilité de pluie dépasse 70 %, aide-moi à calculer combien le tarif de taxi serait plus cher (normalement 15 yuans, augmentation de prix de 30 % les jours de pluie)"
response = chat_with_tools(model, tokenizer, user_query, tools)
2. Mémoire de Contexte Ultra-Longue
La longueur de contexte de 128K permet à Kimi-K2 de maintenir un historique de conversation à long terme :
class LongContextAgent:
def __init__(self, model, tokenizer, max_context_length=128000):
self.model = model
self.tokenizer = tokenizer
self.conversation_history = []
self.max_context_length = max_context_length
def add_message(self, role, content):
self.conversation_history.append({"role": role, "content": content})
self._trim_context()
def _trim_context(self):
# Rester dans la limite de longueur de contexte
total_tokens = 0
trimmed_history = []
for message in reversed(self.conversation_history):
message_tokens = len(self.tokenizer.encode(message["content"]))
if total_tokens + message_tokens > self.max_context_length:
break
trimmed_history.insert(0, message)
total_tokens += message_tokens
self.conversation_history = trimmed_history
def generate_response(self, user_input):
self.add_message("user", user_input)
# Construire l'historique complet de la conversation
text = self.tokenizer.apply_chat_template(
self.conversation_history,
tokenize=False,
add_generation_prompt=True
)
inputs = self.tokenizer(text, return_tensors="pt")
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=512,
temperature=0.6,
do_sample=True
)
response = self.tokenizer.decode(
outputs[0][inputs.input_ids.shape[-1]:],
skip_special_tokens=True
)
self.add_message("assistant", response)
return response
3. Avantages de la Collaboration Multi-Expert
L'architecture MoE permet à différents types de tâches d'invoquer les experts les plus adaptés :
class MultiExpertAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.task_types = {
"coding": "Programmation et code",
"math": "Calculs mathématiques et raisonnement",
"writing": "Rédaction et édition de texte",
"analysis": "Analyse de données et résumé"
}
def classify_task(self, user_input):
"""Logique simple de classification des tâches"""
if any(keyword in user_input.lower() for keyword in ["code", "programmation", "programme", "algorithme"]):
return "coding"
elif any(keyword in user_input.lower() for keyword in ["calculer", "math", "formule", "raisonnement"]):
return "math"
elif any(keyword in user_input.lower() for keyword in ["écrire", "article", "résumé", "rapport"]):
return "writing"
elif any(keyword in user_input.lower() for keyword in ["analyser", "statistiques", "données", "graphique"]):
return "analysis"
else:
return "general"
def generate_specialized_response(self, user_input, task_type):
system_prompts = {
"coding": "Vous êtes un assistant de programmation professionnel, compétent dans plusieurs langages de programmation et algorithmes.",
"math": "Vous êtes un expert en mathématiques, capable de résoudre des problèmes mathématiques complexes et de raisonnement logique.",
"writing": "Vous êtes un assistant de rédaction professionnel, capable de créer et d'éditer divers types de textes.",
"analysis": "Vous êtes un expert en analyse de données, capable d'extraire des informations et des tendances à partir des données.",
"general": "Vous êtes un assistant IA polyvalent, capable de traiter divers types de problèmes."
}
messages = [
{"role": "system", "content": system_prompts.get(task_type, system_prompts["general"])},
{"role": "user", "content": user_input}
]
# Logique de génération de réponse...
return self._generate_response(messages)
Cas d'Application Réels
Cas 1 : Assistant Intelligent de Service Client
class CustomerServiceAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.knowledge_base = {
"refund_policy": "Supporte les retours sans raison pendant 7 jours, les produits doivent rester en bon état...",
"shipping_info": "Généralement livré dans un délai de 1 à 3 jours ouvrables, livraison le lendemain disponible...",
"product_warranty": "Les produits électroniques sont garantis 1 an, extensible à 3 ans..."
}
self.conversation_state = {}
def handle_query(self, user_id, query):
# Récupérer les informations pertinentes
relevant_info = self.search_knowledge(query)
# Construire le contexte
context = f"Informations pertinentes : {relevant_info}\nQuestion de l'utilisateur : {query}"
messages = [
{"role": "system", "content": "Vous êtes un assistant de service client professionnel, fournissant des réponses polies et précises aux questions des utilisateurs."},
{"role": "user", "content": context}
]
response = self._generate_response(messages)
# Mettre à jour l'état de la conversation
self.conversation_state[user_id] = {
"last_query": query,
"last_response": response,
"context": relevant_info
}
return response
def search_knowledge(self, query):
# Logique simple de récupération de connaissances
for key, value in self.knowledge_base.items():
if any(keyword in query for keyword in key.split()):
return value
return "Aucune information pertinente trouvée, veuillez contacter le service client humain."
Cas 2 : Assistant de Revue de Code
class CodeReviewAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.review_criteria = [
"Correction de la logique du code",
"Suggestions d'optimisation des performances",
"Vérifications de sécurité",
"Style de code et normes",
"Mécanismes de gestion des erreurs"
]
def review_code(self, code, language="python"):
# Construire le texte des critères de révision
criteria_text = "\n".join([f"- {criterion}" for criterion in self.review_criteria])
# Créer l'invite de révision
review_prompt = f"Veuillez effectuer une révision complète du code {language}"
messages = [
{"role": "system", "content": "Vous êtes un expert senior en révision de code, capable d'identifier les problèmes de code et de fournir des suggestions professionnelles."},
{"role": "user", "content": review_prompt}
]
return self._generate_response(messages)
def suggest_improvements(self, code, issues):
# Créer l'invite d'amélioration
improvement_prompt = "Sur la base des problèmes de révision du code, veuillez fournir un code amélioré"
messages = [
{"role": "system", "content": "Veuillez fournir un code amélioré et expliquer les raisons des modifications."},
{"role": "user", "content": improvement_prompt}
]
return self._generate_response(messages)
Cas 3 : Assistant de Tutorat Éducatif
class EducationAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.student_progress = {}
def adaptive_tutoring(self, student_id, subject, question, difficulty="medium"):
# Obtenir les performances historiques de l'étudiant
progress = self.student_progress.get(student_id, {"correct": 0, "total": 0})
success_rate = progress["correct"] / max(progress["total"], 1)
# Ajuster la stratégie d'enseignement en fonction du taux de réussite
if success_rate > 0.8:
teaching_style = "Peut essayer un contenu de difficulté plus élevée, fournir des problèmes stimulants"
elif success_rate > 0.6:
teaching_style = "Maintenir la difficulté actuelle, fournir des explications détaillées"
else:
teaching_style = "Besoin d'explications plus basiques, guidance étape par étape"
prompt = f"""
Question de l'étudiant : {question}
Sujet : {subject}
Niveau de difficulté : {difficulty}
Stratégie d'enseignement : {teaching_style}
Taux de réussite de l'étudiant : {success_rate:.2%}
Veuillez agir en tant qu'enseignant professionnel pour répondre à la question, en utilisant des méthodes d'enseignement appropriées.
"""
messages = [
{"role": "system", "content": "Vous êtes un enseignant expérimenté, bon pour adapter l'éducation aux besoins individuels et ajuster les méthodes d'enseignement en fonction des niveaux des étudiants."},
{"role": "user", "content": prompt}
]
response = self._generate_response(messages)
return response
def generate_practice_problems(self, subject, topic, difficulty, count=3):
# Créer l'invite de problèmes pratiques
prompt = f"Veuillez générer des problèmes pratiques sur {topic} en {subject}"
messages = [
{"role": "system", "content": "Vous êtes un concepteur de problèmes professionnel, capable de créer des problèmes pratiques de haute qualité."},
{"role": "user", "content": prompt}
]
return self._generate_response(messages)
Conseils d'Optimisation des Performances
1. Stratégie de Mise en Cache Intelligente
import hashlib
import pickle
from functools import lru_cache
class CachedAgent:
def __init__(self, model, tokenizer, cache_size=1000):
self.model = model
self.tokenizer = tokenizer
self.response_cache = {}
self.cache_size = cache_size
def _hash_input(self, messages):
# Générer un hash pour l'entrée
content = str(messages)
return hashlib.md5(content.encode()).hexdigest()
def generate_with_cache(self, messages):
cache_key = self._hash_input(messages)
if cache_key in self.response_cache:
return self.response_cache[cache_key]
response = self._generate_response(messages)
# Gestion du cache
if len(self.response_cache) >= self.cache_size:
# Supprimer l'entrée de cache la plus ancienne
oldest_key = next(iter(self.response_cache))
del self.response_cache[oldest_key]
self.response_cache[cache_key] = response
return response
2. Traitement Asynchrone
import asyncio
import aiohttp
class AsyncAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.session = None
async def process_multiple_queries(self, queries):
tasks = []
for query in queries:
task = asyncio.create_task(self.process_single_query(query))
tasks.append(task)
results = await asyncio.gather(*tasks)
return results
async def process_single_query(self, query):
# Simuler un traitement asynchrone
await asyncio.sleep(0.1) # Éviter le blocage
messages = [
{"role": "user", "content": query}
]
return self._generate_response(messages)
async def external_api_call(self, url, data):
if not self.session:
self.session = aiohttp.ClientSession()
async with self.session.post(url, json=data) as response:
return await response.json()
Recommandations de Meilleures Pratiques
1. Gestion de l'État de Conversation
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional
class ConversationState(Enum):
GREETING = "greeting"
COLLECTING_INFO = "collecting_info"
PROCESSING = "processing"
CLARIFYING = "clarifying"
COMPLETED = "completed"
@dataclass
class UserContext:
user_id: str
state: ConversationState
collected_info: Dict
preferences: Dict
history: List[Dict]
class StatefulAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.user_contexts = {}
def get_or_create_context(self, user_id):
if user_id not in self.user_contexts:
self.user_contexts[user_id] = UserContext(
user_id=user_id,
state=ConversationState.GREETING,
collected_info={},
preferences={},
history=[]
)
return self.user_contexts[user_id]
def handle_message(self, user_id, message):
context = self.get_or_create_context(user_id)
# Gérer le message en fonction de l'état actuel
if context.state == ConversationState.GREETING:
return self.handle_greeting(context, message)
elif context.state == ConversationState.COLLECTING_INFO:
return self.handle_info_collection(context, message)
# Autre gestion d'état...
return self.generate_default_response(context, message)
2. Gestion des Erreurs et Dégradation Gracieuse
class RobustAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.fallback_responses = {
"generation_failed": "Désolé, je ne peux pas générer de réponse pour le moment, veuillez réessayer plus tard.",
"context_too_long": "L'historique de conversation est trop long, commençons à nouveau.",
"tool_call_failed": "L'appel d'outil a échoué, je vais répondre en utilisant des méthodes alternatives."
}
def safe_generate(self, messages, max_retries=3):
for attempt in range(max_retries):
try:
return self._generate_response(messages)
except torch.cuda.OutOfMemoryError:
torch.cuda.empty_cache()
# Réduire la longueur d'entrée
messages = self._trim_messages(messages)
except Exception as e:
if attempt == max_retries - 1:
return self.fallback_responses["generation_failed"]
continue
return self.fallback_responses["generation_failed"]
def _trim_messages(self, messages, max_length=4096):
# Garder les messages système et les messages utilisateurs récents
system_msgs = [msg for msg in messages if msg["role"] == "system"]
user_msgs = [msg for msg in messages if msg["role"] == "user"]
if user_msgs:
return system_msgs + [user_msgs[-1]]
return system_msgs
Déploiement et Surveillance
1. Surveillance des Performances
import time
import logging
from dataclasses import dataclass
from typing import Dict
@dataclass
class AgentMetrics:
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
average_response_time: float = 0.0
peak_memory_usage: float = 0.0
class MonitoredAgent:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.metrics = AgentMetrics()
self.logger = logging.getLogger(__name__)
def generate_with_monitoring(self, messages):
start_time = time.time()
self.metrics.total_requests += 1
try:
response = self._generate_response(messages)
self.metrics.successful_requests += 1
# Mettre à jour le temps de réponse moyen
elapsed = time.time() - start_time
self.metrics.average_response_time = (
(self.metrics.average_response_time * (self.metrics.successful_requests - 1) + elapsed)
/ self.metrics.successful_requests
)
self.logger.info(f"Demande complétée en {elapsed:.2f}s")
return response
except Exception as e:
self.metrics.failed_requests += 1
self.logger.error(f"Demande échouée : {str(e)}")
raise
def get_metrics_summary(self):
success_rate = (
self.metrics.successful_requests / max(self.metrics.total_requests, 1) * 100
)
return {
"total_requests": self.metrics.total_requests,
"success_rate": f"{success_rate:.2f}%",
"average_response_time": f"{self.metrics.average_response_time:.2f}s",
"failure_count": self.metrics.failed_requests
}
Conclusion
Kimi-K2 fournit une base technique puissante pour le développement d'agents. Ses capacités de spécialisation des experts de l'architecture MoE, sa mémoire de contexte ultra-longue et sa fonctionnalité d'appel d'outils excellente permettent aux développeurs de créer des applications hautement intelligentes et pratiques.
À travers les cas et les meilleures pratiques de cet article, les développeurs peuvent :
- Utiliser les capacités d'appel d'outils pour construire des agents riches en fonctionnalités
- Mettre en œuvre un dialogue continu grâce à une mémoire de long contexte
- Gérer des tâches complexes en utilisant la collaboration multi-expert
- Adopter des meilleures pratiques pour garantir la stabilité et la fiabilité du système
Alors que la technologie continue de se développer, Kimi-K2 continuera de stimuler l'innovation dans les applications d'agents, apportant plus de possibilités à divers secteurs.