Multi-Agent/server.py

"""FastAPI 后端 — SSE 实时推送 Agent 执行 trace"""
import json, asyncio
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel
from config import APP_PORT
from orchestrator import run_supervisor, run_pipeline, run_debate

app = FastAPI(title="Multi-Agent Demo")


class ExecuteRequest(BaseModel):
    mode: str  # supervisor | pipeline | debate
    query: str


MODE_RUNNERS = {
    "supervisor": run_supervisor,
    "pipeline": run_pipeline,
    "debate": run_debate,
}

PRESETS = {
    "supervisor": {
        "title": "Agent 面试题调研 + 数据分析",
        "query": "搜索 2025 年最新的 Agent 开发面试题，重点找 RAG、多 Agent 协同、上下文工程相关的题目，统计各方向的题目数量占比",
    },
    "pipeline": {
        "title": "面试题竞品分析报告",
        "query": "搜索 LangChain、CrewAI、AutoGen 三个多 Agent 框架的核心差异、社区活跃度、适用场景，生成一份结构化的对比分析报告",
    },
    "debate": {
        "title": "技术选型辩论",
        "query": "Agent 开发应该用 LangGraph 还是从零手撸？LangGraph 框架 vs 纯代码手写，哪种更适合生产环境？",
    },
}


@app.get("/")
async def index():
    return FileResponse("index.html")


@app.get("/api/presets")
async def get_presets():
    return PRESETS


@app.post("/api/execute")
async def execute(req: ExecuteRequest):
    runner = MODE_RUNNERS.get(req.mode)
    if not runner:
        return {"error": f"Unknown mode: {req.mode}"}

    async def stream():
        queue = asyncio.Queue()

        def emit(event):
            queue.put_nowait(event)

        # 在线程中运行（LLM 调用是阻塞的）
        import threading
        result_holder = {}

        def run():
            try:
                result = runner(req.query, emit=emit)
                result_holder["result"] = result
            except Exception as e:
                result_holder["error"] = str(e)
            finally:
                queue.put_nowait(None)  # 结束信号

        thread = threading.Thread(target=run, daemon=True)
        thread.start()

        # 流式输出事件
        while True:
            event = await queue.get()
            if event is None:
                break
            yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"

        # 发送最终结果
        if "error" in result_holder:
            yield f"data: {json.dumps({'type': 'error', 'message': result_holder['error']}, ensure_ascii=False)}\n\n"
        elif "result" in result_holder:
            yield f"data: {json.dumps({'type': 'final', **result_holder['result']}, ensure_ascii=False)}\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=APP_PORT)