Magistral 추론을 활용한 HubSpot 동적 멀티 에이전트 시스템
Magistral 추론을 활용한 HubSpot 동적 멀티 에이전트 시스템 (HubSpot Dynamic Multi-Agent System with Magistral Reasoning)
이 쿡북에서는 Magistral 추론 모델의 강력함과 HubSpot CRM 통합을 결합해서, 복잡한 비즈니스 쿼리를 이해하고 정교한 CRM 작업을 자동으로 실행하는 지능형 멀티 에이전트 시스템을 만드는 방법을 보여줘요. 자연어 비즈니스 질문을 실행 가능한 통찰과 자동화된 CRM 업데이트로 바꿔서, 고급 AI 추론이 영업 운영과 전략적 의사결정을 어떻게 간소화하는지 보여주죠.
출처: 문서
본문
문제 정의 (Problem Statement)
전통적인 CRM의 과제 (Traditional CRM Challenges)
현대의 영업·마케팅 팀은 HubSpot 같은 CRM 시스템을 다룰 때 몇 가지 치명적인 과제에 직면해요.
- 수동 데이터 분석 (Manual Data Analysis): 팀은 딜, 연락처, 회사를 수동으로 분석해서 통찰을 뽑아내느라 많은 시간을 보내요.
- 복잡한 쿼리 처리 (Complex Query Processing): 비즈니스 이해관계자들은 여러 CRM 객체의 데이터를 필요로 하는 다면적인 질문에 대한 답을 얻는 데 어려움을 겪어요.
- 전략 기획 (Strategic Planning): 시장 분석과 확장 계획은 CRM 데이터와 비즈니스 인텔리전스를 결합해야 하는데, 이는 기본적으로 지원되지 않아요.
샘플 쿼리 (Sample Query)
"Assign priorities to all deals based on deal value"
이런 쿼리들은 다음을 요구해요.
- 비즈니스 맥락 이해하기
- 여러 데이터 소스 분석하기
- 비즈니스 로직 적용하기
- 실행 가능한 추천 생성하기
- 때로는 CRM 레코드를 자동으로 업데이트하기
솔루션 아키텍처 (Solution Architecture)
핵심 혁신: Magistral Reasoning + HubSpot Integration + Multi-Agent Orchestration
우리 솔루션은 Mistral의 Magistral 추론 모델과 HubSpot의 포괄적인 CRM API를 정교한 멀티 에이전트 시스템으로 결합해서 다음을 할 수 있어요.
- 이해 (Understand): Magistral의 고급 추론 능력으로 복잡한 비즈니스 쿼리를 이해해요.
- 계획 (Plan): 동적으로 생성된 전문 에이전트로 다단계 실행 전략을 계획해요.
- 실행 (Execute): 조정된 에이전트 워크플로우를 통해 데이터 분석과 CRM 업데이트를 모두 수행해요.
- 종합 (Synthesize): 결과를 전략적 추천과 함께 실행 가능한 비즈니스 통찰로 종합해요.
AgentOrchestrator
마스터 조정자로, 전체 멀티 에이전트 워크플로우와 HubSpot 통합을 관리해요. 쿼리 분석부터 하위 에이전트 실행, 최종 종합까지 전체 흐름을 조율하면서 에이전트 수명주기와 데이터 연결을 관리해요.
LeadAgent
Magistral 추론 모델이 구동하며 thinking 패턴 처리를 사용해요. 리드 에이전트는 정교한 쿼리 분석을 수행해서 비즈니스 의도를 이해하고, 데이터 요구사항을 결정하며, 어떤 하위 에이전트를 동적으로 생성할지 지정하는 상세한 실행 계획을 만들어요.
동적 하위 에이전트 (Dynamic Sub-Agents)
하위 에이전트는 미리 정의된 템플릿이 아니라 특정 쿼리 요구사항에 따라 즉석에서(on-the-fly) 생성돼요. 각 에이전트는 전문 역할(예: priority_calculator, market_analyzer, deals_updater), 특정 작업, 목표 지향적인 데이터 접근 패턴을 가진 채 동적으로 생성되며, 빠른 실행을 위해 Mistral Small을 사용해요.
HubSpot API 커넥터 (HubSpot API Connector)
CRM 데이터와 작업에 포괄적으로 접근할 수 있는 전용 커넥터예요.
- 속성 탐색 (Property Discovery): 사용 가능한 모든 HubSpot 필드와 유효한 값을 자동으로 매핑해요.
- 데이터 가져오기 (Data Fetching): 딜, 연락처, 회사를 전체 속성 세트와 함께 가져와요.
- 배치 업데이트 (Batch Updates): 100개 단위의 배치로 여러 레코드를 효율적으로 업데이트해요.
SynthesisAgent
최종 오케스트레이터로, Mistral Small을 사용해 모든 하위 에이전트 결과를 일관되고 실행 가능한 비즈니스 통찰로 결합해요. 기술적인 에이전트 출력을 전략적 추천과 다음 단계가 포함된 사용자 친화적 응답으로 변환해요.
설치 (Installation)
데모를 위해 hubspot-api-client와 mistralai 패키지가 필요해요.
Python
!pip install hubspot-api-client=="12.0.0" mistralai=="1.9.3"
임포트 (Imports)
import requests
import json
from mistralai.client import Mistral, ThinkChunk, TextChunk
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import re
API 키 설정 (Setup API Keys)
HUBSPOT_API_KEY = "<YOUR HUBSPOT API KEY>" # Replace with your HubSpot API key
MISTRAL_API_KEY = "<YOUR MISTRAL API KEY>" # Get it from https://console.mistral.ai/api-keys
MistralAI 클라이언트 설정 (Setup MistralAI Client)
mistral_client = Mistral(api_key=MISTRAL_API_KEY)
HubSpot API 커넥터 (HubSpot API connector)
get_data: HubSpot API에서 CRM 데이터를 가져와요. 분석을 위해 딜, 연락처, 회사 데이터를 가져옵니다.batch_update: HubSpot 레코드에 배치 업데이트를 수행해요. 100개 레코드의 효율적인 배치로 업데이트를 작성합니다.get_properties: 유효한 값과 드롭다운 옵션을 포함한 모든 HubSpot 딜·연락처·회사 속성을 자동으로 가져와서, 에이전트가 오류 없이 신뢰성 있게 데이터를 업데이트할 수 있게 해줘요.
class HubSpotConnector:
"""Handles all HubSpot API operations"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.hubapi.com/crm/v3/objects"
self.headers = {"Authorization": f"Bearer {api_key}"}
def get_properties(self) -> Dict:
"""Load all HubSpot properties"""
print("📡 HubSpotConnector: Loading properties...")
properties = {}
for obj_type in ['deals', 'contacts', 'companies']:
url = f"https://api.hubapi.com/crm/v3/properties/{obj_type}"
response = requests.get(url, headers=self.headers)
if response.status_code == 200:
props = response.json()['results']
prop_list = []
for prop in sorted(props, key=lambda x: x['name']):
prop_str = f"'{prop['name']}' - {prop['label']}"
if 'options' in prop and prop['options']:
valid_values = [opt['value'] for opt in prop['options']]
prop_str += f" | Valid values: {valid_values}"
prop_list.append(prop_str)
properties[obj_type] = prop_list
print(f"✅ HubSpotConnector: Loaded properties for {len(properties)} object types")
return properties
def get_data(self, object_type: str) -> List[Dict]:
"""Fetch data from HubSpot"""
print(f"📡 HubSpotConnector: Fetching {object_type} data...")
url = f"{self.base_url}/{object_type}"
params = {"limit": 100}
all_data = []
while url:
response = requests.get(url, headers=self.headers, params=params)
if response.status_code != 200:
raise Exception(f"HubSpot API error: {response.text}")
data = response.json()
all_data.extend(data.get("results", []))
url = data.get("paging", {}).get("next", {}).get("link")
params = {}
print(f"✅ HubSpotConnector: Loaded {len(all_data)} {object_type}")
return all_data
def batch_update(self, updates: Dict) -> None:
"""Perform batch updates to HubSpot"""
for object_type, update_list in updates.items():
if not update_list:
continue
print(f"📡 HubSpotConnector: Updating {len(update_list)} {object_type}...")
url = f"{self.base_url}/{object_type}/batch/update"
headers = {**self.headers, "Content-Type": "application/json"}
# Process in batches of 100
for i in range(0, len(update_list), 100):
batch = update_list[i:i+100]
payload = {"inputs": batch}
response = requests.post(url, headers=headers, json=payload)
if response.status_code not in [200, 202]:
raise Exception(f"HubSpot update error: {response.text}")
print(f"✅ HubSpotConnector: {object_type} updates completed")
Magistral(추론) 및 Mistral small LLM 함수 (Magistral (reasoning) and Mistral small LLM functions)
magistral_reasoning: 복잡한 쿼리 분석과 실행 계획에 Magistral 추론 모델을 사용해요. 사고 과정(thinking process)을 포함합니다.mistral_small_execution: 하위 에이전트 작업 실행에 Mistral Small 모델을 사용해요.
def magistral_reasoning(prompt: str) -> Dict[str, str]:
"""Use reasoning model for query analysis and planning"""
response = mistral_client.chat.complete(
model="magistral-medium-latest",
messages=[{"role": "user", "content": prompt}]
)
content = response.choices[0].message.content
reasoning = ""
conclusion = ""
for r in content:
if isinstance(r, ThinkChunk):
reasoning = r.thinking[0].text
elif isinstance(r, TextChunk):
conclusion = r.text
return {
"reasoning": reasoning,
"conclusion": conclusion
}
def mistral_small_execution(prompt: str) -> str:
"""Use Mistral Small for content generation"""
response = mistral_client.chat.complete(
model="mistral-small-latest",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
LeadAgent
Magistral 추론 모델이 구동하는 정교한 쿼리 분석과 실행 계획 담당이에요.
- Magistral의
thinking패턴을 사용해서 비즈니스 의도를 이해하고, 데이터 요구사항을 결정하며, 동적 하위 에이전트 사양과 함께 상세한 실행 계획을 만들어요. - 쿼리가 읽기 전용 분석인지 HubSpot에 대한 쓰기-백(write-back) 작업인지 판단해요.
analyze_query
class LeadAgent:
"""Lead Agent powered by Magistral reasoning model for query analysis and planning"""
def __init__(self, hubspot_properties):
self.hubspot_properties = hubspot_properties
self.name = "LeadAgent"
def analyze_query(self, query: str) -> Dict:
"""Analyze query using Magistral reasoning and create execution plan"""
print(f"🧠 {self.name}: Analyzing query with Magistral reasoning...")
analysis_prompt = f"""
Analyze this HubSpot query and create a detailed execution plan based on different hubspot properties provided by following the shared rules:
HUBSPOT_PROPERTIES: {self.hubspot_properties}
QUERY: {query}
RULES:
1. What is the user asking for?
2. Is this a read-only query or does it require HubSpot updates?
3. What sub-agents are needed to accomplish this?
4. What HubSpot data is required?
5. What's the execution sequence?
6. What should be the final output format?
7. Query can also be combination of read-only and write-back.
8. Query is read-only if it requires data read from HubSpot.
9. Query is write-back if it requires an update to existing values or writing/ assigning new values.
10. In the final conclusion just give only one JSON string nothing else. I don't need any explanation.
Provide a JSON execution plan with:
{{
"sub_agents": [
{{
"name": "agent_name",
"task": "specific task description",
"task_type": "read_only" or "write_back",
"input_data": ["deals", "contacts", "companies"],
"output_format": "expected output"
}}
]
}}
"""
# Use existing magistral_reasoning function
analysis = magistral_reasoning(analysis_prompt)
try:
# Extract JSON execution plan from conclusion
json_match = re.search(r'\{.*\}', analysis["conclusion"], re.DOTALL)
if json_match:
execution_plan = json.loads(json_match.group(0))
else:
raise ValueError("No JSON found in analysis")
except Exception as e:
print(f"⚠️ {self.name}: JSON parsing failed, using fallback plan")
execution_plan = {
"sub_agents": [{
"name": "general_analyzer",
"task": query,
"task_type": "read_only",
"input_data": ["deals", "contacts", "companies"],
"output_format": "summary"
}]
}
print(f"✅ {self.name}: Plan created - {len(execution_plan['sub_agents'])} sub-agents needed")
return {
"reasoning": analysis["reasoning"],
"execution_plan": execution_plan,
"conclusion": analysis["conclusion"]
}
SubAgent
쿼리 복잡성과 요구사항에 따라 즉석에서 생성되는 동적 에이전트예요.
- 데이터 분석, 비즈니스 로직 적용, CRM 업데이트를 포함한 빠른 작업 실행에 Mistral Small을 사용해요.
- 전문 역할(예:
priority_calculator,market_analyzer,deals_updater)이 자동으로 생성돼요. - 적절한 HubSpot 속성 검증과 함께 읽기 전용 작업과 쓰기-백 작업을 모두 처리해요.
execute
class SubAgent:
"""Dynamic Sub-Agent created on-the-fly for specific tasks"""
def __init__(self, name: str, task: str, task_type: str, input_data: List[str],
output_format: str):
self.name = name
self.task = task
self.task_type = task_type
self.input_data = input_data
self.output_format = output_format
def execute(self, data: Dict, properties_context: str, hubspot_updater=None) -> Dict:
"""Execute the assigned task"""
print(f"🤖 {self.name} ({self.task_type}): Executing task...")
if self.task_type == 'read_only':
agent_prompt = f"""
You are a {self.name} agent.
TASK: {self.task}
AVAILABLE HUBSPOT PROPERTIES:
{properties_context}
DATA AVAILABLE:
{json.dumps(data, indent=2)}
OUTPUT FORMAT: {self.output_format}
Provide your analysis only based on the available data.
"""
else: # write_back
agent_prompt = f"""
You are a {self.name} agent.
TASK: {self.task}
AVAILABLE HUBSPOT PROPERTIES:
{properties_context}
DATA AVAILABLE:
{json.dumps(data, indent=2)}
CRITICAL: Use exact HubSpot property names from the list above in your JSON output.
OUTPUT FORMAT: JSON format with the properties to be written to HubSpot
Provide updates using exact HubSpot property names.
"""
# Use existing mistral_small_execution function
result = mistral_small_execution(agent_prompt)
# Handle write-back operations
if self.task_type == 'write_back' and hubspot_updater:
try:
json_match = re.search(r'\{.*\}', result, re.DOTALL)
if json_match:
updates = json.loads(json_match.group(0))
hubspot_updater.batch_update(updates)
print(f"✅ {self.name}: Successfully updated HubSpot records")
except Exception as e:
print(f"❌ {self.name}: Update failed - {str(e)}")
return {"status": "error", "error": str(e), "raw_result": result}
print(f"✅ {self.name}: Task completed successfully")
return {"status": "success", "result": result}
SynthesisAgent
모든 하위 에이전트 결과를 일관된 비즈니스 통찰로 결합하는 최종 오케스트레이터예요.
- 실행 가능한 추천과 다음 단계가 포함된 사용자 친화적 응답을 만들기 위해 Mistral Small을 사용해요.
- 기술적인 에이전트 출력을 경영진이 바로 읽을 수 있는 요약과 전략적 지침으로 변환해요.
synthesize
class SynthesisAgent:
"""Final agent to synthesize all results into user-friendly response"""
def __init__(self):
self.name = "SynthesisAgent"
def synthesize(self, query: str, sub_agent_results: List[Dict], execution_plan: Dict) -> str:
"""Combine all sub-agent results into final answer"""
print(f"🔄 {self.name}: Synthesizing results from {len(sub_agent_results)} agents...")
# Prepare context from all sub-agent results
results_context = ""
for result in sub_agent_results:
results_context += f"\n{result['agent'].upper()} ({result['task_type']}):\n"
if result['result']['status'] == 'success':
results_context += f"{result['result']['result']}\n"
else:
results_context += f"Error: {result['result'].get('error', 'Unknown error')}\n"
results_context += "---\n"
synthesis_prompt = f"""
You are a final synthesizer agent. Create a comprehensive, user-friendly response based on all sub-agent results.
ORIGINAL QUERY: {query}
SUB-AGENT RESULTS:
{results_context}
TASK: Synthesize all the above results into a clear, actionable response for the user.
Guidelines:
1. Start with a direct answer to the user's query
2. Include key insights and findings
3. If updates were made, summarize what was changed
4. Provide actionable next steps if relevant
5. Keep it concise but comprehensive
6. Use a professional but friendly tone
Provide the final synthesized response:
"""
# Use existing mistral_small_execution function
final_answer = mistral_small_execution(synthesis_prompt)
print(f"✅ {self.name}: Final answer synthesized")
return final_answer
AgentOrchestrator
전체 멀티 에이전트 워크플로우와 HubSpot 통합을 관리하는 마스터 조정자예요.
- 쿼리 분석부터 하위 에이전트 실행, 최종 종합까지 전체 흐름을 조율해요.
- 에이전트 수명주기, 에이전트 간 데이터 흐름, HubSpot 연결을 관리해요.
- 멀티 에이전트 프로세스에 대한 풍부한 로깅과 모니터링을 제공해요.
process_query
class AgentOrchestrator:
"""Main orchestrator that coordinates all agents"""
def __init__(self, hubspot_api_key: str, mistral_api_key: str):
# Initialize global mistral client for existing functions
global mistral_client
mistral_client = Mistral(api_key=mistral_api_key)
# Initialize HubSpot connector
self.hubspot_connector = HubSpotConnector(hubspot_api_key)
# Load HubSpot data and properties
self.hubspot_properties = self.hubspot_connector.get_properties()
self.hubspot_data = {
"deals": self.hubspot_connector.get_data("deals"),
"contacts": self.hubspot_connector.get_data("contacts"),
"companies": self.hubspot_connector.get_data("companies")
}
# Initialize agents
self.lead_agent = LeadAgent(self.hubspot_properties)
self.synthesis_agent = SynthesisAgent()
self.active_sub_agents = []
print(f"🚀 AgentOrchestrator: System initialized with {sum(len(data) for data in self.hubspot_data.values())} HubSpot records")
def process_query(self, query: str) -> Dict:
"""Main method to process user queries through multi-agent workflow"""
print(f"\n🎯 Processing Query: {query[:100]}...")
print("=" * 80)
# Step 1: Lead Agent analyzes query using Magistral reasoning
analysis = self.lead_agent.analyze_query(query)
execution_plan = analysis["execution_plan"]
# Step 2: Create and execute sub-agents dynamically
sub_agent_results = []
self.active_sub_agents = []
for agent_config in execution_plan["sub_agents"]:
# Create sub-agent dynamically
sub_agent = SubAgent(
name=agent_config["name"],
task=agent_config["task"],
task_type=agent_config["task_type"],
input_data=agent_config["input_data"],
output_format=agent_config["output_format"]
)
self.active_sub_agents.append(sub_agent)
# Prepare data and context for this sub-agent
agent_data = {data_type: self.hubspot_data.get(data_type, [])
for data_type in agent_config["input_data"]}
# Build properties context
properties_context = ""
for data_type in agent_config["input_data"]:
if data_type in self.hubspot_properties:
properties_context += f"\n{data_type.upper()} PROPERTIES:\n"
properties_context += "\n".join(self.hubspot_properties[data_type])
properties_context += "\n"
# Execute sub-agent using mistral_small_execution
result = sub_agent.execute(agent_data, properties_context, self.hubspot_connector)
sub_agent_results.append({
"agent": sub_agent.name,
"task_type": sub_agent.task_type,
"result": result
})
# Step 3: Synthesis Agent creates final answer using mistral_small_execution
final_answer = self.synthesis_agent.synthesize(query, sub_agent_results, execution_plan)
print("=" * 80)
print("✨ Query processing completed!")
return {
"query": query,
"reasoning": analysis["reasoning"],
"execution_plan": execution_plan,
"sub_agent_results": sub_agent_results,
"active_agents": [agent.name for agent in self.active_sub_agents],
"final_answer": final_answer
}
멀티 에이전트 시스템 초기화 (Initialize the multi-agent system)
Python Output
orchestrator = AgentOrchestrator(
hubspot_api_key=HUBSPOT_API_KEY,
mistral_api_key=MISTRAL_API_KEY
)
테스트 쿼리 (Test Queries)
Query-1
Python Output
query = "Assign priorities to all deals based on deal value."
result = orchestrator.process_query(query)
업데이트 전 HubSpot 상태 (HubSpot status before updation)
업데이트 후 HubSpot 상태 (HubSpot status after updation)
동적으로 생성된 에이전트 (Dynamically Created Agents)
Python Output
agents = '\n'.join([f"{i + 1}. {agent}" for i, agent in enumerate(result['active_agents'])])
display(Markdown(agents))
답변 (Answer)
Python Output
from IPython.display import display, Markdown, Latex
display(Markdown(result['final_answer']))
Query-2
Python Output
query = """We're considering expanding into three new industry verticals and need comprehensive market
intelligence to inform our go-to-market strategy. Analyze our current customer base to identify
patterns in successful account profiles, understand the characteristics that predict customer
success, and use these insights to evaluate market opportunities. The analysis should identify
which industries show the strongest fit with our solution, what use cases resonate most effectively,
and what competitive landscape we would face. Develop ideal customer profiles for each target
market, estimate market size and penetration potential, and create a prioritized market entry
strategy with resource requirements and timeline projections for successful market penetration."""
result = orchestrator.process_query(query)
동적으로 생성된 에이전트 (Dynamically Created Agents)
Python Output
agents = '\n'.join([f"{i + 1}. {agent}" for i, agent in enumerate(result['active_agents'])])
display(Markdown(agents))
답변 (Answer)
Python Output
from IPython.display import display, Markdown, Latex
display(Markdown(result['final_answer']))