Agent工程化实战：从原型到生产级多智能体系统的构建与部署

一、为什么Agent工程化如此重要？

2025年，大语言模型（LLM）的推理能力已突破实用门槛，基于LLM的智能体（Agent）从Demo走向真实业务场景。然而，原型与生产之间隔着一条巨大的工程化鸿沟：

可靠性：LLM输出不确定，如何保证业务流程的稳定性？可观测性：Agent的每一次工具调用、每一步推理都可能是黑盒，如何追踪和调试？扩展性：单Agent能力有限，如何设计多智能体协作并动态伸缩？性能：LLM响应延迟高，如何异步化、缓存、流式处理？运维：模型版本更新、API限流、成本控制，如何自动化管理？

本文将围绕上述痛点，从零构建一个生产级多智能体系统，涵盖设计、编码、测试、容器化、部署与监控全流程。

---

二、系统架构设计

我们设计一个智能客服 代码辅助的双Agent协作系统，包含以下角色：

Agent	职责	工具
Planner	分析用户问题，拆解任务，调度执行	任务分解、资源检索
Executor	执行具体动作，如查询数据库、调用API、生成代码	SQL执行器、代码解释器、向量检索
Reviewer	校验执行结果，必要时回退或重试	结果校验、反思

整体架构如下：

代码语言：ja vascript

复制

用户请求 → FastAPI网关 → 任务队列(Celery) → Agent Orchestrator ├─ Planner Agent (LLM) ├─ Executor Agent (LLM Tools) └─ Reviewer Agent (LLM)↓状态存储(Redis) 日志(ELK) 监控(Prometheus)

关键设计决策：

异步解耦：通过Celery将长耗时Agent推理任务与HTTP请求分离，避免超时和资源阻塞。状态管理：使用Redis维护会话上下文和Agent执行状态，支持断点续传。容错机制：每个Agent步骤增加超时、重试和降级逻辑（如LLM不可用时返回缓存或默认答案）。可观测性：集成OpenTelemetry，全链路追踪Agent的推理链和工具调用。

---

三、核心实现：基于LangChain的自定义Agent

我们使用LangChain 0.3 （已适配v1.0 API）构建自定义Agent，并接入OpenAI兼容模型（支持GPT-4o、Claude、Qwen等）。

3.1 自定义工具

代码语言：ja vascript

复制

# tools.pyfrom langchain.tools import BaseToolfrom pydantic import BaseModel, Fieldimport subprocessimport sqlite3import jsonclass SQLQueryInput(BaseModel):query: str = Field(description="SQL查询语句")class SQLExecutorTool(BaseTool):name: str = "sql_executor"description: str = "执行SQL查询并返回结果，仅限SELECT语句"args_schema: type[BaseModel] = SQLQueryInputdef _run(self, query: str) -> str:# 实际生产环境应使用连接池和只读账户conn = sqlite3.connect("data.db")try:cursor = conn.cursor()cursor.execute(query)rows = cursor.fetchall()columns = [desc[0] for desc in cursor.description]return json.dumps([dict(zip(columns, row)) for row in rows], ensure_ascii=False)except Exception as e:return f"SQL执行错误: {str(e)}"finally:conn.close()class CodeInterpreterTool(BaseTool):name: str = "python_interpreter"description: str = "执行Python代码并返回标准输出，用于数学计算或数据处理"def _run(self, code: str) -> str:try:# 使用subprocess隔离执行，限制资源result = subprocess.run(["python3", "-c", code],capture_output=True,text=True,timeout=10,env={"PYTHONPATH": ""})return result.stdout or result.stderrexcept subprocess.TimeoutExpired:return "代码执行超时（10秒）"except Exception as e:return f"执行异常: {str(e)}"

3.2 Agent定义与提示工程

每个Agent拥有独立的系统提示和工具集，我们采用ReAct风格提示，并加入结构化输出要求。

代码语言：ja vascript

复制

# agents.pyfrom langchain_openai import ChatOpenAIfrom langchain.agents import create_react_agent, AgentExecutorfrom langchain.prompts import PromptTemplatefrom tools import SQLExecutorTool, CodeInterpreterTool# 基础LLM（支持环境变量切换模型）llm = ChatOpenAI(model=os.getenv("LLM_MODEL", "gpt-4o-mini"),temperature=0.1,timeout=60,max_retries=3,)planner_prompt = PromptTemplate.from_template("""你是一个任务规划专家。根据用户输入，拆解为可执行的子任务，并决定由哪个Agent执行。用户输入: {input}已有上下文: {context}可用工具: {tools}请按JSON格式输出计划:{{"plan": [{{"step": 1, "agent": "executor", "action": "sql_query", "params": {{"query": "..."}}}},{{"step": 2, "agent": "reviewer", "action": "validate", "params": {{"expected": "..."}}}}]}}""")executor_prompt = PromptTemplate.from_template("""你是一个执行专家，负责调用工具完成具体任务。任务描述: {task}可用工具: {tools}请逐步思考并调用工具，最后给出执行结果。""")def create_planner_agent():tools = []# Planner本身不直接调用工具，而是输出计划# 但为了统一，我们使用工具调用方式，这里特殊处理# 实际使用中，可让Planner直接调用"任务分解"工具，但为了演示，我们直接解析JSONpass# 更实用的方式：使用langchain的structured outputfrom langchain.output_parsers import PydanticOutputParserfrom pydantic import BaseModel, Fieldfrom typing import Listclass PlanStep(BaseModel):step: intagent: straction: strparams: dictclass Plan(BaseModel):plan: List[PlanStep]parser = PydanticOutputParser(pydantic_object=Plan)planner_llm = llm.with_structured_output(Plan)# 注意：with_structured_output需要模型支持JSON模式，否则使用function calling# 实际执行Agentdef create_executor_agent():tools = [SQLExecutorTool(), CodeInterpreterTool()]agent = create_react_agent(llm, tools, executor_prompt)return AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)executor = create_executor_agent()

3.3 多智能体编排器（Orchestrator）

编排器负责管理对话状态、调用各Agent、处理结果汇总。

代码语言：ja vascript

复制

# orchestrator.pyimport jsonimport asynciofrom typing import Dict, Anyfrom agents import planner_llm, executor, reviewer_llm# reviewer类似定义，省略from redis import Redisimport celeryredis_client = Redis(host='redis', decode_responses=True)class AgentOrchestrator:def __init__(self, session_id: str):self.session_id = session_idself.context = self._load_context()def _load_context(self) -> Dict:key = f"session:{self.session_id}:context"data = redis_client.get(key)return json.loads(data) if data else {"history": [], "last_plan": None}def _sa ve_context(self):key = f"session:{self.session_id}:context"redis_client.setex(key, 3600, json.dumps(self.context))async def process_request(self, user_input: str) -> str:# 1. 规划plan_prompt = f"用户输入: {user_input}上下文: {self.context['history'][-5:]}"try:plan = await planner_llm.ainvoke(plan_prompt)except Exception as e:# 降级：直接执行默认动作return self._fallback(user_input)self.context['last_plan'] = plan.dict()self._sa ve_context()# 2. 按计划执行（可并行执行独立步骤）results = []for step in plan.plan:if step.agent == "executor":result = await self._run_executor(step.action, step.params)elif step.agent == "reviewer":result = await self._run_reviewer(step.action, step.params, previous=results)# 其他agent...results.append({"step": step.step, "result": result})# 3. 汇总生成最终回复final = await self._generate_final_response(user_input, results)self.context['history'].append({"user": user_input, "assistant": final})self._sa ve_context()return finalasync def _run_executor(self, action: str, params: dict) -> str:# 映射action到具体工具调用if action == "sql_query":tool = SQLExecutorTool()return tool._run(params.get("query", ""))elif action == "python_code":tool = CodeInterpreterTool()return tool._run(params.get("code", ""))else:# 使用AgentExecutor通用处理response = await executor.ainvoke({"input": f"执行{action}，参数{params}"})return response["output"]# reviewer和最终回复方法类似，使用LLM调用

---

四、工程化关键：异步任务与API层

为避免HTTP请求阻塞，我们将Agent执行放入Celery异步任务，并支持WebSocket流式返回。

4.1 Celery任务定义

代码语言：ja vascript

复制

# tasks.pyfrom celery import Celeryfrom orchestrator import AgentOrchestratorapp = Celery('agent_tasks', broker='redis://redis:6379/0')app.conf.update(task_serializer='json',accept_content=['json'],result_serializer='json',timezone='Asia/Shanghai',task_track_started=True,task_time_limit=300,# 5分钟超时task_soft_time_limit=240,task_acks_late=True,# 防止任务丢失)@app.task(bind=True, max_retries=3)def process_agent_request(self, session_id: str, user_input: str):try:orch = AgentOrchestrator(session_id)# 异步执行loop = asyncio.new_event_loop()asyncio.set_event_loop(loop)result = loop.run_until_complete(orch.process_request(user_input))loop.close()return resultexcept Exception as e:# 重试机制self.retry(exc=e, countdown=2 ** self.request.retries)

4.2 FastAPI接口（支持同步轮询和WebSocket流式）

代码语言：ja vascript

复制

# api.pyfrom fastapi import FastAPI, WebSocket, BackgroundTasksfrom pydantic import BaseModelfrom tasks import process_agent_requestfrom celery.result import AsyncResultimport jsonapp = FastAPI(title="Multi-Agent System")class RequestPayload(BaseModel):session_id: strmessage: str# 同步轮询接口（适用于短任务）@app.post("/agent/sync")async def sync_agent(payload: RequestPayload):task = process_agent_request.delay(payload.session_id, payload.message)# 阻塞等待（生产环境不建议，这里仅演示）result = task.get(timeout=60)return {"status": "success", "data": result}# 异步提交，轮询结果@app.post("/agent/async")async def async_agent(payload: RequestPayload):task = process_agent_request.delay(payload.session_id, payload.message)return {"task_id": task.id, "status_url": f"/agent/status/{task.id}"}@app.get("/agent/status/{task_id}")async def get_status(task_id: str):task = AsyncResult(task_id)if task.ready():return {"status": "completed", "result": task.result}elif task.failed():return {"status": "failed", "error": str(task.info)}else:return {"status": "pending", "progress": task.info.get('progress', 0)}# WebSocket流式返回（支持推理过程实时展示）@app.websocket("/ws/agent")async def websocket_agent(websocket: WebSocket):await websocket.accept()try:data = await websocket.receive_json()session_id = data.get("session_id")message = data.get("message")# 使用Celery的异步任务，但通过Redis pub/sub推送中间结果# 这里简化：直接调用orchestrator并流式生成from orchestrator import AgentOrchestratororch = AgentOrchestrator(session_id)# 模拟流式输出（实际可用async生成器）async for chunk in orch.stream_process(message):await websocket.send_text(json.dumps({"type": "chunk", "data": chunk}))await websocket.send_text(json.dumps({"type": "end"}))except Exception as e:await websocket.send_text(json.dumps({"type": "error", "data": str(e)}))finally:await websocket.close()

---

五、可观测性：全链路追踪与日志

使用OpenTelemetry Jaeger实现分布式追踪，每条请求生成唯一trace_id，贯穿网关、Celery、LLM调用、工具执行。

代码语言：ja vascript

复制

# tracing.pyfrom opentelemetry import tracefrom opentelemetry.exporter.jaeger.thrift import JaegerExporterfrom opentelemetry.instrumentation.fastapi import FastAPIInstrumentorfrom opentelemetry.instrumentation.celery import CeleryInstrumentorfrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.sdk.trace.export import BatchSpanProcessordef setup_tracing(app):provider = TracerProvider()processor = BatchSpanProcessor(JaegerExporter(agent_host_name=os.getenv("JAEGER_HOST", "jaeger"),agent_port=6831,))provider.add_span_processor(processor)trace.set_tracer_provider(provider)FastAPIInstrumentor.instrument_app(app)CeleryInstrumentor().instrument()

日志结构化：使用structlog，每个日志条目包含trace_id、session_id、agent_name等。

代码语言：ja vascript

import structloglogger = structlog.get_logger()# 在orchestrator中logger.info("agent_plan", trace_id=current_trace_id, plan=plan.dict())

监控层面可接入 Prometheus，持续采集请求总量、响应延迟、LLM 调用次数以及工具调用成功率等关键指标，用于观察编排链路的运行状态。

代码语言：ja vascript

from prometheus_client import Counter, Histogram, start_http_serverREQUEST_COUNT = Counter('agent_requests_total', 'Total requests')LLM_CALL_HISTOGRAM = Histogram('llm_call_duration_seconds', 'LLM call latency')# 使用装饰器进行统计

六、容器化与Kubernetes部署

6.1 Dockerfile（多阶段构建）

代码语言：ja vascript

复制

# DockerfileFROM python:3.11-slim as builderWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtFROM python:3.11-slimWORKDIR /appCOPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packagesCOPY . .ENV PYTHONPATH=/appEXPOSE 8000CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

6.2 Kubernetes资源编排（Deployment Service ConfigMap）

代码语言：ja vascript

复制

# deployment.yamlapiVersion: apps/v1kind: Deploymentmetadata:name: agent-apispec:replicas: 3selector:matchLabels:app: agent-apitemplate:metadata:labels:app: agent-apispec:containers:- name: apiimage: agent-system:latestports:- containerPort: 8000env:- name: REDIS_URLvalueFrom:configMapKeyRef:name: agent-configkey: redis_url- name: OPENAI_API_KEYvalueFrom:secretKeyRef:name: openai-secretkey: api_keyresources:limits:cpu: "1000m"memory: "2Gi"requests:cpu: "500m"memory: "1Gi"livenessProbe:httpGet:path: /healthport: 8000initialDelaySeconds: 30periodSeconds: 10readinessProbe:httpGet:path: /readyport: 8000---apiVersion: v1kind: Servicemetadata:name: agent-api-svcspec:selector:app: agent-apiports:- port: 80targetPort: 8000type: LoadBalancer

水平自动伸缩（HPA）基于CPU和自定义指标（如队列长度）：

代码语言：ja vascript

复制

apiVersion: autoscaling/v2kind: HorizontalPodAutoscalermetadata:name: agent-api-hpaspec:scaleTargetRef:apiVersion: apps/v1kind: Deploymentname: agent-apiminReplicas: 2maxReplicas: 10metrics:- type: Resourceresource:name: cputarget:type: Utilizationa verageUtilization: 60- type: Podspods:metric:name: celery_queue_lengthtarget:type: A verageValuea verageValue: 100

---

七、性能优化与成本控制实践

7.1 缓存语义相似查询

使用向量数据库（如Milvus）缓存常见问答对，LLM调用前先检索，命中则直接返回，降低延迟和费用。

7.2 模型降级策略

当主模型（GPT-4o）限流或超时时，自动切换至轻量级模型（如GPT-3.5-turbo），并告警。

代码语言：ja vascript

复制

class ModelRouter:def __init__(self):self.primary = ChatOpenAI(model="gpt-4o", timeout=10)self.fallback = ChatOpenAI(model="gpt-3.5-turbo", timeout=5)async def ainvoke_with_fallback(self, prompt):try:return await self.primary.ainvoke(prompt)except Exception:logger.warning("primary model failed, using fallback")return await self.fallback.ainvoke(prompt)

7.3 流式响应与首字延迟优化

对于WebSocket，首字延迟是关键。我们采用astream_events逐token返回，减少用户等待感。

代码语言：ja vascript

复制

async def stream_process(self, user_input):async for event in self.llm.astream_events(user_input, version="v1"):if event["event"] == "on_chat_model_stream":yield event["data"]["chunk"].content

---

八、测试与质量保障

8.1 单元测试（pytest mocking）

代码语言：ja vascript

复制

# test_agent.pyimport pytestfrom unittest.mock import patchfrom orchestrator import AgentOrchestrator@pytest.mark.asyncioasync def test_planner_parsing():orch = AgentOrchestrator("test_sess")with patch.object(orch, '_run_executor') as mock_exec:mock_exec.return_value = "mocked result"result = await orch.process_request("查询总销售额")assert "总销售额" in result

8.2 集成测试与混沌工程

在预发布环境模拟LLM超时、Redis断连、磁盘满等故障，验证降级和恢复能力。

8.3 A/B测试与灰度发布

通过Istio实现流量切分，新版本Agent只接收5%流量，对比成功率和用户满意度。

---

九、总结与展望

本文从零构建了一个生产级多智能体系统，覆盖了工程化的核心维度：异步解耦、状态管理、可观测性、容器编排、性能优化。在实战中，我们深刻体会到：

Agent不是单点，而是流程：成功的关键在于清晰的编排、明确的工具契约和鲁棒的错误处理。可观测性必须内建：无法调试的Agent系统是无法运维的，OpenTelemetry 结构化日志是标配。成本与性能需动态平衡：缓存、模型路由、自适应超时等机制必不可少。

未来演进方向：

自进化Agent：基于历史执行数据微调小型模型，减少对大型模型的依赖。多模态Agent：结合视觉、语音，实现更丰富的交互。联邦多智能体：跨组织协作，保护数据隐私。

Agent工程化才刚刚开始，期待更多开发者将智能化能力融入业务，创造真正的价值。