拒绝“玩具”Demo:如何构建生产级 Agent 系统并落地?

大家好,我是你们的老朋友,一名在代码堆里摸爬滚打多年的程序员。

最近 LLM(大语言模型)的热度居高不下,很多团队都在尝试接入 AI 能力。但我发现一个普遍现象:大多数项目还停留在“把 LLM 调通”的 Demo 阶段。

一旦试图将其应用到真实的业务场景(比如医疗、金融、企业内网),问题就接踵而至:回答幻觉严重、响应极慢、Token 费用爆炸、甚至因为一次错误的工具调用导致数据事故。

今天,我想和大家深入聊聊:如何自建一个 Agent 系统,并将其做到真正的“生产级”落地?

一句话核心:

企业级 Agent 系统,本质上不是“把 LLM 调通”,而是**“围绕 LLM 构建完整的软件工程体系”**。

这包括架构设计、工具调用治理、RAG 优化、记忆管理、工作流编排以及可观测性。下面我们将拆解这六大关键步骤。


一、宏观视角:生产级 Agent 的整体架构

很多初学者容易把 Agent 想象成一个简单的 while True 循环。但在生产环境中,我们需要分层解耦。

通常我会将架构拆分为以下层级:

支撑体系

用户层

Agent Layer / 路由与决策

Workflow / Orchestrator / 编排引擎

Tools / RAG / Memory / 能力层

Model Layer / LLM 基座

Observability & Evaluation / 监控与评估

Guardrails / 安全护栏

这个架构的核心在于:LLM 只是大脑,而周围的工程体系才是身体。


二、第一步:明确 Agent 边界(职责分离)

很多项目失败的根本原因是:Agent 职责不清

如果你试图让一个 Agent 同时负责“理解意图”、“查询数据库”、“撰写报告”和“检查错误”,它的上下文会迅速混乱,成功率大幅下降。

在生产环境中,我们通常采用多 Agent 协作模式,明确定义每个角色的边界:

Agent 角色 核心职责 典型输入/输出
Router Agent 流量分发,判断用户意图 用户Query -> 目标 Agent ID
Knowledge Agent 专注知识检索与问答 问题 -> RAG 结果 + 答案
Data Agent 结构化数据分析 SQL/Python Code -> 图表/结论
Report Agent 长文本生成与格式化 数据片段 -> Markdown/PDF报告
QA Agent 最终输出质量检查 草稿 -> 修正后的最终稿

最佳实践: 不要迷信“全能型 Agent”。小而专的 Agent 组合,远比一个大而全的 Agent 稳定。


三、第二步:设计健壮的 Tool Calling 体系

生产级的 Agent 一定不是纯聊天,其核心价值在于 Tool Use(工具使用)

无论是查数据库、调 HIS/LIS 接口,还是发送 Webhook,都需要一个统一的 Tool Registry(工具注册中心) 来管理。

为什么需要 Tool Registry?

如果直接在 Prompt 里硬编码工具描述,一旦工具参数变更,整个系统都要重新调试。我们需要统一管理:

  1. Tool Schema:标准化的 JSON Schema 描述。
  2. 权限控制:谁可以调用这个工具?
  3. 超时与重试:网络波动时的兜底策略。
  4. 参数校验:在发给 LLM 之前或之后,强制校验参数合法性。

代码示例:一个简单的工具装饰器

我们可以用 Python 装饰器来简化工具的注册和元数据管理:

import json
from typing import Dict, Any

class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def register(self, name: str, description: str, params_schema: Dict):
        def decorator(func):
            self.tools[name] = {
                "function": func,
                "schema": {
                    "name": name,
                    "description": description,
                    "parameters": params_schema
                }
            }
            return func
        return decorator

    def execute(self, tool_name: str, args: Dict[str, Any]) -> Any:
        if tool_name not in self.tools:
            raise ValueError(f"Tool {tool_name} not found")
        
        # 这里可以加入日志、权限校验、超时控制等逻辑
        print(f"[LOG] Executing tool: {tool_name} with args: {args}")
        return self.tools[tool_name]["function"](**args)

# 初始化工具注册中心
registry = ToolRegistry()

# 注册一个查询患者信息的工具
@registry.register(
    name="get_patient_info",
    description="根据患者ID获取基本信息",
    params_schema={
        "type": "object",
        "properties": {
            "patient_id": {"type": "string", "description": "患者的唯一标识ID"}
        },
        "required": ["patient_id"]
    }
)
def get_patient_info(patient_id: str) -> str:
    # 模拟数据库查询
    return json.dumps({"id": patient_id, "name": "张三", "age": 30})

# 执行工具
result = registry.execute("get_patient_info", {"patient_id": "P1001"})
print(result)

四、第三步:构建高精度的 RAG 系统

在企业场景(如医疗、法律、金融)中,没有 RAG(检索增强生成),Agent 几乎不可落地。因为模型参数里的知识是滞后的,且容易产生幻觉。

但普通的 RAG 往往效果不佳,生产级 RAG 需要关注以下三个关键点:

1. Chunk Strategy(分块策略)

不要简单地按固定字符数切割!

  • 推荐做法:使用 Parent-Child Chunking
    • Parent Chunk:较大的段落,保留完整语境。
    • Child Chunk:较小的片段,用于向量检索匹配。
    • 原理:检索时命中 Child,但返回给 LLM 的是其对应的 Parent,从而保证上下文完整性。

2. Hybrid Search(混合检索)

企业数据中充满了专有名词(如药品名、检验指标缩写)。纯向量检索对关键词不敏感,纯关键词检索无法理解语义。

  • 推荐做法BM25(关键词) + Vector(向量)
  • 使用 Reciprocal Rank Fusion (RRF) 算法合并两路搜索结果。

3. Rerank(重排序)

这是提升精度的“杀手锏”。

  • 向量检索召回 Top 50 个文档片段。
  • 使用 Cross Encoder 模型(如 BGE-Reranker)对这 50 个片段与问题进行相关性打分。
  • 取 Top 5 高分片段注入 Prompt。

原始文档

智能分块

Embedding

向量数据库

用户提问

混合检索

关键词索引

候选集 Top 50

Rerank 重排序模型

最终上下文 Top 5

大模型生成


五、第四步:上下文与 Memory 治理

生产系统最大的敌人是:Token 爆炸信息污染

如果不加治理,随着对话轮数增加,Prompt 会越来越长,不仅贵,而且会让模型“注意力分散”,忽略关键指令。

我们需要对 Memory 进行分层治理:

记忆类型 作用 实现方式
Short-term 当前会话的近期交互 滑动窗口(Sliding Window),只保留最近 N 轮
Long-term 用户画像、偏好 存入向量库或 KV 存储,按需检索
Episodic 历史重要事件摘要 定期调用 LLM 对旧对话进行总结压缩
Semantic 领域长期知识 固化在 System Prompt 或 RAG 知识库中

常见做法:Conversation Summary

当对话超过一定长度时,触发后台任务,将早期的对话压缩成一段摘要:

def compress_history(messages: list) -> str:
    """
    将历史消息压缩为摘要
    """
    summary_prompt = f"请总结以下对话的关键信息和用户意图:\n{messages}"
    # 调用 LLM 生成摘要
    summary = llm.invoke(summary_prompt)
    return summary.content

核心原则:不是把所有历史都塞进 Prompt,而是按需召回(Memory Retrieval)


六、第五步:Workflow 编排(从线性到图状)

简单的问答可以用 AgentExecutor 线性执行。但复杂的业务流程(如:先查数据 -> 再分析 -> 如果异常则报警 -> 最后写报告)需要更强大的编排能力。

这里强烈推荐使用 LangGraph 或类似的状态机框架,而不是简单的 Chain。

为什么选择 LangGraph?

能力 简单 AgentExecutor LangGraph (State Graph)
状态管理 弱,难以追踪中间状态 ,显式维护 State
循环与分支 支持有限 原生支持 DAG 和循环
并行执行 困难 原生支持 并行节点
Human-in-the-loop 难以实现 原生支持 中断与人工介入
容错与恢复 ,可从断点继续执行

场景示例:带人工审核的报告生成流程

需要修改

通过

GenerateDraft

HumanReview

Edit

Approve

SendReport

在这种流程中,Agent 生成初稿后,流程暂停,等待人类专家在 UI 上确认或修改。确认后,流程继续。这种“人机协同”是生产级落地的标配。


七、总结与建议

构建生产级 Agent 系统,是一场从“魔法”到“工程”的修行。

  1. 架构先行:拆分 Router、Worker、QA 等角色,不要试图用一个 Prompt 解决所有问题。
  2. 工具治理:建立统一的 Tool Registry,做好权限、日志和异常处理。
  3. RAG 精细化:混合检索 + Rerank 是提升准确率的必经之路。
  4. 记忆瘦身:定期摘要,按需召回,控制 Token 成本。
  5. 流程编排:使用 LangGraph 等工具处理复杂逻辑和人机协同。
  6. 可观测性:必须记录每一次 LLM 的输入输出、Token 消耗和延迟,否则无法优化。

最后的一句忠告:
不要为了用 Agent 而用 Agent。如果一个简单的规则引擎或搜索功能能解决问题,那就不要用 LLM。Agent 的价值在于处理非结构化、不确定性和需要推理的复杂任务。

希望这篇文章能为你构建企业级 Agent 系统提供清晰的路径。如果有具体的技术细节想探讨,欢迎在评论区留言!


参考资料

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐