如何用 MLflow Agent Server 把 AI 代理托管为生产 REST API【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow如果你已经用openai-agents-sdk或其他框架写好了一个 AI 代理想把它暴露成一个可以被 HTTP 请求调用的服务而不是只能在脚本里跑MLflow 的 Agent Server 可以直接完成这件事。它是一个基于 FastAPI 的服务器把代理函数注册在POST /invocations端点上并自动处理请求/响应校验、MLflow tracing 记录以及可选的流式响应。适用前提需要安装openai-agents-sdk和mlflow3.0系列中的 3.6.0 及以上版本文档给出的安装命令是mlflow3.6.0代理底层调用 OpenAI 时需要设置OPENAI_API_KEY服务端代码必须导入你的 agent 模块invoke注册的函数才会被服务器发现。安装依赖并准备环境pip install -U openai-agents mlflow3.6.0 export OPENAI_API_KEYsk-...OPENAI_API_KEY替换为你自己的密钥。定义带invoke装饰器的代理函数创建一个agent.py在其中定义代理并把处理请求的函数用invoke()标注。函数签名要使用ResponsesAgentRequest/ResponsesAgentResponse这两个类型服务器会据此自动做输入输出校验from agents import Agent, Runner from mlflow.genai.agent_server import invoke, stream from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentResponse agent Agent( nameMath Tutor, instructionsYou provide help with math problems. Explain your reasoning and include examples, ) invoke() async def non_streaming(request: ResponsesAgentRequest) - ResponsesAgentResponse: msgs [i.model_dump() for i in request.input] result await Runner.run(agent, msgs) return ResponsesAgentResponse(output[item.to_input_item() for item in result.new_items]) # 可选再注册一个 stream 函数即可支持流式响应两点限制需要注意invoke装饰器只能使用一次重复注册会抛出ValueError只有注册了stream函数后请求体里的stream: true才会生效否则流式请求没有处理函数可用。编写服务器入口start_server.py# 必须导入 agent 模块invoke 注册的函数才能被服务器发现 import agent # noqa: F401 from mlflow.genai.agent_server import ( AgentServer, setup_mlflow_git_based_version_tracking, ) agent_server AgentServer(ResponsesAgent) app agent_server.app # 可选开启基于 git 的版本跟踪 # 让每次 trace 对应到一个具体的 git commit setup_mlflow_git_based_version_tracking() def main(): # 支持多 worker 时以 import string 的形式传入 app agent_server.run(app_import_stringstart_server:app) if __name__ __main__: main()AgentServer(ResponsesAgent)中的类型参数开启自动的输入/输出校验和流式 tracing 聚合目前只支持ResponsesAgent。setup_mlflow_git_based_version_tracking()是可选的它会把活跃模型名与当前 git commit 短哈希关联本地开发时 app 名为local部署为 Databricks App 时取DATABRICKS_APP_NAME环境变量。agent_server.run(app_import_stringstart_server:app)以模块内app对象作为 import string 传给 uvicorn这是多 worker 部署的方式。启动服务器python3 start_server.py --reload--reload会在代码变更时自动重载服务器适合开发阶段。生产环境常用的参数来自同一启动入口的命令行参数解析# 传入 worker 数以支持多个并发请求默认 1 python3 start_server.py --workers 4 # 指定端口默认 8000 python3 start_server.py --port 8000服务器默认监听0.0.0.0因此局域网内或其他容器可通过主机地址访问。验证服务是否可用启动成功后向/invocations端点发送一个 JSON 请求测试curl -X POST http://localhost:8000/invocations \ -H Content-Type: application/json \ -d { input: [{ role: user, content: What is the 14th Fibonacci number?}]}请求成功时服务器返回代理的输出ResponsesAgentResponse结构。如果请求体不是合法 JSON 或不满足ResponsesAgent的校验会收到 400代理执行抛异常时会收到 500 及错误信息。除了/invocations服务器还暴露了以下端点可用于健康检查和部署监控GET /health返回{status: healthy}GET /agent/info返回 app 名、use_case: agent、mlflow 版本ResponsesAgent 类型下还包含agent_api: responsesPOST /responses与/invocations同源的端点用于兼容 OpenAI 客户端的client.responses.create(...)调用方式。如果注册了stream函数可以在请求体中加入stream: true发送流式请求响应以data: json分块推送并以data: [DONE]结束curl -X POST http://localhost:8000/invocations \ -H Content-Type: application/json \ -d { input: [{ role: user, content: What is the 14th Fibonacci number?}], stream: true }查看自动记录的 TracesAgent Server 会自动为每次调用创建 MLflow trace请求作为输入、响应作为输出写入对应 span。测试你的代理后打开 MLflow UI点击 Traces 标签页即可看到这些调用记录。若开启了setup_mlflow_git_based_version_tracking()trace 会关联到对应的 git commit 版本方便回溯是哪个代码版本产生了某次调用。可选用mlflow.genai.evaluate评估代理测试通过后可以用同一套invoke注册的函数做离线评估而不用真正启动服务器。创建一个eval_agent.pyimport asyncio import mlflow # 导入 agent 模块invoke 注册的函数才能被找到 from agent import agent # noqa: F401 from mlflow.genai.agent_server import get_invoke_function from mlflow.genai.scorers import RelevanceToQuery, Safety from mlflow.types.responses import ResponsesAgentRequest, ResponsesAgentResponse eval_dataset [ { inputs: { request: {input: [{role: user, content: Whats the 15th Fibonacci number}]} }, expected_response: The 15th Fibonacci number is 610., } ] def sync_invoke_fn(request: dict) - ResponsesAgentResponse: # 获取通过 invoke 装饰器注册的函数 invoke_fn get_invoke_function() return asyncio.run(invoke_fn(ResponsesAgentRequest(**request))) mlflow.genai.evaluate( dataeval_dataset, predict_fnsync_invoke_fn, scorers[RelevanceToQuery(), Safety()], )运行python3 eval_agent.py控制台会输出评估结果和 MLflow run 信息在 MLflow UI 的实验页面可以找到对应的 run点击 run 名称查看聚合的指标和元数据。限制说明服务器类型目前仅支持ResponsesAgent不传agent_type时不做输入/输出校验也不做流式 tracing 聚合。invoke与stream各只能注册一个函数。--reload依赖热重载适合开发生产部署应使用--workers控制并发。AgentServer还支持enable_chat_proxyTrue参数把未匹配的请求代理到CHAT_APP_PORT默认 3000端口上的 chat 应用只放行/、/favicon.ico、/ping、/assets/*、/api/*、/chat/*等路径以防 SSRF这是搭配前端聊天应用部署时的可选项不是本场景必需步骤。完整示例与参数说明可对照仓库中的 Agent Server 文档端点与命令行参数的实现见 server.py。【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考