最近在尝试将 AI Agent 应用到实际业务场景时我遇到了一个普遍痛点市面上的开源框架和商业平台要么过于“黑盒”难以深度定制和集成要么过于“原始”需要从零搭建大量基础设施开发效率极低。这种“高不成低不就”的现状促使我萌生了自己动手构建一个 AI Agent 平台的想法。本文将从一个实践者的角度系统性地拆解我构建这个平台的核心动机、技术选型思考以及平台工程的关键设计希望能为同样在探索 AI Agent 落地的开发者提供一份从“为什么”到“怎么做”的实战参考。1. 从概念到现实AI Agent 与平台工程的交汇点在深入平台构建之前我们有必要厘清几个核心概念以及它们如何共同指向了“平台工程”的必要性。1.1 AI Agent 的本质与挑战AI Agent智能体并非一个全新的概念但在大语言模型LLM的加持下其内涵和能力发生了质变。简单来说一个现代的 AI Agent 是一个能够感知环境、进行决策并执行动作以达成目标的智能系统。其核心通常包含几个关键组件大脑Brain通常由 LLM 担任负责理解、规划和推理。记忆Memory用于存储对话历史、工具调用结果、知识片段等分为短期记忆会话上下文和长期记忆向量数据库等。工具Tools赋予 Agent 执行具体任务的能力如调用 API、查询数据库、操作文件等。规划Planning将复杂任务分解为可执行的子任务序列。然而当我们试图将一个简单的 Agent 原型例如用 LangChain 或 LlamaIndex 快速拼装一个推向生产环境时一系列工程挑战便接踵而至开发效率低下每个新 Agent 都需要重复搭建项目结构、配置模型连接、编写工具封装、处理记忆存储。运维复杂度高模型的版本管理、Prompt 的迭代与版本控制、工具 API 的稳定性监控、Agent 运行状态的追踪和日志收集这些都需要大量手工工作。可观测性差当 Agent 执行一个包含多步工具调用的复杂任务时很难清晰地回答“它现在在做什么”“为什么失败了”“哪一步最耗时”缺乏规模化能力如何管理成百上千个不同职能的 Agent如何调度它们的协作如何做统一的权限控制和成本核算1.2 平台工程为 AI Agent 构建“基础设施”这正是平台工程Platform Engineering要解决的问题。平台工程的核心是为内部开发者构建和维护一套标准化的、自助服务的工具链与工作流旨在提升开发者的生产力和体验同时降低系统的复杂性和运维负担。将其映射到 AI Agent 领域一个 AI Agent 平台应致力于解决上述挑战它需要提供一站式开发环境提供从 Agent 创建、工具注册、Prompt 编排、到测试部署的完整流水线。核心能力抽象与复用将模型接入、记忆存储、工具调用、流程编排等能力抽象为平台服务让开发者聚焦业务逻辑。强大的可观测性提供完整的执行链路追踪、详细的日志记录、性能指标监控和成本分析面板。生产级运维支持包括版本管理、灰度发布、弹性伸缩、故障自愈等能力。网络上热议的Harness概念正是一套包裹在 AI Agent 核心推理逻辑之外的基础设施层。它不替代 Agent 的“思考”而是为“思考”提供稳定、高效、可管理的执行环境。我构建平台的目标正是打造这样一个“Harness”。2. 环境准备与核心架构选型在动手之前明确技术边界和选型至关重要。我的目标是构建一个兼顾灵活性、性能和生产就绪性的平台。2.1 技术栈与工具考量后端框架选择Python生态。虽然 Java (Spring AI) 和 Go 也在快速发展但 Python 在 AI/ML 领域的库丰富度、社区活跃度和原型开发速度上仍有显著优势。FastAPI 凭借其异步高性能和自动 API 文档生成成为构建 RESTful 服务层的理想选择。Agent 核心框架不绑定单一框架。平台应能兼容LangChain、LlamaIndex甚至原生的 OpenAI SDK。通过定义统一的接口将框架的具体实现细节封装起来为上层提供一致的体验。LLM 接入层需要支持多模型、多供应商。除了 OpenAI GPT、Anthropic Claude 等闭源模型必须支持通过Ollama、vLLM等方案本地部署的开源模型如 Llama、Qwen、DeepSeek。这要求平台有一个灵活的模型路由和抽象层。记忆与知识库短期记忆会话上下文使用 Redis 进行高效管理。长期记忆和知识检索RAG则集成向量数据库如 Pinecone云服务、Qdrant可自托管或 PGVector与 Postgres 集成。工作流与编排对于复杂的、多步骤的 Agent 任务需要工作流引擎来管理状态和顺序。Dify等平台的工作流功能给了我很大启发但我们需要更底层的控制。可以考虑使用Prefect或Airflow的轻量级模式甚至基于状态机自行实现。可观测性集成 OpenTelemetry 用于分布式追踪将 Agent 的每次思考、每次工具调用都记录为一个 Span。日志集中收集到 ELK 或 Loki指标如 Token 消耗、延迟推送到 Prometheus。前端考虑到交互复杂性如拖拽式工作流编排、实时日志查看采用React或Vue.js构建一个现代化的管理控制台是必要的。2.2 示例平台模型接入抽象层设计为了让不同来源的模型能够被统一调用我们需要设计一个抽象层。以下是一个高度简化的核心接口设计示例# 文件路径platform/core/llm/base.py from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional from pydantic import BaseModel class LLMMessage(BaseModel): role: str # “system”, “user”, “assistant”, “tool” content: str class LLMResult(BaseModel): content: str model: str usage: Optional[Dict[str, int]] None # prompt_tokens, completion_tokens finish_reason: Optional[str] None class BaseLLMClient(ABC): LLM客户端的抽象基类 def __init__(self, model_name: str, **kwargs): self.model_name model_name self.config kwargs abstractmethod async def generate( self, messages: List[LLMMessage], temperature: float 0.7, max_tokens: Optional[int] None, **kwargs ) - LLMResult: 生成聊天补全 pass abstractmethod async def generate_stream(self, messages: List[LLMMessage], **kwargs): 流式生成 pass # 文件路径platform/core/llm/openai_client.py import openai from .base import BaseLLMClient, LLMMessage, LLMResult class OpenAIClient(BaseLLMClient): def __init__(self, model_name: str, api_key: str, base_url: Optional[str]None): super().__init__(model_name) self.client openai.AsyncOpenAI(api_keyapi_key, base_urlbase_url) async def generate(self, messages: List[LLMMessage], **kwargs) - LLMResult: # 将通用消息格式转换为 OpenAI 格式 openai_messages [{role: m.role, content: m.content} for m in messages] response await self.client.chat.completions.create( modelself.model_name, messagesopenai_messages, **kwargs ) choice response.choices[0] return LLMResult( contentchoice.message.content, modelresponse.model, usageresponse.usage.dict() if response.usage else None, finish_reasonchoice.finish_reason ) async def generate_stream(self, messages: List[LLMMessage], **kwargs): # 流式处理实现略 pass # 文件路径platform/core/llm/ollama_client.py import aiohttp from .base import BaseLLMClient, LLMMessage, LLMResult class OllamaClient(BaseLLMClient): def __init__(self, model_name: str, base_url: str http://localhost:11434): super().__init__(model_name) self.base_url base_url async def generate(self, messages: List[LLMMessage], **kwargs) - LLMResult: # 将消息格式转换为 Ollama 格式Ollama 通常使用 prompt 和 context这里做简化适配 # 注意这是一个简化示例实际 Ollama API 可能需要调整 prompt self._format_messages(messages) async with aiohttp.ClientSession() as session: async with session.post( f{self.base_url}/api/generate, json{model: self.model_name, prompt: prompt, stream: False, **kwargs} ) as resp: data await resp.json() return LLMResult( contentdata.get(response, ), modelself.model_name, usageNone, # Ollama 可能不返回用量 finish_reasonNone ) def _format_messages(self, messages: List[LLMMessage]) - str: # 一个简单的消息格式化方法 formatted [] for msg in messages: formatted.append(f{msg.role}: {msg.content}) return \n.join(formatted)通过这样的设计上层业务代码只需要调用BaseLLMClient.generate()而无需关心底层是 OpenAI、Ollama 还是其他任何模型提供商。平台管理员可以在后台动态注册和配置不同的 LLM 客户端。3. 平台核心模块设计与实战一个完整的 AI Agent 平台至少应包含以下几个核心模块。我将以“创建一个能查询天气并给出穿衣建议的 Agent”为例串联这些模块的运作。3.1 Agent 生命周期管理平台需要提供 Agent 的 CRUD创建、读取、更新、删除和版本管理能力。每个 Agent 定义包含其元数据名称、描述、所有者、核心配置使用的 LLM、系统 Prompt、温度等参数和关联的能力工具、知识库。# 示例一个 Agent 的配置定义 (YAML 格式) agent_id: weather-advisor-v1 name: 天气与穿衣助手 description: 根据用户提供的城市查询天气并给出穿衣建议。 owner: team-ai llm_config: provider: openai # 或 ollama, anthropic 等 model: gpt-4o-mini temperature: 0.2 max_tokens: 500 system_prompt: | 你是一个贴心的生活助手。请根据真实的天气数据为用户提供简洁、实用的穿衣和出行建议。 回答要友好并提及关键天气因素如温度、降水、风速。 tools: - tool_id: get_current_weather knowledge_bases: [] # 可以关联 RAG 知识库 version: 1.0.0 status: active # active, inactive, deprecated3.2 工具Tools注册与执行工具是 Agent 的手臂。平台需要提供一个安全、统一的工具注册和执行框架。工具定义开发者通过装饰器或配置文件声明一个工具。安全沙箱对于执行代码或访问敏感资源的工具平台应提供沙箱环境。执行引擎负责调用工具处理输入输出并将结果格式化为 LLM 能理解的消息。# 文件路径platform/tools/weather.py from pydantic import BaseModel, Field from typing import Optional import httpx import os class WeatherToolInput(BaseModel): 获取天气工具的输入参数模型 location: str Field(description城市名称例如北京、Shanghai) unit: Optional[str] Field(defaultcelsius, description温度单位celsius 或 fahrenheit) class WeatherTool: name get_current_weather description 获取指定城市的当前天气情况 args_schema WeatherToolInput def __init__(self): self.api_key os.getenv(WEATHER_API_KEY) self.base_url https://api.weatherapi.com/v1 async def run(self, location: str, unit: str celsius) - str: 执行工具调用 try: async with httpx.AsyncClient() as client: params {key: self.api_key, q: location, aqi: no} resp await client.get(f{self.base_url}/current.json, paramsparams) resp.raise_for_status() data resp.json() current data[current] location_name data[location][name] temp_c current[temp_c] temp_f current[temp_f] condition current[condition][text] wind_kph current[wind_kph] result f{location_name}的当前天气{condition}。温度{temp_c}°C ({temp_f}°F)风速{wind_kph} km/h。 return result except Exception as e: return f查询天气失败{str(e)} # 文件路径platform/core/tool_registry.py class ToolRegistry: 工具注册表单例 _instance None _tools {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, tool_class): 注册一个工具类 tool_instance tool_class() self._tools[tool_instance.name] tool_instance print(f工具已注册: {tool_instance.name}) def get_tool(self, name: str): 根据名称获取工具实例 return self._tools.get(name) async def execute_tool(self, name: str, arguments: dict) - str: 执行指定工具 tool self.get_tool(name) if not tool: return f错误未找到工具 {name} try: # 使用 Pydantic 模型验证输入参数 validated_args tool.args_schema(**arguments) result await tool.run(**validated_args.dict()) return result except Exception as e: return f工具执行出错{str(e)} # 注册工具 registry ToolRegistry() registry.register(WeatherTool)3.3 记忆Memory管理记忆模块负责维护 Agent 与用户交互的上下文。平台需要提供短期会话和长期向量化记忆的存储与检索服务。# 文件路径platform/core/memory/session_memory.py import json from datetime import datetime, timedelta from typing import List, Dict, Any import redis.asyncio as redis class SessionMemory: 基于 Redis 的会话记忆管理 def __init__(self, redis_client: redis.Redis, session_ttl: int 3600): self.redis redis_client self.ttl session_ttl def _get_key(self, session_id: str) - str: return fagent:session:{session_id} async def add_message(self, session_id: str, role: str, content: str, metadata: Dict None): 向会话中添加一条消息 key self._get_key(session_id) message { role: role, content: content, timestamp: datetime.utcnow().isoformat(), metadata: metadata or {} } # 使用 Redis list 存储消息历史 await self.redis.rpush(key, json.dumps(message, ensure_asciiFalse)) await self.redis.expire(key, self.ttl) # 设置过期时间 async def get_messages(self, session_id: str, limit: int 20) - List[Dict]: 获取最近的会话消息 key self._get_key(session_id) messages_json await self.redis.lrange(key, -limit, -1) # 获取最后 limit 条 messages [json.loads(m) for m in messages_json] return messages async def clear(self, session_id: str): 清空会话记忆 key self._get_key(session_id) await self.delete(key)3.4 工作流Workflow编排对于“查询天气 - 分析数据 - 生成建议”这样的多步任务工作流引擎可以更可靠地管理执行顺序和状态。这里展示一个基于状态机的简单编排概念。# 文件路径platform/core/workflow/engine.py from enum import Enum from typing import Dict, Any, Callable, Optional from pydantic import BaseModel class NodeStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed class WorkflowNode(BaseModel): node_id: str name: str action: Callable # 该节点要执行的函数 inputs: Dict[str, Any] {} outputs: Dict[str, Any] {} status: NodeStatus NodeStatus.PENDING next_node_id: Optional[str] None # 下一个节点ID用于线性流程 class SimpleWorkflowEngine: 一个简单的工作流引擎 def __init__(self): self.nodes: Dict[str, WorkflowNode] {} self.context: Dict[str, Any] {} # 工作流共享上下文 def add_node(self, node: WorkflowNode): self.nodes[node.node_id] node async def run(self, start_node_id: str): 从起始节点开始执行工作流 current_node_id start_node_id while current_node_id: node self.nodes.get(current_node_id) if not node: raise ValueError(f节点不存在: {current_node_id}) node.status NodeStatus.RUNNING print(f执行节点: {node.name}) try: # 执行节点动作并传入上下文 result await node.action(self.context, **node.inputs) node.outputs result node.status NodeStatus.SUCCESS # 更新上下文供后续节点使用 self.context.update(result) except Exception as e: node.status NodeStatus.FAILED print(f节点 {node.name} 执行失败: {e}) break # 转移到下一个节点 current_node_id node.next_node_id print(工作流执行完毕。) return self.context # 定义工作流中的具体动作 async def call_weather_tool(context: Dict, city: str) - Dict: registry ToolRegistry() result await registry.execute_tool(get_current_weather, {location: city}) return {weather_data: result} async def call_llm_for_advice(context: Dict) - Dict: # 这里简化处理实际应从 context 中取出 weather_data构造 Prompt 调用 LLM advice f根据天气数据 {context.get(weather_data)}建议穿轻薄外套。 return {advice: advice} # 构建并运行工作流 async def main(): engine SimpleWorkflowEngine() engine.add_node(WorkflowNode( node_idstep1, name查询天气, actioncall_weather_tool, inputs{city: 北京}, next_node_idstep2 )) engine.add_node(WorkflowNode( node_idstep2, name生成穿衣建议, actioncall_llm_for_advice, next_node_idNone # 结束节点 )) final_context await engine.run(step1) print(最终建议, final_context.get(advice))4. 平台的可观测性与运维实践一个看不见、摸不着的 Agent 系统是危险的。平台必须内置强大的可观测性。4.1 链路追踪与日志集成 OpenTelemetry为每次 Agent 调用、每个工具执行、每次 LLM 请求创建追踪 Span。# 文件路径platform/core/observability/tracing.py from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator # 初始化追踪 trace.set_tracer_provider(TracerProvider()) trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) tracer trace.get_tracer(__name__) async def run_agent_with_tracing(session_id: str, user_input: str): with tracer.start_as_current_span(agent_execution) as agent_span: agent_span.set_attribute(session.id, session_id) agent_span.set_attribute(user.input, user_input) # 1. 获取会话历史记忆 with tracer.start_as_current_span(retrieve_memory): memory SessionMemory(redis_client) history await memory.get_messages(session_id) # 2. 调用 LLM 进行规划/思考 with tracer.start_as_current_span(llm_reasoning): llm_client OpenAIClient(...) # ... 构造 messages ... response await llm_client.generate(messages) agent_span.set_attribute(llm.model, response.model) agent_span.set_attribute(llm.token.usage, str(response.usage)) # 3. 如果 LLM 决定调用工具 if need_to_call_tool(response): with tracer.start_as_current_span(tool_execution) as tool_span: tool_name parse_tool_name(response) tool_span.set_attribute(tool.name, tool_name) tool_result await registry.execute_tool(tool_name, ...) tool_span.set_attribute(tool.result.status, success if 失败 not in tool_result else error) # 4. 将结果存入记忆 with tracer.start_as_current_span(update_memory): await memory.add_message(session_id, assistant, final_response) agent_span.set_status(trace.Status(trace.StatusCode.OK))4.2 监控与告警关键指标需要被监控性能指标Agent 请求延迟P50, P95, P99、LLM 响应 Token 速率。业务指标工具调用成功率、各 Agent 的调用频率。成本指标各模型消耗的 Token 数量折算成费用。系统指标内存使用、队列长度。使用 Prometheus 暴露这些指标并在 Grafana 中绘制仪表盘。设定告警规则例如当工具调用失败率连续5分钟超过5%时触发告警。5. 常见问题与排查思路在平台开发和 Agent 应用过程中你会遇到一些典型问题。问题现象可能原因排查步骤与解决方案Agent 响应慢超时1. LLM API 响应慢。2. 工具调用如网络请求耗时过长。3. 记忆检索向量搜索慢。4. 工作流步骤过多。1. 检查链路追踪定位耗时最长的 Span。2. 为 LLM 调用和外部工具调用设置合理的超时时间。3. 优化向量检索的索引和查询语句。4. 考虑对耗时工具进行异步化或缓存结果。LLM 不调用工具或调用错误工具1. 系统 Prompt 中对工具的指令不清晰。2. 工具的描述description不够准确。3. LLM 温度参数过高导致输出不稳定。1. 优化系统 Prompt明确指令格式和调用条件。2. 精炼工具描述确保 LLM 能准确理解其功能。3. 适当降低temperature参数如从 0.8 降至 0.2。4. 在调用 LLM 前可以将可用工具列表再次注入上下文。工具执行权限错误或数据泄露1. 工具代码有安全漏洞如 SQL 注入。2. 平台未对工具输入做严格校验和过滤。3. 敏感信息如 API Key硬编码在代码中。1.所有工具输入必须经过严格的验证和清洗利用 Pydantic。2. 对执行外部命令或代码的工具必须在沙箱环境中运行。3. 敏感配置一律使用环境变量或安全的配置管理服务。4. 实施最小权限原则每个工具只拥有完成其任务所必需的最低权限。会话记忆混乱或丢失1. Redis 内存不足或连接断开。2. Session ID 生成或传递有误。3. 记忆上下文窗口过长导致 LLM 无法有效处理。1. 监控 Redis 状态设置合理的内存淘汰策略和 TTL。2. 确保客户端在连续对话中正确传递唯一的 Session ID。3. 实现记忆摘要Summarization功能将过长的历史压缩成摘要只保留最近几条原始消息。平台自身 API 性能瓶颈1. 同步阻塞式 I/O 操作多。2. 数据库查询未优化。3. 未使用缓存。1.全面采用异步编程async/await。2. 对高频读取的数据如 Agent 配置、工具定义使用 Redis 缓存。3. 使用数据库连接池对复杂查询添加索引。6. 最佳实践与工程建议基于构建过程中的经验教训总结以下几点建议设计先行接口驱动在编码前先定义好清晰的核心接口如BaseLLMClient、BaseTool。这能保证系统各模块间的松耦合便于未来替换具体实现例如从 LangChain 切换到其他框架。配置外置动态生效Agent 的 Prompt、模型参数、工具列表等都应作为配置管理支持热更新。这样可以在不重启服务的情况下快速进行 Prompt 优化和功能迭代。安全第一输入验证对所有用户输入和工具参数进行严格的验证和转义。沙箱隔离对于执行不确定代码的工具如 Python 解释器工具必须运行在安全的容器或沙箱环境中。权限控制实现细粒度的权限系统控制哪些用户/应用可以创建、执行或修改哪些 Agent 和工具。审计日志记录所有关键操作如 Agent 执行、配置修改、工具调用的详细日志用于安全审计和问题回溯。成本控制LLM API 调用是主要成本来源。平台应集成用量监控和配额管理为每个团队或项目设置预算和告警。对于非实时任务可以考虑使用更便宜的模型或批量处理。测试策略单元测试针对每个工具、记忆模块、工具执行引擎进行单元测试。集成测试测试整个 Agent 的端到端流程模拟用户对话。对抗测试设计一些“刁钻”或带有诱导性的输入测试 Agent 是否会被“越狱”或产生有害输出。渐进式演进不要试图一开始就构建一个功能大而全的平台。可以从一个最核心的“Agent 运行时”开始然后逐步添加管理控制台、工作流编排、可观测性等功能。每增加一个特性都要明确它解决了什么具体问题。构建一个 AI Agent 平台是一项复杂的系统工程但它带来的价值是巨大的它将 AI Agent 的开发从“手工作坊”模式升级为“现代化软件工程”模式。通过平台提供的标准化工具、可观测性保障和运维支持开发团队能够更快速、更可靠、更规模化地构建和部署智能体应用真正让 AI Agent 技术赋能于千行百业。希望本文的分享能为你启动自己的 AI Agent 平台工程提供一张实用的地图。
最近在尝试将 AI Agent 应用到实际业务场景时我遇到了一个普遍痛点市面上的开源框架和商业平台要么过于“黑盒”难以深度定制和集成要么过于“原始”需要从零搭建大量基础设施开发效率极低。这种“高不成低不就”的现状促使我萌生了自己动手构建一个 AI Agent 平台的想法。本文将从一个实践者的角度系统性地拆解我构建这个平台的核心动机、技术选型思考以及平台工程的关键设计希望能为同样在探索 AI Agent 落地的开发者提供一份从“为什么”到“怎么做”的实战参考。1. 从概念到现实AI Agent 与平台工程的交汇点在深入平台构建之前我们有必要厘清几个核心概念以及它们如何共同指向了“平台工程”的必要性。1.1 AI Agent 的本质与挑战AI Agent智能体并非一个全新的概念但在大语言模型LLM的加持下其内涵和能力发生了质变。简单来说一个现代的 AI Agent 是一个能够感知环境、进行决策并执行动作以达成目标的智能系统。其核心通常包含几个关键组件大脑Brain通常由 LLM 担任负责理解、规划和推理。记忆Memory用于存储对话历史、工具调用结果、知识片段等分为短期记忆会话上下文和长期记忆向量数据库等。工具Tools赋予 Agent 执行具体任务的能力如调用 API、查询数据库、操作文件等。规划Planning将复杂任务分解为可执行的子任务序列。然而当我们试图将一个简单的 Agent 原型例如用 LangChain 或 LlamaIndex 快速拼装一个推向生产环境时一系列工程挑战便接踵而至开发效率低下每个新 Agent 都需要重复搭建项目结构、配置模型连接、编写工具封装、处理记忆存储。运维复杂度高模型的版本管理、Prompt 的迭代与版本控制、工具 API 的稳定性监控、Agent 运行状态的追踪和日志收集这些都需要大量手工工作。可观测性差当 Agent 执行一个包含多步工具调用的复杂任务时很难清晰地回答“它现在在做什么”“为什么失败了”“哪一步最耗时”缺乏规模化能力如何管理成百上千个不同职能的 Agent如何调度它们的协作如何做统一的权限控制和成本核算1.2 平台工程为 AI Agent 构建“基础设施”这正是平台工程Platform Engineering要解决的问题。平台工程的核心是为内部开发者构建和维护一套标准化的、自助服务的工具链与工作流旨在提升开发者的生产力和体验同时降低系统的复杂性和运维负担。将其映射到 AI Agent 领域一个 AI Agent 平台应致力于解决上述挑战它需要提供一站式开发环境提供从 Agent 创建、工具注册、Prompt 编排、到测试部署的完整流水线。核心能力抽象与复用将模型接入、记忆存储、工具调用、流程编排等能力抽象为平台服务让开发者聚焦业务逻辑。强大的可观测性提供完整的执行链路追踪、详细的日志记录、性能指标监控和成本分析面板。生产级运维支持包括版本管理、灰度发布、弹性伸缩、故障自愈等能力。网络上热议的Harness概念正是一套包裹在 AI Agent 核心推理逻辑之外的基础设施层。它不替代 Agent 的“思考”而是为“思考”提供稳定、高效、可管理的执行环境。我构建平台的目标正是打造这样一个“Harness”。2. 环境准备与核心架构选型在动手之前明确技术边界和选型至关重要。我的目标是构建一个兼顾灵活性、性能和生产就绪性的平台。2.1 技术栈与工具考量后端框架选择Python生态。虽然 Java (Spring AI) 和 Go 也在快速发展但 Python 在 AI/ML 领域的库丰富度、社区活跃度和原型开发速度上仍有显著优势。FastAPI 凭借其异步高性能和自动 API 文档生成成为构建 RESTful 服务层的理想选择。Agent 核心框架不绑定单一框架。平台应能兼容LangChain、LlamaIndex甚至原生的 OpenAI SDK。通过定义统一的接口将框架的具体实现细节封装起来为上层提供一致的体验。LLM 接入层需要支持多模型、多供应商。除了 OpenAI GPT、Anthropic Claude 等闭源模型必须支持通过Ollama、vLLM等方案本地部署的开源模型如 Llama、Qwen、DeepSeek。这要求平台有一个灵活的模型路由和抽象层。记忆与知识库短期记忆会话上下文使用 Redis 进行高效管理。长期记忆和知识检索RAG则集成向量数据库如 Pinecone云服务、Qdrant可自托管或 PGVector与 Postgres 集成。工作流与编排对于复杂的、多步骤的 Agent 任务需要工作流引擎来管理状态和顺序。Dify等平台的工作流功能给了我很大启发但我们需要更底层的控制。可以考虑使用Prefect或Airflow的轻量级模式甚至基于状态机自行实现。可观测性集成 OpenTelemetry 用于分布式追踪将 Agent 的每次思考、每次工具调用都记录为一个 Span。日志集中收集到 ELK 或 Loki指标如 Token 消耗、延迟推送到 Prometheus。前端考虑到交互复杂性如拖拽式工作流编排、实时日志查看采用React或Vue.js构建一个现代化的管理控制台是必要的。2.2 示例平台模型接入抽象层设计为了让不同来源的模型能够被统一调用我们需要设计一个抽象层。以下是一个高度简化的核心接口设计示例# 文件路径platform/core/llm/base.py from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional from pydantic import BaseModel class LLMMessage(BaseModel): role: str # “system”, “user”, “assistant”, “tool” content: str class LLMResult(BaseModel): content: str model: str usage: Optional[Dict[str, int]] None # prompt_tokens, completion_tokens finish_reason: Optional[str] None class BaseLLMClient(ABC): LLM客户端的抽象基类 def __init__(self, model_name: str, **kwargs): self.model_name model_name self.config kwargs abstractmethod async def generate( self, messages: List[LLMMessage], temperature: float 0.7, max_tokens: Optional[int] None, **kwargs ) - LLMResult: 生成聊天补全 pass abstractmethod async def generate_stream(self, messages: List[LLMMessage], **kwargs): 流式生成 pass # 文件路径platform/core/llm/openai_client.py import openai from .base import BaseLLMClient, LLMMessage, LLMResult class OpenAIClient(BaseLLMClient): def __init__(self, model_name: str, api_key: str, base_url: Optional[str]None): super().__init__(model_name) self.client openai.AsyncOpenAI(api_keyapi_key, base_urlbase_url) async def generate(self, messages: List[LLMMessage], **kwargs) - LLMResult: # 将通用消息格式转换为 OpenAI 格式 openai_messages [{role: m.role, content: m.content} for m in messages] response await self.client.chat.completions.create( modelself.model_name, messagesopenai_messages, **kwargs ) choice response.choices[0] return LLMResult( contentchoice.message.content, modelresponse.model, usageresponse.usage.dict() if response.usage else None, finish_reasonchoice.finish_reason ) async def generate_stream(self, messages: List[LLMMessage], **kwargs): # 流式处理实现略 pass # 文件路径platform/core/llm/ollama_client.py import aiohttp from .base import BaseLLMClient, LLMMessage, LLMResult class OllamaClient(BaseLLMClient): def __init__(self, model_name: str, base_url: str http://localhost:11434): super().__init__(model_name) self.base_url base_url async def generate(self, messages: List[LLMMessage], **kwargs) - LLMResult: # 将消息格式转换为 Ollama 格式Ollama 通常使用 prompt 和 context这里做简化适配 # 注意这是一个简化示例实际 Ollama API 可能需要调整 prompt self._format_messages(messages) async with aiohttp.ClientSession() as session: async with session.post( f{self.base_url}/api/generate, json{model: self.model_name, prompt: prompt, stream: False, **kwargs} ) as resp: data await resp.json() return LLMResult( contentdata.get(response, ), modelself.model_name, usageNone, # Ollama 可能不返回用量 finish_reasonNone ) def _format_messages(self, messages: List[LLMMessage]) - str: # 一个简单的消息格式化方法 formatted [] for msg in messages: formatted.append(f{msg.role}: {msg.content}) return \n.join(formatted)通过这样的设计上层业务代码只需要调用BaseLLMClient.generate()而无需关心底层是 OpenAI、Ollama 还是其他任何模型提供商。平台管理员可以在后台动态注册和配置不同的 LLM 客户端。3. 平台核心模块设计与实战一个完整的 AI Agent 平台至少应包含以下几个核心模块。我将以“创建一个能查询天气并给出穿衣建议的 Agent”为例串联这些模块的运作。3.1 Agent 生命周期管理平台需要提供 Agent 的 CRUD创建、读取、更新、删除和版本管理能力。每个 Agent 定义包含其元数据名称、描述、所有者、核心配置使用的 LLM、系统 Prompt、温度等参数和关联的能力工具、知识库。# 示例一个 Agent 的配置定义 (YAML 格式) agent_id: weather-advisor-v1 name: 天气与穿衣助手 description: 根据用户提供的城市查询天气并给出穿衣建议。 owner: team-ai llm_config: provider: openai # 或 ollama, anthropic 等 model: gpt-4o-mini temperature: 0.2 max_tokens: 500 system_prompt: | 你是一个贴心的生活助手。请根据真实的天气数据为用户提供简洁、实用的穿衣和出行建议。 回答要友好并提及关键天气因素如温度、降水、风速。 tools: - tool_id: get_current_weather knowledge_bases: [] # 可以关联 RAG 知识库 version: 1.0.0 status: active # active, inactive, deprecated3.2 工具Tools注册与执行工具是 Agent 的手臂。平台需要提供一个安全、统一的工具注册和执行框架。工具定义开发者通过装饰器或配置文件声明一个工具。安全沙箱对于执行代码或访问敏感资源的工具平台应提供沙箱环境。执行引擎负责调用工具处理输入输出并将结果格式化为 LLM 能理解的消息。# 文件路径platform/tools/weather.py from pydantic import BaseModel, Field from typing import Optional import httpx import os class WeatherToolInput(BaseModel): 获取天气工具的输入参数模型 location: str Field(description城市名称例如北京、Shanghai) unit: Optional[str] Field(defaultcelsius, description温度单位celsius 或 fahrenheit) class WeatherTool: name get_current_weather description 获取指定城市的当前天气情况 args_schema WeatherToolInput def __init__(self): self.api_key os.getenv(WEATHER_API_KEY) self.base_url https://api.weatherapi.com/v1 async def run(self, location: str, unit: str celsius) - str: 执行工具调用 try: async with httpx.AsyncClient() as client: params {key: self.api_key, q: location, aqi: no} resp await client.get(f{self.base_url}/current.json, paramsparams) resp.raise_for_status() data resp.json() current data[current] location_name data[location][name] temp_c current[temp_c] temp_f current[temp_f] condition current[condition][text] wind_kph current[wind_kph] result f{location_name}的当前天气{condition}。温度{temp_c}°C ({temp_f}°F)风速{wind_kph} km/h。 return result except Exception as e: return f查询天气失败{str(e)} # 文件路径platform/core/tool_registry.py class ToolRegistry: 工具注册表单例 _instance None _tools {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def register(self, tool_class): 注册一个工具类 tool_instance tool_class() self._tools[tool_instance.name] tool_instance print(f工具已注册: {tool_instance.name}) def get_tool(self, name: str): 根据名称获取工具实例 return self._tools.get(name) async def execute_tool(self, name: str, arguments: dict) - str: 执行指定工具 tool self.get_tool(name) if not tool: return f错误未找到工具 {name} try: # 使用 Pydantic 模型验证输入参数 validated_args tool.args_schema(**arguments) result await tool.run(**validated_args.dict()) return result except Exception as e: return f工具执行出错{str(e)} # 注册工具 registry ToolRegistry() registry.register(WeatherTool)3.3 记忆Memory管理记忆模块负责维护 Agent 与用户交互的上下文。平台需要提供短期会话和长期向量化记忆的存储与检索服务。# 文件路径platform/core/memory/session_memory.py import json from datetime import datetime, timedelta from typing import List, Dict, Any import redis.asyncio as redis class SessionMemory: 基于 Redis 的会话记忆管理 def __init__(self, redis_client: redis.Redis, session_ttl: int 3600): self.redis redis_client self.ttl session_ttl def _get_key(self, session_id: str) - str: return fagent:session:{session_id} async def add_message(self, session_id: str, role: str, content: str, metadata: Dict None): 向会话中添加一条消息 key self._get_key(session_id) message { role: role, content: content, timestamp: datetime.utcnow().isoformat(), metadata: metadata or {} } # 使用 Redis list 存储消息历史 await self.redis.rpush(key, json.dumps(message, ensure_asciiFalse)) await self.redis.expire(key, self.ttl) # 设置过期时间 async def get_messages(self, session_id: str, limit: int 20) - List[Dict]: 获取最近的会话消息 key self._get_key(session_id) messages_json await self.redis.lrange(key, -limit, -1) # 获取最后 limit 条 messages [json.loads(m) for m in messages_json] return messages async def clear(self, session_id: str): 清空会话记忆 key self._get_key(session_id) await self.delete(key)3.4 工作流Workflow编排对于“查询天气 - 分析数据 - 生成建议”这样的多步任务工作流引擎可以更可靠地管理执行顺序和状态。这里展示一个基于状态机的简单编排概念。# 文件路径platform/core/workflow/engine.py from enum import Enum from typing import Dict, Any, Callable, Optional from pydantic import BaseModel class NodeStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed class WorkflowNode(BaseModel): node_id: str name: str action: Callable # 该节点要执行的函数 inputs: Dict[str, Any] {} outputs: Dict[str, Any] {} status: NodeStatus NodeStatus.PENDING next_node_id: Optional[str] None # 下一个节点ID用于线性流程 class SimpleWorkflowEngine: 一个简单的工作流引擎 def __init__(self): self.nodes: Dict[str, WorkflowNode] {} self.context: Dict[str, Any] {} # 工作流共享上下文 def add_node(self, node: WorkflowNode): self.nodes[node.node_id] node async def run(self, start_node_id: str): 从起始节点开始执行工作流 current_node_id start_node_id while current_node_id: node self.nodes.get(current_node_id) if not node: raise ValueError(f节点不存在: {current_node_id}) node.status NodeStatus.RUNNING print(f执行节点: {node.name}) try: # 执行节点动作并传入上下文 result await node.action(self.context, **node.inputs) node.outputs result node.status NodeStatus.SUCCESS # 更新上下文供后续节点使用 self.context.update(result) except Exception as e: node.status NodeStatus.FAILED print(f节点 {node.name} 执行失败: {e}) break # 转移到下一个节点 current_node_id node.next_node_id print(工作流执行完毕。) return self.context # 定义工作流中的具体动作 async def call_weather_tool(context: Dict, city: str) - Dict: registry ToolRegistry() result await registry.execute_tool(get_current_weather, {location: city}) return {weather_data: result} async def call_llm_for_advice(context: Dict) - Dict: # 这里简化处理实际应从 context 中取出 weather_data构造 Prompt 调用 LLM advice f根据天气数据 {context.get(weather_data)}建议穿轻薄外套。 return {advice: advice} # 构建并运行工作流 async def main(): engine SimpleWorkflowEngine() engine.add_node(WorkflowNode( node_idstep1, name查询天气, actioncall_weather_tool, inputs{city: 北京}, next_node_idstep2 )) engine.add_node(WorkflowNode( node_idstep2, name生成穿衣建议, actioncall_llm_for_advice, next_node_idNone # 结束节点 )) final_context await engine.run(step1) print(最终建议, final_context.get(advice))4. 平台的可观测性与运维实践一个看不见、摸不着的 Agent 系统是危险的。平台必须内置强大的可观测性。4.1 链路追踪与日志集成 OpenTelemetry为每次 Agent 调用、每个工具执行、每次 LLM 请求创建追踪 Span。# 文件路径platform/core/observability/tracing.py from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator # 初始化追踪 trace.set_tracer_provider(TracerProvider()) trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) tracer trace.get_tracer(__name__) async def run_agent_with_tracing(session_id: str, user_input: str): with tracer.start_as_current_span(agent_execution) as agent_span: agent_span.set_attribute(session.id, session_id) agent_span.set_attribute(user.input, user_input) # 1. 获取会话历史记忆 with tracer.start_as_current_span(retrieve_memory): memory SessionMemory(redis_client) history await memory.get_messages(session_id) # 2. 调用 LLM 进行规划/思考 with tracer.start_as_current_span(llm_reasoning): llm_client OpenAIClient(...) # ... 构造 messages ... response await llm_client.generate(messages) agent_span.set_attribute(llm.model, response.model) agent_span.set_attribute(llm.token.usage, str(response.usage)) # 3. 如果 LLM 决定调用工具 if need_to_call_tool(response): with tracer.start_as_current_span(tool_execution) as tool_span: tool_name parse_tool_name(response) tool_span.set_attribute(tool.name, tool_name) tool_result await registry.execute_tool(tool_name, ...) tool_span.set_attribute(tool.result.status, success if 失败 not in tool_result else error) # 4. 将结果存入记忆 with tracer.start_as_current_span(update_memory): await memory.add_message(session_id, assistant, final_response) agent_span.set_status(trace.Status(trace.StatusCode.OK))4.2 监控与告警关键指标需要被监控性能指标Agent 请求延迟P50, P95, P99、LLM 响应 Token 速率。业务指标工具调用成功率、各 Agent 的调用频率。成本指标各模型消耗的 Token 数量折算成费用。系统指标内存使用、队列长度。使用 Prometheus 暴露这些指标并在 Grafana 中绘制仪表盘。设定告警规则例如当工具调用失败率连续5分钟超过5%时触发告警。5. 常见问题与排查思路在平台开发和 Agent 应用过程中你会遇到一些典型问题。问题现象可能原因排查步骤与解决方案Agent 响应慢超时1. LLM API 响应慢。2. 工具调用如网络请求耗时过长。3. 记忆检索向量搜索慢。4. 工作流步骤过多。1. 检查链路追踪定位耗时最长的 Span。2. 为 LLM 调用和外部工具调用设置合理的超时时间。3. 优化向量检索的索引和查询语句。4. 考虑对耗时工具进行异步化或缓存结果。LLM 不调用工具或调用错误工具1. 系统 Prompt 中对工具的指令不清晰。2. 工具的描述description不够准确。3. LLM 温度参数过高导致输出不稳定。1. 优化系统 Prompt明确指令格式和调用条件。2. 精炼工具描述确保 LLM 能准确理解其功能。3. 适当降低temperature参数如从 0.8 降至 0.2。4. 在调用 LLM 前可以将可用工具列表再次注入上下文。工具执行权限错误或数据泄露1. 工具代码有安全漏洞如 SQL 注入。2. 平台未对工具输入做严格校验和过滤。3. 敏感信息如 API Key硬编码在代码中。1.所有工具输入必须经过严格的验证和清洗利用 Pydantic。2. 对执行外部命令或代码的工具必须在沙箱环境中运行。3. 敏感配置一律使用环境变量或安全的配置管理服务。4. 实施最小权限原则每个工具只拥有完成其任务所必需的最低权限。会话记忆混乱或丢失1. Redis 内存不足或连接断开。2. Session ID 生成或传递有误。3. 记忆上下文窗口过长导致 LLM 无法有效处理。1. 监控 Redis 状态设置合理的内存淘汰策略和 TTL。2. 确保客户端在连续对话中正确传递唯一的 Session ID。3. 实现记忆摘要Summarization功能将过长的历史压缩成摘要只保留最近几条原始消息。平台自身 API 性能瓶颈1. 同步阻塞式 I/O 操作多。2. 数据库查询未优化。3. 未使用缓存。1.全面采用异步编程async/await。2. 对高频读取的数据如 Agent 配置、工具定义使用 Redis 缓存。3. 使用数据库连接池对复杂查询添加索引。6. 最佳实践与工程建议基于构建过程中的经验教训总结以下几点建议设计先行接口驱动在编码前先定义好清晰的核心接口如BaseLLMClient、BaseTool。这能保证系统各模块间的松耦合便于未来替换具体实现例如从 LangChain 切换到其他框架。配置外置动态生效Agent 的 Prompt、模型参数、工具列表等都应作为配置管理支持热更新。这样可以在不重启服务的情况下快速进行 Prompt 优化和功能迭代。安全第一输入验证对所有用户输入和工具参数进行严格的验证和转义。沙箱隔离对于执行不确定代码的工具如 Python 解释器工具必须运行在安全的容器或沙箱环境中。权限控制实现细粒度的权限系统控制哪些用户/应用可以创建、执行或修改哪些 Agent 和工具。审计日志记录所有关键操作如 Agent 执行、配置修改、工具调用的详细日志用于安全审计和问题回溯。成本控制LLM API 调用是主要成本来源。平台应集成用量监控和配额管理为每个团队或项目设置预算和告警。对于非实时任务可以考虑使用更便宜的模型或批量处理。测试策略单元测试针对每个工具、记忆模块、工具执行引擎进行单元测试。集成测试测试整个 Agent 的端到端流程模拟用户对话。对抗测试设计一些“刁钻”或带有诱导性的输入测试 Agent 是否会被“越狱”或产生有害输出。渐进式演进不要试图一开始就构建一个功能大而全的平台。可以从一个最核心的“Agent 运行时”开始然后逐步添加管理控制台、工作流编排、可观测性等功能。每增加一个特性都要明确它解决了什么具体问题。构建一个 AI Agent 平台是一项复杂的系统工程但它带来的价值是巨大的它将 AI Agent 的开发从“手工作坊”模式升级为“现代化软件工程”模式。通过平台提供的标准化工具、可观测性保障和运维支持开发团队能够更快速、更可靠、更规模化地构建和部署智能体应用真正让 AI Agent 技术赋能于千行百业。希望本文的分享能为你启动自己的 AI Agent 平台工程提供一张实用的地图。
)
