在目前的市場中,找到適合自己的工作非常困難!
最近,我正在探索 OpenAI Agents SDK 並建立 MCP 代理程式和代理程式工作流程。
為了運用我的學習成果,我想,為什麼不解決一個真正的、常見的問題呢?
因此我建立了這個多代理求職工作流程來找到適合我的工作!
在本文中,我將向您展示如何建立此代理工作流程!
開始吧!
在繼續之前,讓我們先了解一下工作流程代理是如何運作的!
首先,用戶提供他們的 LinkedIn 個人資料
第一位代理商使用 BrightData 的 MCP 伺服器分析您的個人資料、經驗、技能和職業道路。
根據分析,它會預測一個符合你的興趣和經驗的領域
然後,它使用 BrightData 的 MCP 伺服器從 Y Combinator 招聘網站上抓取當前空缺職位
最後,它會產生一份詳細的報告,其中包含您的個人資料摘要 + 為您量身定制的最佳工作匹配清單。
我對每項任務都使用了單獨的代理。
是的,我可以把幾個放在一起……但從我的實驗來看,這只會讓藥劑感到困惑並增加幻覺。 🙃
如果您更喜歡影片,您可以觀看這個:👇🏼
以下是我用來建立該專案的主要工具。
我一直在尋找簡單而可靠的選擇,而這些選擇結果證明非常適合這項工作。
我一直在尋找一種經濟實惠的方式,可以在單一工作流程中多次執行 LLM,而不必過度擔心成本或效能。
就在那時我發現了Nebius AI Studio 。
它可以輕鬆執行開源 AI 模型,性能也完全符合我的要求。它速度快、價格實惠,如果你想在不花太多錢的情況下使用強大的模型,它絕對是你的不二之選。
https://studio.nebius.com 嘗試 Nebius
LLM 的一大限制是什麼?他們通常無法即時存取即時資料。
MCP 透過允許 LLM 存取外部工具來解決此問題。
對於這個專案,我使用了Bright Data的 MCP(模型上下文協定)伺服器。
我在尋找讓代理商存取網路的方法時發現了它。有了這個MCP 伺服器,我們的代理可以:
搜尋和瀏覽網站
繞過位置阻止或驗證碼
抓取資料而不被阻止
它非常適合這個專案,我需要來自網路的即時資訊。
https://github.com/brightdata-com/brightdata-mcp 查看 BrightData 的 MCP 伺服器
由於我已經在探索 OpenAI Agents SDK 並學習如何建立 MCP 代理程式和代理程式工作流程,因此我決定將其用作該專案的基礎。
我發現它非常簡單和靈活,正是我建立多代理設定所需要的,它可以真正解決求職等現實任務。
它是一款輕量級工具,可幫助您輕鬆建立基於代理的應用程式。有了它,我可以:
為我的LLMs提供指導和工具
讓代理互相傳遞工作
在將資料傳送到模型之前新增基本檢查
如果您正在建立多步驟或多代理應用程式,它非常有用。
https://openai.github.io/openai-agents-python/ 查看 OpenAI Agents SDK
最後,我需要一種快速而乾淨的方法來建立 UI。
對於 Python 開發人員來說, Streamlit是顯而易見的選擇。
對於 Python 開發人員來說,Streamlit 是為其應用程式建立直覺 UI 的首選。
只需幾行程式碼,我就能擁有一個功能齊全的儀表板,可以在其中輸入 LinkedIn URL 並執行工作流程。
它使整個過程變得非常簡單。
https://streamlit.io/ 看 Streamlit
說得夠多了,讓我們開始建立代理商吧! 🔥
在執行此專案之前,請確保您已:
Python 3.10 或更高版本
Bright Data帳戶和 API 憑證
Nebius AI Studio帳戶和 API 金鑰
為了保持整潔和模組化,我們這樣組織專案:
# Folder Structure 👇🏼
job_finder_agent/
├── app.py # Streamlit web interface
├── job_agents.py # AI agent definitions and analysis logic
├── mcp_server.py # Bright Data MCP server management
├── requirements.txt # Python dependencies
├── assets/ # Static assets (images, GIFs)
└── .env # Environment variables (create this)
以下是簡要分析:
app.py
應用程式的入口點。這是用戶與該工具互動的地方。
job_agents.py
:包含所有與代理相關的邏輯和工作流程。
mcp_server.py
:初始化 MCP 伺服器。
assets/:保存 UI 中使用的任何視覺效果或媒體。
.env
:儲存 API 金鑰等敏感資料,請確保此文件包含在您的.gitignore
中。
有了這種結構,以後除錯、擴展甚至插入新代理都會非常容易。
首先,我們將建立專案所需的 Agents。為此,我們進入job_agents.py
檔。
在這裡,我們將導入所需的模組:
# job_agents.py 👇🏼
import os
import logging
import asyncio
from agents import (
Agent,
OpenAIChatCompletionsModel,
Runner,
set_tracing_disabled,
)
from agents.mcp import MCPServer
from openai import AsyncOpenAI
logger = logging.getLogger(__name__)
我們現在將定義一個非同步函數run_analysis
來啟動整個代理程式工作流程。
async def run_analysis(mcp_server: MCPServer, linkedin_url: str):
logger.info(f"Starting analysis for LinkedIn URL: {linkedin_url}")
api_key = os.environ["NEBIUS_API_KEY"]
base_url = "https://api.studio.nebius.ai/v1"
client = AsyncOpenAI(base_url=base_url, api_key=api_key)
set_tracing_disabled(disabled=True)
這是我們的第一個也是最關鍵的代理商之一。它分析用戶的 LinkedIn 個人資料,以提取相關的職業見解。
linkedin_agent = Agent(
name="LinkedIn Profile Analyzer",
instructions=f"""You are a LinkedIn profile analyzer.
Analyze profiles for:
- Professional experience and career progression
- Education and certifications
- Core skills and expertise
- Current role and company
- Previous roles and achievements
- Industry reputation (recommendations/endorsements)
Provide a structured analysis with bullet points and a brief executive summary.
""",
mcp_servers=[mcp_server],
model=OpenAIChatCompletionsModel(
model="meta-llama/Llama-3.3-70B-Instruct",
openai_client=client
)
)
💡 注意:此代理確實使用 mcp_server,因為它使用 BrightData 的抓取引擎從 LinkedIn 提取資料。
現在我們已經分析了個人資料,我們需要預測用戶最適合哪個領域。
工作建議代理將根據前一個代理的分析,建議用戶的首選領域
job_suggestions_agent = Agent(
name="Job Suggestions",
instructions=f"""You are a domain classifier that identifies the primary professional domain from a LinkedIn profile.
""",
model=OpenAIChatCompletionsModel(
model="meta-llama/Llama-3.3-70B-Instruct",
openai_client=client
)
)
💡 注意:這裡我們沒有使用 MCP 工具,因為它沒有執行任何函數呼叫。
然後,我們將建立一個招募平台 URL 產生器。該代理程式會取得域名,並為 Y Combinator 招聘平台建立 URL。
url_generator_agent = Agent(
name="URL Generator",
instructions=f"""You are a URL generator that creates Y Combinator job board URLs based on domains.
""",
model=OpenAIChatCompletionsModel(
model="meta-llama/Llama-3.3-70B-Instruct",
openai_client=client
)
)
接下來,我們將建立職位搜尋代理。該代理將存取生成的 URL 並提取真實的職位清單。
Job_search_agent = Agent(
name="Job Finder",
instructions=f"""You are a job finder that extracts job listings from Y Combinator's job board.
""",
mcp_servers=[mcp_server],
model=OpenAIChatCompletionsModel(
model="meta-llama/Llama-3.3-70B-Instruct",
openai_client=client
)
)
💡 注意:此代理程式再次使用 MCP 伺服器,因為它需要向 YC 的工作板進行即時抓取呼叫。
有時 YC 職位連結會經過身份驗證重定向🤷🏻♂️。此代理程式會清理並簡化這些 URL。
如果您使用不需要清理的職位公告板,則可以跳過此代理程式。
url_parser_agent = Agent(
name="URL Parser",
instructions=f"""You are a URL parser that transforms Y Combinator authentication URLs into direct job URLs.
""",
model=OpenAIChatCompletionsModel(
model="meta-llama/Llama-3.3-70B-Instruct",
openai_client=client
)
)
最後,我們將所有內容整合在一起。該代理會將使用者的個人資料、領域預測和作業結果匯總到一份簡潔的 Markdown 報告中。
summary_agent = Agent(
name="Summary Agent",
instructions=f"""You are a summary agent that creates comprehensive career analysis reports.
Ensure your response is well-formatted markdown that can be directly displayed.""",
model=OpenAIChatCompletionsModel(
model="meta-llama/Llama-3.3-70B-Instruct",
openai_client=client
)
)
它建立的報告在 Streamlit 甚至 markdown 渲染器中看起來很乾淨,因此您可以輕鬆地共享它或保存以備後用! 😉
現在所有代理程式都已準備就緒,讓我們建立以邏輯順序連接它們的實際流程。
在 Agents SDK 中,要建立工作流程,您必須將前一個代理程式的結果作為輸入傳遞,並指定starting_agent
作為處理目前任務的代理程式。
程式碼如下:
# Get LinkedIn profile analysis
logger.info("Running LinkedIn profile analysis")
linkedin_result = await Runner.run(starting_agent=linkedin_agent, input=query)
logger.info("LinkedIn profile analysis completed")
# Get job suggestions
logger.info("Getting job suggestions")
suggestions_result = await Runner.run(starting_agent=job_suggestions_agent, input=linkedin_result.final_output)
logger.info("Job suggestions completed")
# Get specific job matches
logger.info("Getting job link")
job_link_result = await Runner.run(starting_agent=url_generator_agent, input=suggestions_result.final_output)
logger.info("Job link generation completed")
# Get job matches
logger.info("Getting job matches")
job_search_result = await Runner.run(starting_agent=Job_search_agent, input=job_link_result.final_output)
logger.info("Job search completed")
# Parse URLs to get direct job links
logger.info("Parsing job URLs")
parsed_urls_result = await Runner.run(starting_agent=url_parser_agent, input=job_search_result.final_output)
logger.info("URL parsing completed")
# Create a single input for the summary agent
logger.info("Generating final summary")
summary_input = f"""LinkedIn Profile Analysis:
{linkedin_result.final_output}
Job Suggestions:
{suggestions_result.final_output}
Job Matches:
{parsed_urls_result.final_output}
Please analyze the above information and create a comprehensive career analysis report in markdown format."""
# Get final summary with a single call
summary_result = await Runner.run(starting_agent=summary_agent, input=summary_input)
logger.info("Summary generation completed")
return summary_result.final_output
現在,我們將初始化我們的 MCP 伺服器!
我們將首先建立一個mcp_server.py
文件,它將處理與設定和存取 MCP 伺服器相關的所有事項。
首先,我們導入必要的模組,包括用於處理非同步邏輯的 asyncio 和來自 Agents SDK 的 MCPServerStdio:
import os
import logging
import asyncio
from agents.mcp import MCPServerStdio
logger = logging.getLogger(__name__)
_mcp_server = None
接下來,我們將建立 Initializemcpserver fn,它將使用憑證初始化 mcp 伺服器
async def initialize_mcp_server():
"""Initialize MCP server."""
global _mcp_server
if _mcp_server:
return _mcp_server
try:
server = MCPServerStdio(
cache_tools_list=False,
params={
"command": "npx",
"args": ["-y", "@brightdata/mcp"],
"env": {
"API_TOKEN": os.environ["BRIGHT_DATA_API_KEY"],
"WEB_UNLOCKER_ZONE": "mcp_unlocker",
"BROWSER_AUTH": os.environ["BROWSER_AUTH"],
}
}
)
await asyncio.wait_for(server.__aenter__(), timeout=10)
_mcp_server = server
return server
except Exception as e:
logger.error(f"Error initializing MCP server: {e}")
return None
💡 注意:請確保您已在 .env 中正確設定環境變數(BRIGHT_DATA_API_KEY、BROWSER_AUTH),否則伺服器將無法啟動。
此函數做了幾件重要的事情:
如果伺服器已經啟動,則防止重新初始化。
啟動一個新的 MCP 伺服器實例。
處理超時並正常記錄任何故障。
為了保持整潔,我們加入了兩個實用函數:
等待伺服器準備就緒(wait_for_initialization)
另一個用於檢索伺服器實例(get_mcp_server)
它們在這裡:
async def wait_for_initialization():
"""Wait for MCP initialization to complete."""
return await initialize_mcp_server() is not None
def get_mcp_server():
"""Get the current MCP server instance."""
return _mcp_server
最後,我們將組裝所有內容並使用 Streamlit 為該代理程式建立 UI。
首先,我們導入所有必要的模組並設定 Streamlit 應用程式配置
import streamlit as st
import asyncio
import os
import logging
import nest_asyncio
import base64
from dotenv import load_dotenv
from job_agents import run_analysis
from mcp_server import wait_for_initialization, get_mcp_server
nest_asyncio.apply()
load_dotenv()
logger = logging.getLogger(__name__)
# Set page config
st.set_page_config(
page_title="LinkedIn Profile Analyzer",
page_icon="🔍",
layout="wide"
)
接下來,一旦 MCP 伺服器初始化完成,我們將使用 asyncio 非同步呼叫 run_analysis() 函數。
當使用者點擊「分析設定檔」按鈕時,將觸發此功能。
# Initialize session state
if 'analysis_result' not in st.session_state:
st.session_state.analysis_result = ""
if 'is_analyzing' not in st.session_state:
st.session_state.is_analyzing = False
async def analyze_profile(linkedin_url: str):
try:
if not await wait_for_initialization():
st.error("Failed to initialize MCP server")
return
result = await run_analysis(get_mcp_server(), linkedin_url)
st.session_state.analysis_result = result
except Exception as e:
logger.error(f"Error analyzing LinkedIn profile: {str(e)}")
st.error(f"Error analyzing LinkedIn profile: {str(e)}")
finally:
st.session_state.is_analyzing = False
我們也初始化了一些會話狀態變數來管理互動之間的應用程式狀態。
現在到了最有趣的部分,也就是主 UI 🤩!我們將建立一個簡單的 UI 來與我們的 Agent 互動。
def main():
# Load and encode images
with open("./assets/bright-data-logo.png", "rb") as bright_data_file:
bright_data_base64 = base64.b64encode(bright_data_file.read()).decode()
# Create title with embedded images
title_html = f"""
<div style="display: flex; align-items: center; gap: 0px; margin: 0; padding: 0;">
<h1 style="margin: 0; padding: 0;">
Job Searcher Agent with
<img src="data:image/png;base64,{bright_data_base64}" style="height: 110px; margin: 0; padding: 0;"/>
</h1>
</div>
"""
st.markdown(title_html, unsafe_allow_html=True)
st.markdown("---")
# Sidebar
with st.sidebar:
st.image("./assets/Nebius.png", width=150)
api_key = st.text_input("Enter your API key", type="password")
st.divider()
st.subheader("Enter LinkedIn Profile URL")
linkedin_url = st.text_input("LinkedIn URL", placeholder="https://www.linkedin.com/in/username/")
if st.button("Analyze Profile", type="primary", disabled=st.session_state.is_analyzing):
if not linkedin_url:
st.error("Please enter a LinkedIn profile URL")
return
if not api_key:
st.error("Please enter your API key")
return
st.session_state.is_analyzing = True
st.session_state.analysis_result = ""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(analyze_profile(linkedin_url))
finally:
loop.close()
# Results section
if st.session_state.analysis_result:
st.subheader("Analysis Results")
st.markdown(st.session_state.analysis_result)
# Loading state
if st.session_state.is_analyzing:
st.markdown("---")
with st.spinner("Analyzing profile... This may take a few minutes."):
st.empty()
if __name__ == "__main__":
main()
就是這樣!現在,我們擁有一個功能齊全的 Streamlit UI,它連接到一個多智能體 AI 系統,具有瀏覽器自動化功能,並透過 MCP 進行抓取。
現在一切都已設定完畢,讓我們在本地執行這個應用程式!
首先,我們將建立一個虛擬環境:
python -m venv venv
source venv/bin/activate # On Windows, use: venv\Scripts\activate
接下來,安裝requirements.txt
中列出的所有必需軟體包:
pip install -r requirements.txt
最後,執行應用程式:
streamlit run app.py
現在前往http://localhost:8501 :
我們的代理商已準備好分析 LinkedIn 個人資料並像魔術一樣找到匹配的工作。 🪄
就這樣! 🎉
我們已經成功地建立了一個多代理工作流程,可以使用 MCP 伺服器為我們尋找工作。
您可以在這裡找到完整的程式碼: Github Repo
如果您覺得這篇文章有用,請與您的同儕分享
另外,請關注我以獲取更多類似內容:
{% 嵌入https://dev.to/arindam_1729
對於付費合作,請發送電子郵件至: [email protected] 。
感謝您的閱讀!
原文出處:https://dev.to/arindam_1729/i-built-an-ai-agent-that-finds-jobs-for-me-5427