尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

从Demo到生产:用Python构建可观测的智能体产品化闭环

从Demo到生产:用Python构建可观测的智能体产品化闭环 智能体Agent的开发热潮已经进入产品化阶段。大量团队能在一周内做出一个调用大模型、执行简单工具的 Demo但真正把 Agent 稳定放到业务里运行情况就完全不同上下文会越来越长、工具调用会突然失败、模型输出格式偶尔不可解析、记忆不知道存多少日志也无法回答“上一轮到底发生了什么”。Charlie Holtz 等一线 AI 产品负责人近期反复强调同一个方向当前最值得做的不是再造一个“参数更高”的模型而是打造智能体真正需要的产品。这里的产品不是单一 App而是模型层、框架层、工具层、数据层、可观测层和评估体系组成的完整支撑系统。这篇文章不会停留在概念层面。我会先拆解智能体产品的分层结构再带你从零实现一个带天气查询、计算器和当前时间工具的最小智能体。代码使用 Python采用 OpenAI 兼容协议调用大模型接口。你会看到工具调用闭环、记忆组织、上下文管理和报错排查是怎么做的最后会给出从 Demo 到生产环境的工程化清单和常见排错方法。1. 智能体产品到底由哪些部分组成1.1 智能体不只是一个“会聊天的模型”智能体的技术定义可以简化为一句话一个能够感知环境、做出决策、调用工具并完成任务的系统。它和普通聊天机器人最大的区别不是能“聊天”而是能“行动”。例如用户说“帮我查一下北京天气并把结果写成日报”聊天机器人可能只会生成一段文字而智能体需要拆解任务、查询天气 API、把结果格式化甚至在指定存储位置写入文件。这个区别意味着智能体必须有一个稳定的执行闭环模型负责理解和规划代码负责执行动作系统负责记录结果。如果没有外部工具、状态管理和异常处理模型再强也只是“嘴皮子好”。在实际工程里智能体至少需要四个基本模块模型调用层负责与大模型通信传入 prompt 和上下文获取输出。工具层把 API、数据库、文件系统、计算器等能力封装成可调用函数。记忆层保存短期对话历史和长期知识避免每次请求都重复灌输全部内容。循环控制层决定模型输出是否需要继续调用工具以及多久必须终止。这四个模块缺一不可。很多项目跑不通问题往往不是模型不够强而是循环控制和上下文处理写得太随意。1.2 智能体产品矩阵从模型到评估的六层结构如果把“打造智能体所需产品”看成一个技术栈可以拆成六层。每一层解决一个不同的问题。层级解决什么问题常见方案类型生产环境关注点模型层语言理解、推理、生成大模型 API、私有化模型模型版本、延迟、成本、输出稳定性框架层简化 Agent 循环、工具注册、记忆管理LangChain、LlamaIndex、AutoGen抽象是否透明、升级是否兼容平台层可视化搭建、工作流编排、运维管理Dify、Coze、HiAgent 等灵活度、私有化部署、权限体系工具层让 Agent 和业务系统交互API 网关、RPA、代码解释器鉴权、超时、幂等性、错误码数据层向量检索、长期记忆、知识库向量数据库、关系型数据库、对象存储数据一致性、召回质量、隐私合规观测评估层日志、链路追踪、自动评测Langfuse、自建日志平台、评测集是否覆盖每一轮工具调用和 token 消耗这六层并不是每个智能体项目都需要立刻全部搭建。学习阶段可以先只写模型层、工具层和循环控制层但进入生产后观测评估层和数据层往往比模型层更影响稳定性。1.3 为什么平台和工具型产品会成为下一阶段重点模型能力的提升会让“搭一个 Agent”变得更简单但不会自动解决“让 Agent 稳定工作”的问题。近两年智能体相关热搜词里出现最多的是框架、搭建平台、落地流程和开发工程师岗位而不是单一模型名称。这说明行业关注点正在从“模型能做什么”转向“产品能不能让 Agent 被可靠地交付”。一个企业要落地智能体通常不会只接一个模型就结束。它需要把内部业务 API 标准化成 Agent 可识别的工具。设计用户意图识别和任务拆分规则。构建统一日志能追溯每一轮决策。建立回归测试集防止模型升级后行为漂移。这些工作本质上是“智能体所需的产品”。Chat 界面只是入口真正的产品价值在入口背后的工程体系。2. 搭建前先想清楚技术选型2.1 场景决定智能体的复杂度不同智能体场景的复杂度差异很大。例如客服问答型主要依赖检索增强生成和记忆管理工具调用较少。业务操作型需要调用多个系统 API对工具调用稳定性和权限控制要求高。数据分析型需要生成代码并执行还要防止危险操作。多智能体协作型需要设计任务分发、结果汇总和冲突处理。在开始写代码之前先回答三个问题用户任务是否需要真实改变外部系统状态一个任务是否必须经过多步工具调用才能完成多轮对话之间是否需要保存状态如果三个问题的答案都是否那你要做的可能只是增强版聊天助手不必引入复杂 Agent 框架。如果答案有“是”才值得投入完整 Agent 架构。2.2 三类主流搭建方式目前常见的智能体搭建方式可以分成三类。搭建方式适合场景优点需要注意的问题低代码平台业务人员快速搭建、流程相对固定上手快、内置组件多定制扩展受限平台更新可能影响现有流程代码框架研发深度定制、需要复用社区生态灵活、可扩展框架抽象多版本升级容易踩坑完全自研强定制、离线环境、长期运维逻辑透明、可控性强开发成本高需要自己处理很多细节低代码平台适合“先验证业务价值”。代码框架适合“已经明确要做复杂 Agent”。完全自研适合“团队对 Agent 内部机制有足够把控力且外部框架无法满足需求”。没有绝对正确选型关键是不要为了追新而引入不必要复杂度。如果业务只是“调模型、检索、返回答案”直接写几十行代码比套框架更容易排查。2.3 我的选择用最少依赖实现一个可观察的 Agent后面所有示例我会采用一个很轻量的自研实现。它不依赖 LangChain 这类重量级框架只用 OpenAI Python 包和一个核心循环。这样做的原因是代码路径短每一轮模型输出、工具调用、结果返回都能清楚看到。方便打断点也方便把日志输出到文件或链路追踪系统。更容易理解 Agent 的本质而不是被框架抽象淹没。这个实现适合学习也适合作为生产项目的起点。生产项目可以在它的基础上加入消息队列、缓存、权限校验和可观测性组件。3. 实现一个最小可运行智能体3.1 环境准备和依赖示例环境建议使用 Python 3.10 及以上版本。核心依赖只有一个 OpenAI SDK它不仅能访问 OpenAI 模型也能通过base_url接入兼容 OpenAI 协议的大模型服务。mkdir agent-demo cd agent-demo python -m venv .venv source .venv/bin/activate pip install openai python-dotenv代码中会通过环境变量读取模型配置。在项目目录下创建.env文件LLM_API_KEY你的_LLM_API_KEY LLM_BASE_URLhttps://api.example.com/v1 LLM_MODELgpt-4o-mini注意LLM_BASE_URL需要根据你实际使用的模型服务地址填写。不同服务商的协议路径不同落地前要先确认是否兼容 OpenAI 的/chat/completions接口格式。然后创建入口文件agent.py。后面的示例都写在这个文件里。3.2 项目文件结构最小智能体的项目结构可以保持非常简单agent-demo/ ├── .env ├── agent.py └── requirements.txtrequirements.txt内容openai1.0.0 python-dotenv1.0.0如果原始项目依赖版本未固定建议在安装后执行pip freeze requirements.lock锁定版本这能降低“昨天还能跑今天突然报错”的概率。3.3 工具层先定义能力边界智能体的工具层就是一组普通函数。关键在于统一输入输出格式这样循环控制层才能稳定调用。import json import os import re from datetime import datetime from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), ) MODEL os.getenv(LLM_MODEL, gpt-4o-mini) def get_weather(city: str) - str: 演示工具真实项目应替换为天气 API 请求。 return f{city} 今天晴气温 22 摄氏度。 def calculate(expression: str) - str: 计算数学表达式。 教学示例直接使用 eval生产环境不要这样做。 生产环境可以改用 ast 解析或专用计算库防止任意代码执行。 return str(eval(expression)) def get_current_time() - str: return datetime.now().strftime(%Y-%m-%d %H:%M:%S) TOOLS { get_weather: get_weather, calculate: calculate, get_current_time: get_current_time, }这里有几个关键点每个函数都设置了明确的参数名和类型方便模型生成 JSON 输入。TOOLS字典是工具注册中心新的工具只要加入这个字典Agent 循环就能识别。返回值统一是字符串后面会作为Observation塞回上下文。如果返回复杂对象会造成格式化混乱。calculate中的eval是一个安全隐患只用于教学。生产环境要替换成安全计算方案例如ast.literal_eval或表达式解析库。3.4 核心循环模型出计划代码去执行Agent 的核心循环常被称为 ReAct 循环Thought思考、Action行动、Observation观察。模型先观察用户输入决定调用哪个工具代码执行工具后把结果作为新的观察交回模型模型决定是继续行动还是给出最终答案。下面先定义 system prompt 和模型调用函数。SYSTEM_PROMPT 你是一个智能体助手。请按以下格式处理任务 Thought: 你观察输入决定下一步行动。 Action: 工具名 Action Input: 给工具的 JSON 输入 Observation: 工具返回结果由系统提供 重复Thought/Action/Action Input/Observation直到任务完成最后输出 Final Answer: 给用户的最终回答 可用工具 - get_weather: 查询天气输入 {city: 城市名} - calculate: 计算数学表达式输入 {expression: 例如 17*23} - get_current_time: 获取当前时间输入 {} 只使用上述工具不要编造工具结果。 def call_llm(messages): response client.chat.completions.create( modelMODEL, messagesmessages, temperature0, ) return response.choices[0].message.content将temperature设置为 0是希望模型尽可能稳定输出可解析格式而不是发挥创造性。Agent 场景和写作场景相反这里需要确定性优先。然后是主循环def run_agent(user_input: str, max_steps: int 5) - str: messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_input}, ] for step in range(max_steps): reply call_llm(messages) print(f\n[step {step}] model output:\n{reply}\n) if Final Answer: in reply: return reply.split(Final Answer:, 1)[1].strip() action_match re.search(rAction:\s*(\w), reply) input_match re.search(rAction Input:\s*(\{.*?\}), reply, re.DOTALL) if not action_match or not input_match: messages.append({role: assistant, content: reply}) messages.append( {role: user, content: 你的输出缺少 Action 或 Action Input请严格按格式重新输出。} ) continue action action_match.group(1) try: action_input json.loads(input_match.group(1)) except json.JSONDecodeError as exc: messages.append({role: assistant, content: reply}) messages.append( {role: user, content: fAction Input 不是合法 JSON解析失败{exc}请重新输出。} ) continue if action not in TOOLS: result f错误未知工具 {action} else: try: result TOOLS[action](**action_input) except Exception as exc: result f工具调用失败{exc} messages.append({role: assistant, content: reply}) messages.append({role: user, content: fObservation: {result}}) return 达到最大步数无法得到最终结果。这个循环的关键设计是每轮都把模型完整的输出追加到messages保证多轮上下文连续。工具调用结果通过Observation: {result}注入模型能明确区分哪些是用户输入哪些是工具结果。解析失败时不会直接报错而是把错误信息作为一条用户消息回传让模型自行纠正。max_steps是必设的终止条件防止模型反复调用工具进入死循环。3.5 记忆处理和上下文组装上面示例中的messages列表就是短期记忆。每一轮模型输出和工具观察都会追加到列表里所以模型能记住“已经查过北京天气”不会重复查。但这只是最基础的记忆。真实项目中记忆需要更多策略截断旧消息上下文过长时丢弃最早的非核心消息。摘要记忆定期把前面的对话摘要成一段文本节省 token。长期记忆把重要事实写入向量库需要时检索回来。工具观察压缩工具返回超长 JSON 时提取关键字段而不是全文塞回。例如在messages超过一定条数后可以简单丢弃最早的用户消息但保留 system prompt。更复杂的项目可以增加摘要节点。3.6 运行验证与预期输出在agent.py末尾加入测试入口if __name__ __main__: result run_agent(北京天气怎么样顺便算一下 17 乘以 23 等于多少) print(\n最终回答) print(result)运行python agent.py正常执行时控制台会输出类似下面的日志[step 0] model output: Thought: 用户想知道北京天气还要计算 17 乘以 23。先查天气。 Action: get_weather Action Input: {city: 北京} [step 1] model output: Thought: 已经得到北京天气。再计算 17 乘以 23。 Action: calculate Action Input: {expression: 17*23} [step 2] model output: Thought: 已经知道天气和计算结果可以回答用户。 Final Answer: 北京今天晴气温22摄氏度17乘以23等于391。 最终回答 北京今天晴气温22摄氏度17乘以23等于391。实际输出会随模型版本和 prompt 风格变化但结构与上面类似。只要看到“Action → Observation → Final Answer”的闭环就说明这个最小智能体已经能自主完成多步任务。4. 关键参数和行为控制4.1 不要无限塞给模型上下文Agent 中messages列表会随着工具调用增长。假设一次任务需要 5 步每步模型输出约 200 token工具结果约 100 token那一次对话就可能消耗 1500 token。看起来不多但如果是长期运行的客服助手单用户多轮交互后上下文会迅速膨胀。常用控制手段策略做法适用场景滑动窗口保留最近 N 条消息丢弃更早内容短期任务型会话Token 截断超过阈值后丢弃最旧消息通用场景摘要压缩把旧消息摘要成一段话长对话、需要长期语义只保留必要字段工具结果裁剪为关键字段工具返回大 JSON 时实际项目中不建议一开始就实现很复杂的记忆策略。先统计平均每轮 token 消耗再决定是否需要压缩。4.2 工具调用的超时、重试和异常工具本身不能假设永远成功。天气 API 可能超时数据库可能连接失败第三方接口可能返回 500。生产环境中工具函数必须做好三件事设置超时。例如 requests 请求设置timeout10避免线程被卡死。明确返回错误结果。不要自己吞异常要把错误信息返回给模型让模型决定下一步。区分“业务错误”和“系统错误”。业务错误可以原样返回系统错误要记录日志并触发告警。一个更健壮的get_weather可以写成def get_weather(city: str) - str: try: # 真实项目中发送 HTTP 请求 return f{city} 今天晴气温 22 摄氏度。 except Exception as exc: return f天气查询失败{exc}请稍后重试。这样模型看到 “查询失败”会自动判断是否需要换一种方式处理而不是让整个 Agent 崩溃。4.3 并发与限流学习环境可以直接同步调用模型接口。生产环境需要考虑并发和限流如果入口是 REST API需要设置用户配额防止单个用户刷爆 token。如果模型接口有 RPM/TPM 限制需要做请求队列和重试。如果同一任务要调用多个工具部分工具之间可以并行执行但要控制并发数。一个简单的并发限制思路是使用线程池from concurrent.futures import ThreadPoolExecutor def run_parallel_tools(tool_calls, max_workers3): with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(tool_calls[action], **tool_calls[input]): tool_calls} results {} for future in futures: results[future.result()] future return results这只是一个演示结构实际还要处理异常返回、超时和顺序一致性。如果多个工具调用之间没有依赖并行可以缩短延迟但要注意部分 API 不支持并发请求或并发量过大会触发限流。5. 常见问题与排查链路5.1 现象一模型不调用工具直接给答案这是最常见的现象。模型收到“查询北京天气”后可能会直接回答“北京今天晴”而不是执行get_weather。原因是模型用自己的训练知识猜测了结果而不是调用工具。排查思路检查 system prompt 是否明确规定了“必须使用工具”。检查工具描述是否清晰模型可能不知道在什么时候调用。检查模型本身是否支持函数调用部分轻量模型对复杂指令理解能力弱。解决方法是增加一条约束“如果用户请求中涉及可用工具必须调用工具不能直接编造结果。”同时在工具描述中写清楚使用条件。5.2 现象二Action Input 解析失败模型输出的 JSON 可能是单引号、缺少引号、包含多余换行导致json.loads失败。常见的错误输出Action Input: {city: 北京}这是 Python 风格字典不是合法 JSON。单纯用json.loads会报错。处理方式在 prompt 中明确要求输出合法 JSON。解析失败时不要直接终止而是把错误信息回传给模型修正。可以在json.loads前先做简单清洗例如把单引号替换为双引号但要谨慎避免改坏字符串内容。更稳妥的做法是让模型用 JSON 格式输出完整{ action: ..., input: {...} }再用 JSON Schema 校验。5.3 现象三工具执行成功后 Agent 仍不收敛如果模型在拿到Observation后没有输出 Final Answer而是继续调用同一个工具很可能是因为 prompt 没有说明何时停止。比如工具返回很长模型认为任务还没完成。解决办法在 system prompt 中写清楚“一旦得到所有需要的信息立即输出 Final Answer。”设置max_steps避免无限循环。记录每轮 Action 和工具调用次数如果同一工具被重复调用超过阈值直接强制结束。5.4 排查顺序和日志规范当智能体出现问题时按照从外到内的顺序排查排查步骤检查内容1. 输入用户输入是否包含不可达或歧义指令2. 配置API Key、Base URL、模型名是否正确3. 模型输出是否生成合法 Action 和 Action Input4. 工具执行工具是否超时、是否抛异常5. 上下文组装Observation 是否被正确追加6. 终止判断是否达到 max_steps 或误判 Final Answer为了支持这条链路日志至少要记录以下字段会话 ID步骤序号请求发送给模型的完整 messages可脱敏模型原始输出工具名称和输入输出每次调用的 token 消耗和耗时没有这些日志排查 Agent 问题就像盲人摸象。6. 从 Demo 到生产的工程化落地6.1 学习环境和生产环境的差异上面的最小实现适合理解原理但不能直接搬到生产。生产环境至少要多出以下保障。能力学习环境生产环境密钥管理本地 .env密钥管理服务环境变量动态注入日志print 控制台结构化日志集中采集可观测性无Trace、指标、告警限流无用户配额、接口限流多租户无用户隔离、权限审计评估人工看一两条结果自动化回归测试集部署本地 Python 进程容器化、弹性伸缩、滚动发布异常恢复直接失败重试、降级、人工兜底这些差异不需要一次全部实现。生产上线前一版至少要把密钥管理和日志链路做对。6.2 可观测性每一步都要能回放智能体项目比普通 API 项目更需要可观测性因为一个用户请求背后可能有多轮模型调用和多步工具执行。建议为每个 Agent 请求生成一个唯一的request_id并在日志中带上这个 ID。结构化日志示例{ request_id: req_001, step: 2, action: get_weather, action_input: {city: 北京}, action_result: 北京 今天晴气温 22 摄氏度。, token_usage: {prompt: 1200, completion: 80}, latency_ms: 345 }采集后运维人员可以根据request_id还原完整链路用户说了什么、模型想了什么、工具返回了什么、最终答了什么。这是排查“为什么 Agent 给了错误结果”的基础。6.3 评估用自动化测试替代肉眼判断大模型的不确定性决定了不能靠“看起来还行”来验收智能体。建议建立一个小型回归测试集准备 20 到 50 条典型用户请求。每一条标注期望的工具调用序列和最终回答要点。每次修改 prompt、升级模型或调整工具后跑一遍。记录通过率和失败案例。评估维度可以包括工具调用准确率是否调用了正确的工具。参数正确率工具入参是否解析正确。最终回答完整率关键信息是否完整返回。失败率是否出现解析错误或达到最大步数。成本变化平均 token 消耗是否明显增加。只有当评估集通过率稳定时模型升级才不是一场赌注。6.4 安全与权限Agent 一旦能调用工具就相当于有了执行能力。生产环境必须把工具函数的权限控制看得很重。工具层要做鉴权不是每个用户都能调用所有工具。敏感操作要二次确认比如删除、转账、发送消息等操作可以让 Agent 先返回“待确认操作”人工批准后再执行。工具调用要审计记录谁在什么时候调用了什么工具参数是什么。Prompt 注入要防御用户输入可能诱导模型调用危险工具。系统内部指令要高于用户输入工具输入参数要做白名单校验。不要把工具权限直接暴露给所有用户。Agent 越强大越需要权限边界。6.5 发布前检查清单以下清单可以直接复制到项目上线前检查是否配置了生产密钥且不会出现在日志里模型调用 API 是否有超时和重试工具函数是否设置了超时并返回结构化错误是否限制了max_steps避免循环调用失控是否设置了用户级限流和全局限流是否输出结构化日志并包含request_id是否建立了评估集并跑过至少一轮回归测试是否明确哪些工具属于高风险操作是否有模型版本回滚方案是否统计了单次请求的平均成本和 P95 延迟这些检查项不需要多复杂但每一项都可能在线上出大问题之前拦住风险。7. 扩展方向从单体智能体到多智能体协作7.1 多智能体解决什么问题当一个任务可以被拆成相互独立的子任务时多智能体协作才有价值。例如一个“竞品分析”需求可以让一个智能体负责搜集信息另一个智能体负责数据整理还有一个智能体负责撰写报告。每个智能体只负责一个方向prompt 可以更聚焦工具权限也可以更细。多智能体不是银弹。它的代价是引入额外的通信开销、任务编排复杂度和错误传播链条。如果单体智能体已经能满足需求不要为了架构好看强行拆分。7.2 通信与编排模式常见的多智能体编排模式有三种模式工作方式适用场景串联A 处理完后把结果交给 B流水线式任务并联多个 Agent 并行处理最后汇总独立子任务主从主 Agent 负责拆解子 Agent 执行复杂任务实现上每个 Agent 仍然是“模型 工具 记忆”的闭环。区别在于一个 Agent 可以把结果作为另一个 Agent 的输入或者由主 Agent 决定任务派发策略。这里不需要引入厚重框架。先用函数调用把多个run_agent串起来验证业务价值后再考虑是否引入多智能体框架。7.3 什么时候不该用多智能体以下情况不建议使用多智能体子任务之间强依赖必须频繁交换中间状态。每个子任务只需要一步工具调用拆分后反而增加延迟。团队没有完整的日志和 trace 体系出问题无法定位。成本预算紧张多轮模型调用会显著提高 token 消耗。先把单体智能体做成一个稳定可观测的产品再扩展成多智能体是最稳妥的路径。智能体产品化的重点不是在于模型有多强而是在于围绕模型搭建一套“可执行、可观测、可评估、可控制”的工程系统。这个系统包含工具层、记忆层、循环控制、权限边界和日志追溯。本文给出的最小 Python 智能体可以用几十行代码跑通完整闭环但真正让它成为产品的是后续不断补全的稳定性能力。如果你正在做智能体相关产品先把“用户输入 → 模型决策 → 工具执行 → 最终回答”这条链路打牢加上结构化日志和回归测试再逐步扩展记忆和并发能力会比直接套一个大而全框架更可控。
返回列表