构建强大的AI应用：从候选人分析到会议助手

from agno.agent import Agent
from agno.models.nebius import Nebius
from agno.tools.github import GithubTools

# 创建分析代理
analyzer_agent = Agent(
    model=Nebius(id="deepseek-ai/DeepSeek-V3-0324"),
    tools=[GithubTools()],
    instructions="分析GitHub候选人资料并提供专业评估"
)

# 运行分析
response = analyzer_agent.run("分析用户john_doe的技术能力")

from agno.workflow import Workflow
from agno.tools.linear import LinearTools
from agno.tools.slack import SlackTools

# 创建工作流
meeting_workflow = Workflow(
    steps=[
        {"name": "transcribe", "agent": transcription_agent},
        {"name": "create_tasks", "agent": task_agent},
        {"name": "notify", "agent": notification_agent}
    ]
)

# 执行会议处理
result = meeting_workflow.run(meeting_notes)

from crewai import Agent, Task, Crew
from tools.custom_tools import DecisionTool, NotifyTool

# 创建监控代理
monitor_agent = Agent(
    role="价格监控专家",
    goal="检测产品价格变化并发送警报",
    tools=[DecisionTool(), NotifyTool()],
    verbose=True
)

# 设置监控任务
monitor_task = Task(
    description="监控产品价格变化",
    agent=monitor_agent,
    expected_output="价格变化警报"
)

class MultiAgentSystem:
    """多代理系统协调器"""
    
    def __init__(self):
        self.agents = {
            'analyzer': CandidateAnalyzerAgent(),
            'researcher': ResearchAgent(),
            'reporter': ReportGeneratorAgent()
        }
        self.workflow = SequentialWorkflow()
    
    async def process_request(self, request):
        """处理请求并通过工作流协调代理"""
        results = {}
        for agent_name in self.workflow.get_sequence():
            agent = self.agents[agent_name]
            results[agent_name] = await agent.process(request)
        return results

class VectorSearchManager:
    """向量搜索管理器"""
    
    def __init__(self, vector_store):
        self.vector_store = vector_store
        self.index_name = "conferences-talks-index"
    
    async def search_similar(self, query, top_k=5):
        """搜索相似内容"""
        query_embedding = await self._generate_embedding(query)
        results = await self.vector_store.search(
            query_embedding=query_embedding,
            top_k=top_k,
            index_name=self.index_name
        )
        return self._format_results(results)
    
    async def _generate_embedding(self, text):
        """生成文本嵌入"""
        response = self.embedding_model.embed(text)
        return response.embeddings[0]

class ResearchOrchestrator:
    """研究协调器"""
    
    def __init__(self):
        self.exa_client = Exa(api_key=os.getenv("EXA_API_KEY"))
        self.tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
    
    async def run_parallel_research(self, topic):
        """并行运行多源研究"""
        with concurrent.futures.ThreadPoolExecutor() as executor:
            exa_future = executor.submit(self._exa_search, topic)
            tavily_future = executor.submit(self._tavily_search, topic)
            
            exa_results = exa_future.result()
            tavily_results = tavily_future.result()
        
        return self._synthesize_results(exa_results, tavily_results)
    
    def _exa_search(self, topic):
        """Exa搜索"""
        return self.exa_client.search_and_contents(
            query=f"最新发展关于{topic}",
            num_results=5,
            text=True
        )