researcher =Agent(
    role='Research Specialist',
    goal='Conduct comprehensive research on {topic}',
    backstory='Expert at finding and analyzing information',
    tools=[search_tool],
    verbose=True
)

writer =Agent(
    role='Content Writer',
    goal='Create comprehensive study materials',
    backstory='Skilled at organizing complex information',
    tools=[image_tool],
    verbose=True
)

@server.call_tool()
async def search_web(arguments: dict)-> list[TextContent]:
"""Brave Search API integration"""
    query = arguments.get("query","")


    headers ={"X-Subscription-Token": BRAVE_API_KEY}
params={"q": query,"count":10}


    response = requests.get(BRAVE_SEARCH_URL, headers=headers,params=params)
    results = response.json()


return[TextContent(type="text", text=json.dumps(results))]

@server.call_tool()
async def generate_image(arguments: dict)-> list[TextContent]:
"""Segmind API image generation"""
    prompt = arguments.get("prompt","")


    data ={
"prompt": prompt,
"style":"photographic",
"samples":1
}


    response = requests.post(SEGMIND_URL, jsnotallow=data, headers=headers)
# Save and return image path

def apply_custom_css():
    st.markdown("""
<style>
.main-header {
        background: linear-gradient(90deg,#667eea 0%, #764ba2 100%);
        padding:2rem;
        border-radius:10px;
        color: white;
        text-align: center;
        margin-bottom:2rem;
}


.result-card {
        background: white;
        padding:1.5rem;
        border-radius:10px;
        box-shadow:02px4px rgba(0,0,0,0.1);
        margin:1rem0;
}
</style>
""", unsafe_allow_html=True)

def display_results():
    tab1, tab2, tab3 = st.tabs([
"?? Search Results",
"?? Summary",
"?? Generated Images"
])


with tab1:
        display_search_results()
with tab2:
        display_summary_with_download()
with tab3:
        display_image_gallery()

def run_research(topic: str)->Dict:
"""Execute research workflow"""
try:
# Run main.py as subprocess
        result = subprocess.run(
[sys.executable,"main.py", topic],
            capture_output=True,
            text=True,
            timeout=300# 5-minute timeout
)


return{
"search_results": extract_search_results(),
"summary": extract_summary_from_output(result.stdout),
"images": get_generated_images(),
"success":True
}
except subprocess.TimeoutExpired:
return{"success":False,"error":"Research timeout"}

def robust_mcp_call(server_path: str, max_retries:int=3):
for attempt in range(max_retries):
try:
# MCP server communication
return call_mcp_server(server_path)
exceptExceptionas e:
if attempt == max_retries -1:
                st.error(f"?? Server Caiunavailable: {e}")
            time.sleep(2** attempt)# Exponential backoff

def extract_summary_from_output(output: str)-> str:
    patterns =[
        r"FINAL RESULT:\s*(.+?)(?=\n\n|\Z)",
        r"## Final Answer:\s*(.+?)(?=\n\n|\Z)",
        r"Summary:\s*(.+?)(?=\n\n|\Z)"
]


for pattern in patterns:
        match = re.search(pattern, output, re.DOTALL | re.IGNORECASE)
if match:
return clean_summary_text(match.group(1))


return"Summary extraction failed"

pip install -r requirements.txt


# 配置 API 密鑰
export BRAVE_API_KEY="your-brave-key"
export SEGMIND_API_KEY="your-segmind-key"
export OPENAI_API_KEY="your-openai-key"

# 啟動(dòng) MCP 服務(wù)器
python servers/search_server.py &
python servers/image_server.py &


# 啟動(dòng) Streamlit 應(yīng)用
streamlit run streamlit_app.py

from typing importAny
import httpx
from mcp.server.fastmcp importFastMCP
import os
import requests
import base64
import logging
from pathlib importPath
from dotenv import load_dotenv
load_dotenv()


# 設(shè)置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("image_server")


# 初始化 FastMCP 服務(wù)器
mcp =FastMCP("image_server")


# 獲取當(dāng)前目錄
current_dir =Path(__file__).parent
output_dir = current_dir /"images"
os.makedirs(output_dir, exist_ok=True)


# 驗(yàn)證 API 密鑰
api_key = os.getenv("SEGMIND_API_KEY")
ifnot api_key:
    logger.error("SEGMIND_API_KEY environment variable is not set!")
raiseRuntimeError("Missing Segmind API key")


url ="https://api.segmind.com/v1/imagen-4"


@mcp.tool(name="image_creation_openai", descriptinotallow="Create an image using Segmind API")
def image_creation_openai(query: str Latinos, image_name: str)-> str:
try:
        logger.info(f"Creating image for query: {query}")


# 請(qǐng)求負(fù)載
        data ={
"prompt": f"Generate an image: {query}",
"negative_prompt":"blurry, pixelated",
"aspect_ratio":"4:3"
}


        headers ={'x-api-key': os.getenv("SEGMIND_API_KEY")}


# 添加超時(shí)和錯(cuò)誤處理
try:
            response = requests.post(url, jsnotallow=data, headers=headers, timeout=30)
            response.raise_for_status()
except requests.exceptions.RequestExceptionas e:
            logger.error(f"API request failed: {e}")
return{"success":False,"error": f"API request failed: {str(e)}"}


# 保存圖像
        image_path = output_dir / f"{image_name}.jpeg"
with open(image_path,"wb")as f:
            f.write(response.content)


        logger.info(f"Image saved to {image_path}")
return{"success":True,"image_path": str(image_path)}
exceptExceptionas e:
        logger.exception("Image creation failed")
return{"success":False,"error": str(e)}


if __name__ =="__main__":
    logger.info("Starting Image Creation MCP Server")
try:
        mcp.run(transport="stdio")
exceptExceptionas e:
        logger.exception("Server crashed")
# 在 Windows 中添加暫停以查看錯(cuò)誤
        input("Press Enter to exit...")
raise

from typing importAny,Dict,List
import requests
from mcp.server.fastmcp importFastMCP
import os
import logging
import json
from pathlib importPath
from dotenv import load_dotenv


load_dotenv()


# 設(shè)置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("search_server")


# 初始化 FastMCP 服務(wù)器
mcp =FastMCP("search_server")


# 獲取當(dāng)前目錄
current_dir =Path(__file__).parent
results_dir = current_dir /"search_results"
os.makedirs(results_dir, exist_ok=True)


# 驗(yàn)證 API 密鑰
api_key = os.getenv("BRAVE_API_KEY")
ifnot api_key:
    logger.warning("BRAVE_API_KEY environment variable is not set!")
    logger.warning("Search functionality will be limited or unavailable")


# Brave Search API 端點(diǎn)
BRAVE_SEARCH_URL ="https://api.search.brave.com/res/v1/web/search"


@mcp.tool(name="brave_search", descriptinotallow="Search the web using Brave Search API")
def brave_search(query: str, count:int=10)->Dict[str,Any]:
"""
使用BraveSearch API 搜索網(wǎng)絡(luò)


參數(shù)：
        query:搜索查詢字符串
        count:返回的結(jié)果數(shù)量（最大20）


返回：
包含搜索結(jié)果的字典
"""
try:
        logger.info(f"Searching for: {query}")


ifnot api_key:
return{
"success":False,
"error":"BRAVE_API_KEY not configured",
"results":[]
}


# 將計(jì)數(shù)限制在合理范圍內(nèi)
        count = max(1, min(count,20))


# 請(qǐng)求頭
        headers ={
"Accept":"application/json",
"Accept-Encoding":"gzip",
"X-Subscription-Token": api_key
}


# 請(qǐng)求參數(shù)
params={
"q": query,
"count": count,
"search_lang":"en",
"country":"US",
"safesearch":"moderate",
"freshness":"pw",# 過去一周以獲取更新的結(jié)果
"text_decorations":False,
"spellcheck":True
}


# 發(fā)起 API 請(qǐng)求
try:
            response = requests.get(
                BRAVE_SEARCH_URL,
                headers=headers,
params=params,
                timeout=30
)
            response.raise_for_status()
except requests.exceptions.RequestExceptionas e:
            logger.error(f"Search API request failed: {e}")
return{
"success":False,
"error": f"Search API request failed: {str(e)}",
"results":[]
}


# 解析響應(yīng)
try:
            data = response.json()
except json.JSONDecodeErroras e:
            logger.error(f"Failed to parse search response: {e}")
return{
"success":False,
"error":"Failed to parse search response",
"results":[]
}


# 提取并格式化結(jié)果
        search_results =[]
        web_results = data.get("web",{}).get("results",[])


for result in web_results:
            search_result ={
"title": result.get("title",""),
"url": result.get("url",""),
"description": result.get("description",""),
"published": result.get("published",""),
"thumbnail": result.get("thumbnail",{}).get("src","")if result.get("thumbnail")else""
}
            search_results.append(search_result)


# 將結(jié)果保存到文件以供參考
try:
            results_file = results_dir / f"search_{query.replace(' ', '_')[:50]}.json"
with open(results_file,'w', encoding='utf-8')as f:
                json.dump({
"query": query,
"timestamp": data.get("query",{}).get("posted_at",""),
"results": search_results
}, f, indent=2, ensure_ascii=False)
            logger.info(f"Search results saved to {results_file}")
exceptExceptionas e:
            logger.warning(f"Failed to save search results: {e}")


        logger.info(f"Found {len(search_results)} search results")


return{
"success":True,
"query": query,
"total_results": len(search_results),
"results": search_results
}


exceptExceptionas e:
        logger.exception("Search operation failed")
return{
"success":False,
"error": str(e),
"results":[]
}


@mcp.tool(name="search_news", descriptinotallow="Search for news using Brave Search API")
def search_news(query: str, count:int=5)->Dict[str,Any]:
"""
使用BraveSearch API 搜索新聞


參數(shù)：
        query:搜索查詢字符串
        count:返回的新聞結(jié)果數(shù)量（最大20）


返回：
包含新聞搜索結(jié)果的字典
"""
try:
        logger.info(f"Searching news for: {query}")


ifnot api_key:
return{
"success":False,
"error":"BRAVE_API_KEY not configured",
"results":[]
}


# 將計(jì)數(shù)限制在合理范圍內(nèi)
        count = max(1, min(count,20))


# 請(qǐng)求頭
        headers ={
"Accept":"application/json",
"Accept-Encoding":"gzip",
"X-Subscription-Token": api_key
}


# 新聞搜索的請(qǐng)求參數(shù)
params={
"q": query,
"count": count,
"search_lang":"en",
"country":"US",
"safesearch":"moderate",
"freshness":"pd",# 過去一天以獲取最新新聞
"text_decorations":False,
"result_filter":"news"# 專注于新聞結(jié)果
}


# 發(fā)起 API 請(qǐng)求
try:
            response = requests.get(
                BRAVE_SEARCH_URL,
                headers=headers,
params=params,
                timeout=30
)
            response.raise_for_status()
except requests.exceptions.RequestExceptionas e:
            logger.error(f"News search API request failed: {e}")
return{
"success":False,
"error": f"News search API request failed: {str(e)}",
"results":[]
}


# 解析響應(yīng)
try:
            data = response.json()
except json.JSONDecodeErroras e:
            logger.error(f"Failed to parse news search response: {e}")
return{
"success":False,
"error":"Failed to parse news search response",
"results":[]
}


# 提取新聞結(jié)果
        news_results =[]


# 檢查響應(yīng)中的新聞部分
        news_data = data.get("news",{}).get("results",[])
ifnot news_data:
# 如果沒有專用新聞部分，則回退到網(wǎng)頁結(jié)果
            news_data = data.get("web",{}).get("results",[])


for result in news_data:
            news_result ={
"title": result.get("title",""),
"url": result.get("url",""),
"description": result.get("description",""),
"published": result.get("age", result.get("published","")),
"source": result.get("profile",{}).get("name","")if result.get("profile")else"",
"thumbnail": result.get("thumbnail",{}).get("src","")if result.get("thumbnail")else""
}
            news_results.append(news_result)


        logger.info(f"Found {len(news_results)} news results")


return{
"success":True,
"query": query,
"total_results": len(news_results),
"results": news_results
}


exceptExceptionas e:
        logger.exception("News search operation failed")
return{
"success":False,
"error": str(e),
"results":[]
}


if __name__ =="__main__":
    logger.info("Starting Brave Search MCP Server")
try:
        mcp.run(transport="stdio")
exceptExceptionas e:
        logger.exception("Search server crashed")
# 在 Windows 中添加暫停以查看錯(cuò)誤
        input("Press Enter to exit...")
raise

from crewai importAgent,Task,Crew, LLM
from crewai_tools importMCPServerAdapter
from mcp importStdioServerParameters
import sys
import platform
from pathlib importPath
import os
import warnings
from pydantic importPydanticDeprecatedSince20
from dotenv import load_dotenv
import traceback
import subprocess
from pydantic importBaseModel,Field
classSummary(BaseModel):
    summary: str =Field(descriptinotallow="研究成果的詳細(xì)摘要")
    image_path: str =Field(descriptinotallow="代理創(chuàng)建的圖像文件路徑")


# 加載環(huán)境變量
load_dotenv()


def get_available_llm():
"""從環(huán)境變量中獲取第一個(gè)可用的 LLM"""
    llm_configs =[
{
"name":"Groq Llama",
"model":"groq/llama-3.3-70b-versatile",
"api_key_env":"GROQ_API_KEY",
"temperature":0.7
},
{
"name":"OpenAI GPT-4",
"model":"gpt-4o-mini",
"api_key_env":"OPENAI_API_KEY",
"temperature":0.7
},
{
"name":"Anthropic Claude",
"model":"claude-3-haiku-20240307",
"api_key_env":"ANTHROPIC_API_KEY",
"temperature":0.7
},
{
"name":"Ollama Local",
"model":"ollama/llama3.2",
"api_key_env":None,# 本地模型無需 API 密鑰
"temperature":0.7
}
]


print("?? 檢查可用的 LLM 提供者...")


for config in llm_configs:
try:
if config["api_key_env"]isNone:
# 對(duì)于本地模型如 Ollama，嘗試無需 API 密鑰
print(f"? 嘗試 {config['name']} (本地)...")
                llm = LLM(
                    model=config["model"],
                    temperature=config["temperature"],
                    max_tokens=1000,
)
print(f"? 使用 {config['name']}: {config['model']}")
return llm
else:
                api_key = os.getenv(config["api_key_env"])
if api_key:
print(f"? 嘗試 {config['name']}...")
                    llm = LLM(
                        model=config["model"],
                        temperature=config["temperature"],
                        api_key=api_key
)
print(f"? 使用 {config['name']}: {config['model']}")
return llm
else:
print(f"??  {config['name']} API 密鑰未在環(huán)境中找到")
exceptExceptionas e:
print(f"? {config['name']} 失敗: {str(e)[:100]}...")
continue


# 如果全部失敗，則回退到基本配置
print("??  使用回退 LLM 配置...")
return LLM(
        model="groq/llama-3.3-70b-versatile",
        temperature=0.7,
        api_key=os.getenv("GROQ_API_KEY","")
)


# 配置具有回退選項(xiàng)的 LLM
llm = get_available_llm()


# 抑制警告
warnings.filterWarnings("ignore", category=PydanticDeprecatedSince20)


# 獲取當(dāng)前目錄
base_dir =Path(__file__).parent.resolve()


print(f"Python 可執(zhí)行文件: {sys.executable}")
print(f"當(dāng)前目錄: {os.getcwd()}")
print(f"基礎(chǔ)目錄: {base_dir}")


# 確定適用于 Windows 的正確 npx 命令
npx_cmd ="npx.cmd"if platform.system()=="Windows"else"npx"


def check_npx_availability():
"""檢查 npx 是否可用且正常工作"""
try:
        result = subprocess.run([npx_cmd,"--version"],
                              capture_output=True, text=True, timeout=10)
if result.returncode ==0:
print(f"? NPX 可用: {result.stdout.strip()}")
returnTrue
else:
print(f"? NPX 檢查失敗: {result.stderr}")
returnFalse
exceptExceptionas e:
print(f"? NPX 不可用: {e}")
returnFalse


def check_python_server():
"""檢查 Python 圖像服務(wù)器是否存在"""
    server_path = base_dir /"servers"/"image_server.py"
if server_path.exists():
print(f"? 找到 Python 圖像服務(wù)器: {server_path}")
returnTrue
else:
print(f"? 未找到 Python 圖像服務(wù)器: {server_path}")
returnFalse


def check_search_server():
"""檢查 Python 搜索服務(wù)器是否存在"""
    server_path = base_dir /"servers"/"search_server.py"
if server_path.exists():
print(f"? 找到 Python 搜索服務(wù)器: {server_path}")
returnTrue
else:
print(f"? 未找到 Python 搜索服務(wù)器: {server_path}")
returnFalse


def get_working_servers():
"""獲取工作服務(wù)器配置列表"""
    working_servers =[]


print("\n"+"="*50)
print("診斷 MCP 服務(wù)器")
print("="*50)


# 首先檢查 Python 圖像服務(wù)器（最有可能工作）
    python_server_available = check_python_server()
if python_server_available:
        image_server_params =StdioServerParameters(
            command="python",
            args=[
                str(base_dir /"servers"/"image_server.py"),
],
            env={"UV_PYTHON":"3.12",**os.environ},
)
        working_servers.append(("Image Server", image_server_params))
print("? 圖像服務(wù)器已配置")
else:
print("? 跳過圖像服務(wù)器（未找到服務(wù)器文件）")


# 檢查 Python 搜索服務(wù)器
    search_server_available = check_search_server()
if search_server_available:
        search_server_params =StdioServerParameters(
            command="python",
            args=[
                str(base_dir /"servers"/"search_server.py"),
],
            env={"UV_PYTHON":"3.12",**os.environ},
)
        working_servers.append(("Python Search Server", search_server_params))
print("? Python 搜索服務(wù)器已配置")
else:
print("? 跳過 Python 搜索服務(wù)器（未找到服務(wù)器文件）")


# 僅為文件系統(tǒng)服務(wù)器檢查 NPX 可用性
    npx_available = check_npx_availability()


# 僅在 Node.js 版本足夠新時(shí)添加 NPX 服務(wù)器
if npx_available:
        node_version_check = check_node_version()
if node_version_check:
# 文件系統(tǒng)服務(wù)器配置
            filesystem_server_params =StdioServerParameters(
                command=npx_cmd,
                args=[
"-y",
"@modelcontextprotocol/server-filesystem",
                    os.path.join(os.path.expanduser("~"),"Downloads")
],
)
            working_servers.append(("Filesystem Server", filesystem_server_params))
print("? 文件系統(tǒng)服務(wù)器已配置")
else:
print("??  由于 Node.js 版本兼容性問題，跳過 NPX 文件系統(tǒng)服務(wù)器")
print("?? 要啟用文件系統(tǒng)服務(wù)器，請(qǐng)將 Node.js 更新到 18+ 或 20+ 版本")
print("   訪問: https://nodejs.org/en/download/")
else:
print("? 跳過 NPX 文件系統(tǒng)服務(wù)器（NPX 不可用）")


print(f"\n找到 {len(working_servers)} 個(gè)服務(wù)器配置")
return working_servers


def check_node_version():
"""檢查 Node.js 版本是否兼容"""
try:
        result = subprocess.run(["node","--version"],
                              capture_output=True, text=True, timeout=10)
if result.returncode ==0:
            version = result.stdout.strip()
print(f"Node.js 版本: {version}")
# 提取主版本號(hào)
            major_version =int(version.lstrip('v').split('.')[0])
if major_version >=18:
print("? Node.js 版本兼容")
returnTrue
else:
print(f"??  Node.js 版本 {version} 可能過舊（推薦 v18+）")
returnFalse
returnFalse
exceptExceptionas e:
print(f"? 無法檢查 Node.js 版本: {e}")
returnFalse


classCustomMCPServerAdapter(MCPServerAdapter):
"""具有增加超時(shí)的自定義 MCP 服務(wù)器適配器"""
def __init__(self,*args,**kwargs):
super().__init__(*args,**kwargs)
self.timeout =90# 將超時(shí)增加到 90 秒


def test_servers_individually(server_configs):
"""單獨(dú)測試每個(gè)服務(wù)器以識(shí)別問題服務(wù)器"""
    working_servers =[]


print("\n"+"="*50)
print("單獨(dú)測試服務(wù)器")
print("="*50)


for name, server_params in server_configs:
print(f"\n測試 {name}...")
try:
withCustomMCPServerAdapter([server_params])as tools:
print(f"? {name} 連接成功！")
print(f"  可用工具: {[tool.name for tool in tools]}")
                working_servers.append(server_params)
exceptExceptionas e:
print(f"? {name} 失敗: {str(e)[:100]}...")
continue


return working_servers


def create_agent_and_tasks(tools=None):
"""創(chuàng)建代理和任務(wù)（帶或不帶工具）"""
    tools_list = tools or[]


# 根據(jù)可用工具調(diào)整角色和任務(wù)
if tools_list:
        tool_names =[getattr(tool,'name','unknown')for tool in tools_list]
print(f"代理將有權(quán)訪問: {tool_names}")


        role ="AI Research Creator with Tools"
        goal ="使用可用的 MCP 工具深入研究主題，創(chuàng)建全面的圖表，并保存摘要"
        backstory ="擅長使用 MCP 工具收集信息、創(chuàng)建視覺表示并保存研究成果的 AI 研究者和創(chuàng)作者。"
else:
        role ="AI Research Creator"
        goal ="使用內(nèi)置知識(shí)深入研究和分析主題"
        backstory ="擅長分析主題并使用可用知識(shí)提供詳細(xì)見解的 AI 研究者。"


    agent =Agent(
        role=role,
        goal=goal,
        backstory=backstory,
        tools=tools_list,
        llm=llm,
        verbose=True,
)


if tools_list:
        research_task =Task(
            descriptinotallow="使用可用的 MCP 工具深入研究主題 '{topic}'。如果有圖像創(chuàng)建工具，創(chuàng)建一個(gè)深入的圖表，展示主題的工作原理，包括關(guān)鍵組件、流程和關(guān)系。",
            expected_output="全面的研究摘要，如果可能，包括成功創(chuàng)建的圖表/圖像，說明主題。",
            agent=agent,
)


        summary_task =Task(
            descriptinotallow="創(chuàng)建研究成果的詳細(xì)摘要。如果有文件系統(tǒng)工具，將其保存為 Downloads 文件夾中的文本文件。包括關(guān)鍵見解、重要細(xì)節(jié)和對(duì)創(chuàng)建的任何圖表的引用。",
            expected_output="研究成果的詳細(xì)摘要，如果有文件系統(tǒng)訪問權(quán)限，優(yōu)選保存為文本文件。最終響應(yīng)應(yīng)采用 pydantic 模型 Summary 的格式",
            agent=agent,
            output_pydantic=Summary
)
else:
        research_task =Task(
            descriptinotallow="使用您的知識(shí)深入研究和分析主題 '{topic}'。提供有關(guān)其工作原理的詳細(xì)見解，包括關(guān)鍵組件、流程和關(guān)系。",
            expected_output="對(duì)主題的全面分析和解釋，包含詳細(xì)見解。",
            agent=agent,
)


        summary_task =Task(
            descriptinotallow="創(chuàng)建分析的詳細(xì)摘要，突出主題的最重要方面、關(guān)鍵見解和實(shí)際意義。",
            expected_output="結(jié)構(gòu)良好的摘要，包含主題的關(guān)鍵發(fā)現(xiàn)和見解。最終響應(yīng)應(yīng)采用 pydantic 模型 Summary 的格式",
            agent=agent,
            output_pydantic=Summary,
            markdown=True,# 啟用最終輸出的 markdown 格式
            output_file="report.md"
)


return agent,[research_task, summary_task]


def main():
"""運(yùn)行 CrewAI 應(yīng)用的主函數(shù)"""
# 獲取可用服務(wù)器配置
    server_configs = get_working_servers()


ifnot server_configs:
print("\n??  無可用 MCP 服務(wù)器。僅以回退模式運(yùn)行。")
        run_fallback_mode()
return


# 單獨(dú)測試服務(wù)器以找到工作中的服務(wù)器
    working_server_params = test_servers_individually(server_configs)


ifnot working_server_params:
print("\n??  無 MCP 服務(wù)器工作。以回退模式運(yùn)行。")
        run_fallback_mode()
return


try:
print(f"\n? 使用 {len(working_server_params)} 個(gè)工作中的 MCP 服務(wù)器")
print("初始化 MCP 服務(wù)器適配器...")


withCustomMCPServerAdapter(working_server_params)as tools:
print(f"成功連接到 MCP 服務(wù)器！")
print(f"可用工具: {[tool.name for tool in tools]}")


# 使用 MCP 工具創(chuàng)建代理和任務(wù)
            agent, tasks = create_agent_and_tasks(tools)


# 創(chuàng)建具有錯(cuò)誤處理的 crew
            crew =Crew(
                agents=[agent],
                tasks=tasks,
                verbose=True,
                reasnotallow=True,
)


# 獲取用戶輸入
            topic = input("\n請(qǐng)輸入要研究的主題: ").strip()
ifnot topic:
                topic ="artificial intelligence"
print(f"未提供主題，使用默認(rèn)值: {topic}")


# 使用重試機(jī)制執(zhí)行 crew
            max_retries =2
for attempt in range(max_retries +1):
try:
print(f"\n開始研究: {topic} (嘗試 {attempt + 1})")
                    result = crew.kickoff(inputs={"topic": topic})
# print("\n" + "="*50)
# print("來自代理的最終結(jié)果")
# print("="*50)


                    response = result["summary"]
print(response)
print(f"摘要任務(wù)輸出: {tasks[1].output}")
return response
exceptExceptionas e:
if attempt < max_retries:
print(f"??  嘗試 {attempt + 1} 失敗: {str(e)[:100]}...")
print(f"?? 重試... ({attempt + 2}/{max_retries + 1})")
continue
else:
print(f"? 所有嘗試均失敗。錯(cuò)誤: {e}")
raise e


exceptExceptionas e:
print(f"使用 MCP 工具運(yùn)行時(shí)出錯(cuò): {e}")
        traceback.print_exc()
print("\n回退到無 MCP 工具的基本代理...")
        run_fallback_mode()


def run_fallback_mode():
"""在無 MCP 工具的情況下運(yùn)行應(yīng)用"""
print("\n"+"="*50)
print("以回退模式運(yùn)行")
print("="*50)


# 創(chuàng)建不帶 MCP 工具但帶 LLM 的回退代理
    agent, tasks = create_agent_and_tasks()


    crew =Crew(
        agents=[agent],
        tasks=tasks,
        verbose=True,
        reasnotallow=True,
)


# 獲取回退模式的輸入
    topic = input("請(qǐng)輸入要研究的主題（回退模式）: ").strip()
ifnot topic:
        topic ="artificial intelligence"
print(f"未提供主題，使用默認(rèn)值: {topic}")


print(f"\n開始研究: {topic}（無 MCP 工具）")
    result = crew.kickoff(inputs={"topic": topic})
print("\n"+"="*50)
print("最終結(jié)果（回退模式）:")
print("="*50)
print(result["summary"])
return result["summary"]


if __name__ =="__main__":
print("?? 啟動(dòng) CrewAI MCP 演示")
print("\n?? 設(shè)置說明:")
print("   要使用更多 MCP 服務(wù)器，請(qǐng)將 Node.js 更新到 v18+: https://nodejs.org")
print("   在 .env 文件中添加 API 密鑰以支持更多 LLM 提供者")
print("   支持: GROQ_API_KEY, OPENAI_API_KEY, ANTHROPIC_API_KEY, BRAVE_API_KEY")
    result = main()
#print(result)

import streamlit as st
import subprocess
import sys
import os
from pathlib importPath
import glob
from PIL importImage
import re


def find_venv_python():
"""從虛擬環(huán)境中找到正確的 Python 可執(zhí)行文件"""
    current_dir =Path(__file__).parent
    possible_venv_paths =[
        os.path.join(current_dir,".venv","Scripts","python.exe"),
        os.path.join(current_dir,"venv","Scripts","python.exe"),
        os.path.join(current_dir,".venv","bin","python"),
        os.path.join(current_dir,"venv","bin","python"),
]


for path in possible_venv_paths:
if os.path.exists(path):
return path
return sys.executable


def run_research(topic):
"""使用給定主題運(yùn)行 main.py 并返回結(jié)果"""
    current_dir =Path(__file__).parent
    python_executable = find_venv_python()


# 準(zhǔn)備帶 UTF-8 編碼的環(huán)境
    env = os.environ.copy()
    env['PYTHONIOENCODING']='utf-8'
    env['PYTHONLEGACYWINDOWSSTDIO']='1'


try:
# 作為子進(jìn)程運(yùn)行 main.py
        process = subprocess.Popen(
[python_executable,"main.py"],
            cwd=current_dir,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding='utf-8',
            errors='replace',
            env=env
)


# 將主題作為輸入發(fā)送
        stdout, stderr = process.communicate(input=topic +"\n", timeout=300)


if process.returncode ==0:
# 從 stdout 中提取最終結(jié)果
return extract_final_result(stdout),None
else:
returnNone, f"錯(cuò)誤（返回代碼 {process.returncode}）:\n{stderr}"


except subprocess.TimeoutExpired:
        process.kill()
returnNone,"研究在 5 分鐘后超時(shí)"
exceptExceptionas e:
returnNone, f"意外錯(cuò)誤: {str(e)}"


def extract_final_result(output):
"""從 main.py CrewAI 輸出中提取最終結(jié)果"""
    lines = output.split('\n')


# 首先，嘗試找到最終結(jié)果部分
    final_result_start =-1
for i, line in enumerate(lines):
if"FINAL RESULT:"in line or"==================================================\nFINAL RESULT:"in output:
            final_result_start = i
break


if final_result_start !=-1:
# 提取 “FINAL RESULT:” 之后直到結(jié)束的內(nèi)容
        result_lines =[]
for line in lines[final_result_start:]:
# 跳過 “FINAL RESULT:” 行本身
if"FINAL RESULT:"in line:
# 如果同一行存在標(biāo)記后的內(nèi)容，則獲取
                content_after = line.split("FINAL RESULT:",1)
if len(content_after)>1and content_after[1].strip():
                    result_lines.append(content_after[1].strip())
continue


# 跳過 CrewAI 格式化和空行
            cleaned_line = re.sub(r'[╭│╰═─└├┤┬┴┼╔╗╚╝║╠╣╦╩╬▓??]','', line)
            cleaned_line = cleaned_line.strip()


if cleaned_line:
                result_lines.append(cleaned_line)


if result_lines:
return'\n'.join(result_lines).strip()


# 第二次嘗試：尋找 ## Final Answer 模式
    final_answer_lines =[]
    capturing =False


for line in lines:
if"## Final Answer"in line or"Final Answer:"in line:
            capturing =True
# 如果標(biāo)記后有內(nèi)容，則包含
if"Final Answer:"in line:
                content = line.split("Final Answer:",1)
if len(content)>1and content[1].strip():
                    final_answer_lines.append(content[1].strip())
continue


if capturing:
# 跳過 CrewAI 框圖字符和進(jìn)度指示器
            cleaned = re.sub(r'[╭│╰═─└├┤┬┴┼╔╗╚╝║╠╣╦╩╬▓?????????]','', line)
            cleaned = cleaned.strip()


# 在某些模式下停止，表示答案結(jié)束
if any(pattern in line.lower()for pattern in[
'crew execution completed','task completion','crew completion',
'└──','assigned to:','status:','used'
]):
break


# 僅包含實(shí)質(zhì)性內(nèi)容
if cleaned and len(cleaned)>10:
                final_answer_lines.append(cleaned)


if final_answer_lines:
return'\n'.join(final_answer_lines).strip()


# 第三次嘗試：在 crew 完成消息前獲取最后一段實(shí)質(zhì)性內(nèi)容
    substantial_blocks =[]
    current_block =[]


for line in lines:
# 跳過明顯的 CrewAI UI 元素
if any(skip in line for skip in['╭','│','╰','??','??','└──','Assigned to:','Status:']):
if current_block:
                substantial_blocks.append('\n'.join(current_block))
                current_block =[]
continue


        cleaned = line.strip()
if cleaned and len(cleaned)>30:# 僅實(shí)質(zhì)性行
            current_block.append(cleaned)
elif current_block:# 空行結(jié)束一個(gè)塊
            substantial_blocks.append('\n'.join(current_block))
            current_block =[]


# 添加最后一個(gè)塊
if current_block:
        substantial_blocks.append('\n'.join(current_block))


# 返回最后一個(gè)實(shí)質(zhì)性塊（很可能是最終答案）
if substantial_blocks:
return substantial_blocks[-1].strip()


return"研究成功完成。請(qǐng)檢查控制臺(tái)輸出以獲取詳細(xì)結(jié)果。"


def get_latest_images():
"""從 images 文件夾獲取最新圖像"""
    images_dir =Path("servers/images")
ifnot images_dir.exists():
return[]


# 獲取所有圖像文件
    image_extensions =['*.jpg','*.jpeg','*.png','*.gif','*.bmp']
    image_files =[]


for ext in image_extensions:
        image_files.extend(glob.glob(str(images_dir / ext)))


ifnot image_files:
return[]


# 按修改時(shí)間排序（最新優(yōu)先）
    image_files.sort(key=os.path.getmtime, reverse=True)


# 返回前 5 個(gè)最新圖像
return image_files[:1]


def main():
    st.set_page_config(
        page_title="CrewAI-MCP 研究助手",
        page_icnotallow="??",
        layout="wide"
)


    st.title("?? CrewAI-MCP 學(xué)習(xí)助手")
    st.markdown("輸入一個(gè)主題以進(jìn)行研究并生成帶視覺圖表的全面見解。")


# 主題輸入
    topic = st.text_input(
"研究主題:",
        placeholder="例如，解釋光合作用過程、機(jī)器學(xué)習(xí)算法等。",
        help="輸入您想要詳細(xì)研究的任何主題"
)


# 研究按鈕
if st.button("?? 開始研究", type="primary", disabled=not topic.strip()):
if topic.strip():
with st.spinner(f"?? 正在研究 '{topic}'... 這可能需要幾分鐘。"):
                result, error = run_research(topic.strip())
print(f"來自 CREWAI 的結(jié)果: {result}")


if result:
                st.success("? 研究成功完成！")
print(f"來自 CREWAI 的結(jié)果: {result}")
# 將結(jié)果存儲(chǔ)在會(huì)話狀態(tài)中
                st.session_state['research_result']= result
                st.session_state['research_topic']= topic.strip()
                st.session_state['latest_images']= get_latest_images()
else:
                st.error(f"? 研究失敗: {error}")


# 并排顯示結(jié)果和圖像
if'research_result'in st.session_state:
# 創(chuàng)建分隔線
        st.divider()
        st.subheader(f"研究結(jié)果: {st.session_state.get('research_topic', '未知主題')}")


# 創(chuàng)建兩列以并排顯示
        col1, col2 = st.columns([2,1])# 結(jié)果占 2/3 寬度，圖像占 1/3 寬度


# 左側(cè)列 - 研究結(jié)果
with col1:
            st.markdown("### ?? 摘要結(jié)果")


# 以 markdown 格式顯示結(jié)果
            result_text = st.session_state['research_result']
            pattern = re.compile(r'\x1b\[[\d;]*m')
            result_text = pattern.sub('', result_text)


# 為長內(nèi)容創(chuàng)建可滾動(dòng)容器
with st.container():
                st.markdown(result_text)


# 為結(jié)果添加下載按鈕
            st.download_button(
                label="?? 下載結(jié)果為文本",
                data=result_text,
                file_name=f"research_{st.session_state.get('research_topic', 'topic').replace(' ', '_')}.txt",
                mime="text/plain"
)


# 右側(cè)列 - 生成的圖像
with col2:
            st.markdown("### ?? 生成的圖像")


            images = st.session_state.get('latest_images',[])


if images:
                st.success(f"找到 {len(images)} 張圖像")


# 垂直堆疊顯示圖像
for idx, image_path in enumerate(images):
try:
# 打開并顯示圖像
                        img =Image.open(image_path)


                        st.image(
                            img,
                            captinotallow=f"生成: {Path(image_path).name}",
                            use_container_width=True
)


# 為每張圖像添加下載按鈕
with open(image_path,"rb")as file:
                            st.download_button(
                                label=f"?? 下載",
                                data=file.read(),
                                file_name=Path(image_path).name,
                                mime="image/jpeg",
                                key=f"download_img_{idx}"
)


# 如果有多張圖像，添加圖像間距
if idx < len(images)-1:
                            st.markdown("---")


exceptExceptionas e:
                        st.error(f"加載圖像出錯(cuò): {str(e)}")
else:
                st.info("??? 研究完成后，圖像將顯示在此處。")
with st.expander("?? 關(guān)于圖像"):
                    st.markdown("""
**工作原理：**
-研究期間自動(dòng)生成圖像
-保存到`servers/images/`文件夾
-按創(chuàng)建時(shí)間排序顯示在此處
-每張圖像都有下載按鈕
""")


if __name__ =="__main__":
    main()