""" app/api.py - FastAPI 应用主文件 """ import logging from typing import List from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from app.agents import orchestrate_agents from app.config import get_settings # 初始化日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 初始化设置 settings = get_settings() # 创建 FastAPI 应用 app = FastAPI( title=settings.project_name, version=settings.project_version, debug=settings.fastapi_debug, ) # 添加 CORS 中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class RequirementRequest(BaseModel): """需求分析请求模型""" requirement: str class RequirementAnalysisResponse(BaseModel): """需求分析响应""" functional_requirements: List[str] non_functional_requirements: List[str] acceptance_criteria: List[str] edge_cases: List[str] summary: str class TestCase(BaseModel): """测试用例""" test_id: str test_name: str precondition: str steps: List[str] expected_result: str test_type: str class TestCaseResponse(BaseModel): """测试用例响应""" test_cases: List[TestCase] test_strategy: str coverage_plan: str class CodeGenerationResponse(BaseModel): """代码生成响应""" java_code: str unit_tests: str implementation_notes: str class FullWorkflowResponse(BaseModel): """完整工作流响应""" requirement_analysis: RequirementAnalysisResponse test_cases: TestCaseResponse code_generation: CodeGenerationResponse status: str = "success" # 路由 @app.post("/workflow/full", response_model=FullWorkflowResponse) async def full_workflow(request: RequirementRequest): """ 完整工作流:PM Agent -> QA Agent -> Dev Agent 这个端点接收一个简单的需求描述,通过三个Agent的协作, 最终生成完整的需求分析、测试用例和Java代码。 Args: request: RequirementRequest 包含简单的需求描述 Returns: FullWorkflowResponse 包含: - requirement_analysis: PM Agent的需求分析结果 - test_cases: QA Agent生成的测试用例 - code_generation: Dev Agent生成的Java代码和单元测试 Example: { "requirement": "创建一个用户管理系统,支持用户注册、登录、个人信息管理" } """ try: logger.info(f"开始处理需求: {request.requirement}") # 调用编排函数,执行三个Agent的工作流 result = await orchestrate_agents(request.requirement) logger.info("需求处理完成,正在组装响应") # 构建响应 response = FullWorkflowResponse( requirement_analysis=result["requirement_analysis"], test_cases=result["test_cases"], code_generation=result["code_generation"], status="success" ) logger.info("响应组装完成") return response except Exception as e: logger.error(f"处理需求时出错: {str(e)}", exc_info=True) return { "error": str(e), "status": "error" } @app.get("/") async def root(): """根端点""" return { "message": "欢迎使用 AI Agent API", "endpoints": { "health": "/health", "chat": "/chat", "workflow": "/workflow/full", "docs": "/docs", } }