# 代码阅读顺序(编号表示调用阶段,不是源文件从上往下的行号) # 启动 S1:各进程导入 config.py / create_logger.py,准备配置和日志。 # 启动 S2:先准备数据服务,再分别启动 mcp_server 下的工具服务(8001/8002/8003)。 # 启动 S3:分别启动 a2a_server 下的子助手(天气 5005、票务 5006、差旅 5007)。 # 启动 S4:运行本文件,创建 ChatService 并监听 8080;浏览器打开首页。 # 以上服务各自在独立进程运行;本文件不会自动启动 MCP、A2A 或数据库。 # 一次网页对话的主流程: # 1 网页发送 → 2 API 接收 → 3 记录用户消息 → 4 意图识别 → 5 判断是否规划 # → 6 主助手分派 → 7 A2A 子助手选工具 → 8 MCP 执行查询/模拟预订 # → 9 A2A 返回 → 10 主助手整理并输出片段 → 11 保存完整回复 → 12 结束网页接收。 # 第 5 步需要追问时跳过 6~9,输出追问并保存;子助手在第 7 步也可能追问而不调用工具。 # 复杂任务会重复 6~9;第 10 步的片段会边生成边传到网页。 # 各文件中的“内部步骤”“记忆”“辅助”等编号,只描述该函数或独立脚本的局部顺序。 """ 需求:SmartBizTrip FastAPI后端服务器,提供REST API接口 """ import json import os import sys # 路径解释:__file__ 是当前 api_server.py 的位置,dirname 取其目录,.. 指向上一级。 # 把上一级放入模块搜索路径,便于依赖以 SmartBizTrip.xxx 形式导入同一项目。 # 这只影响 Python 查找模块,不会切换进程工作目录,也不会启动任何服务。 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import uvicorn from fastapi import FastAPI from fastapi.responses import FileResponse, StreamingResponse from pydantic import BaseModel # 【启动 S4.1】导入类时先执行依赖模块的顶层语句;函数体要等调用才执行。 from chat_service import ChatService # app 是 HTTP 应用对象,后面的 @app.get/@app.post 都把“路径→函数”登记到这个对象上。 # 装饰器在模块加载时注册函数;访问对应 URL 时,框架才调用该函数体。 app = FastAPI(title="SmartBizTrip API", description="基于A2A的企业差旅智能体") # 全局服务实例 # 【启动 S4.2】创建一个全局对话服务:注册 A2A 客户端、创建模型对象、尝试恢复记忆。 # 该实例供所有请求共用;当前代码没有按用户或会话隔离记忆。 chat_service = ChatService() # 【启动 S4.3】定义请求结构并注册下面的路由;此时尚未处理用户消息。 # FastAPI 会把请求 JSON 校验为模型对象,再调用对应接口。 class ChatRequest(BaseModel): # 请求示例:{"message": "查询北京到上海的高铁"}。 # 字段名 message 要与网页 JSON.stringify 中的字段名一致。 message: str class ProfileRequest(BaseModel): profile: dict # 默认 False:保留旧偏好,只补充/覆盖提交的键;True:旧偏好整体换成本次 profile。 replace: bool = False @app.get("/") async def index(): """返回前端页面""" # 【启动 S4.5】浏览器 GET / 时返回 static/index.html,页面脚本再加载代理卡片和记忆。 static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") # 返回文件响应对象,由框架读取并发送 HTML;此函数不会在 Python 中执行网页 JavaScript。 # JavaScript 要等浏览器收到页面后运行,随后调用 /api/agents、/api/memory。 return FileResponse(os.path.join(static_dir, "index.html")) @app.post("/api/chat") # FastAPI 找到对应的处理函数 async def chat(request: ChatRequest): """发送消息,获取回复""" # 【非流式入口】POST /api/chat 等待 ChatService.chat() 完整结束后一次性返回 JSON。 # 它与网页默认使用的 /api/chat/stream 是两个入口,一次请求只走其中一个。 # request 是校验后的 ChatRequest 对象,request.message 取出其中的字符串。 # await 在这里等待业务方法给出完整字符串;response 不是 HTTP 响应,而是助手回复文本。 response = await chat_service.chat(request.message) return {"status": "success", "message": response} async def sse_generator(message: str): """SSE 生成器,将业务层产出的文本片段封装成事件。""" # 【主流程 2.2 → 3】开始消费业务层异步生成器;每次得到一个 chunk 才向网页发送一次。 # chunk 可能是一段文字,并不保证一个字;等待业务处理期间也不保证立即有输出。 # 异步生成器可以多次 yield,不能按“await 后只拿一次返回值”的方式消费。 # async for 每拿到一段就执行一次循环体;业务生成器完成后才执行循环后面的 [DONE]。 async for chunk in chat_service.chat_stream(message): # SSE 格式:每行以 "data: " 开头,用空行分隔 # 返回链第 1 层:业务文本 chunk → {"chunk": chunk} → JSON 文本 → SSE 事件。 # json.dumps 会转义文本内部换行,避免它把一条 data 事件意外拆开;ensure_ascii=False 保留中文。 # 末尾两个换行用于分隔事件;yield 把本次事件交给 StreamingResponse 后暂时停在这里。 yield f"data: {json.dumps({'chunk': chunk}, ensure_ascii=False)}\n\n" # 发送结束标记 # 【主流程 12.1】业务生成器正常结束后发送结束标记,再结束 HTTP 响应流。 yield "data: [DONE]\n\n" @app.post("/api/chat/stream") async def chat_stream(request: ChatRequest): """发送消息,流式获取回复(SSE)""" # 【主流程 2.1】接收网页的 {message: ...},把生成器交给 StreamingResponse。 # 这里的 chat_stream 是 HTTP 接口;真正处理业务的是 ChatService.chat_stream。 # 创建生成器并不立即执行其函数体,响应开始迭代它时才进入 sse_generator。 # 这里返回的是“边迭代边发送”的响应对象;还没有把整段回复提前拼好。 # media_type 告诉客户端响应体采用 SSE 文本格式,前端仍需自行解析 data 行。 return StreamingResponse(sse_generator(request.message), media_type="text/event-stream") @app.get("/api/memory") async def get_memory(): """获取记忆状态""" # 【旁路:查看记忆 1 → 2】读取当前共享服务的记忆摘要,返回给网页侧栏。 # 外层 status/data 是本项目约定的 JSON 字段,不是 Python 的函数返回状态。 # 网页通过 response.json() 得到字典对应的对象,再用 data.data 读取记忆内容。 return {"status": "success", "data": chat_service.get_memory_state()} @app.post("/api/memory/clear") async def clear_memory(): """清空记忆""" # 【旁路:清空记忆 1 → 2】网页确认后进入此接口,清理内存并尝试删除数据库记忆。 chat_service.clear_memory() return {"status": "success", "message": "记忆已清空"} @app.post("/api/memory/profile") async def update_profile(request: ProfileRequest): """更新用户偏好""" # 【旁路:保存偏好 1 → 2】replace=True 表示用本次完整表单替换旧偏好。 chat_service.update_user_profile(request.profile, replace=request.replace) return {"status": "success", "message": "用户偏好已更新"} @app.get("/api/agents") async def get_agents(): """获取代理卡片信息""" # 【旁路:代理卡片 1 → 2】读取各代理能力说明,供网页展示。卡片不是工具执行结果。 return {"status": "success", "data": chat_service.get_agent_cards()} if __name__ == "__main__": # 【启动 S4.4】只有直接运行本文件才进入这里,开始监听端口并等待请求。 uvicorn.run(app, host="127.0.0.1", port=8080, log_level="info")