引言：从“一团乱麻”到“各司其职”

当你第一次用几行代码调用大模型API做出了一个能回答问题的Demo时，那种兴奋感是难以言喻的。然而，当这个Demo要变成真正的产品——要接入企业知识库、要支持多轮对话、要调用外部工具、要面对高并发请求时，你可能会发现代码变得越来越难以维护。

典型的“坏味道”是这样的：在一个Python文件里，LLM调用、工具集成、业务逻辑和编排代码混杂在一起。一个函数既处理了HTTP请求，又拼接了Prompt，还调用了外部API，最后直接操作了数据库。这种写法能跑，但架构上是错的。

为什么？因为大模型的非确定性让它天生不稳定，而业务系统对稳定性的要求是刚性的。 分层架构的核心价值，就在于用工程化的确定性，去“驯服”模型的不确定性。本文将从实战出发，探讨如何通过分层设计，将业务逻辑与模型调用彻底解耦。

一、为什么必须分层？——单体架构的代价

让我们先看一个反例。这是一个典型的多智能体研究系统的简化版，但它把“所有东西都混在一起”的毛病展现得淋漓尽致：

# orchestrator.py - 智能体、工具、提示词和业务逻辑全部在一起
def run_research(query: str) -> str:
    # 搜索智能体，工具定义在行内
    def search_youtube(q: str) -> str:
        response = requests.get(f"https://youtube.com/results?q={q}")
        return parse_html_for_videos(response.text)

    search_agent = ChatAgent(
        name="SearchAgent",
        instructions="""You search YouTube. Use search_youtube to find videos.""",
        tools=[search_youtube]
    )

    # 字幕智能体，有自己行内的工具
    def get_transcript(video_id: str) -> str:
        transcript = YouTubeTranscriptApi.get_transcript(video_id)
        return " ".join([t["text"] for t in transcript])

    transcript_agent = ChatAgent(
        name="TranscriptAgent",
        instructions="Fetch transcripts using get_transcript tool.",
        tools=[get_transcript]
    )

    # 摘要智能体，提示工程嵌入其中
    summarize_agent = ChatAgent(
        name="SummarizeAgent",
        instructions="Summarize cooking content..."
    )

    # 编排逻辑与智能体调用交织
    videos = search_agent.run(query)
    transcripts = []
    for vid in parse_json(videos)[:3]:
        text = transcript_agent.run(f"Get transcript for {vid['id']}")
        transcripts.append(text)

    summary = summarize_agent.run(f"Summarize:\n{transcripts}")
    Path(f"./outputs/{query}.md").write_text(summary)
    return summary

这段代码的问题是：智能体配置、工具函数、提示词模板和业务流程全部缠在一起。如果要修改YouTube的解析逻辑，你得在这段代码里翻找；如果要换一个摘要风格，你得修改summarize_agent的指令；如果要给搜索添加重试机制，你得在search_youtube里硬编码。这还只是一个简化的例子，真实系统会更糟。

分层架构正是为了解决这个问题。通过合理划分各层职责，可以显著提升系统的可维护性、可扩展性和可测试性。

二、解耦的核心原则：工具 vs. 服务

在智能体系统中，LLM调用工具其实是在做两件完全不同的事：

  1. LLM的视角：用简单参数（字符串、数字）调用一个函数，然后解释返回的字符串结果。
  2. 应用的视角：实际干活的部分——搜索、解析、处理错误，涉及配置、重试，返回的是结构化对象。

这两件事是不同的关注点。LLM要的是简单字符串，应用要的是合理的抽象。把它们搅在一起，就像把SQL查询直接写在视图层：能跑，但架构上是错的。

核心洞见：工具 = LLM接口，服务 = 业务逻辑。

工具层：薄薄的适配层

工具是LLM和应用之间的薄适配层，它的职责是：

  • 接受简单参数（字符串、数字、布尔值）
  • 调用对应的服务
  • 把结果格式化成LLM能理解的字符串

工具本身是无状态的，不包含配置管理，不处理复杂返回类型，不包含业务逻辑。它只干一件事：调用服务、格式化结果。

# tools/youtube.py - 工具层（薄LLM适配器）
async def fetch_video_transcript(
    video_id: Annotated[str, Field(description="YouTube video ID")]
) -> str:
    """获取YouTube视频字幕，返回LLM可读的字符串格式"""
    # 调用服务层（真正干活的）
    result = await transcript_service.fetch(video_id)
    
    # 格式化为LLM能理解的字符串
    return f"Transcript for '{result.metadata.title}':\n\n{result.transcript.full_text}"

服务层：真正的业务逻辑

服务才是真正的“干活的人”。它们是带配置的可复用类，返回丰富的领域对象，可以从任何地方调用（CLI、测试、其他服务），可能维护状态或连接。

# services/youtube.py - 服务层（业务逻辑）
class YouTubeTranscriptFetcher:
    """从YouTube获取字幕——真正干活的"""
    
    def __init__(self, proxy_url: str | None = None):
        self.proxy_url = proxy_url

    async def fetch(
        self,
        video_id: str,
        languages: list[str] | None = None
    ) -> TranscriptResult:
        """获取字幕，返回完整元数据"""
        # 真正的实现：错误处理、重试、缓存……
        raw_transcript = await self._fetch_from_api(video_id, languages)
        metadata = await self._fetch_metadata(video_id)

        return TranscriptResult(
            metadata=metadata,
            transcript=Transcript(
                full_text=self._format_transcript(raw_transcript),
                segments=raw_transcript,
                language=self._detect_language(raw_transcript),
            ),
        )

为什么这样分离很重要？

  1. 可复用性：服务可以直接从CLI、测试脚本、批处理调用，完全绕过LLM。
  2. 可测试性：服务返回类型化对象，断言清晰；工具返回格式化字符串，验证费劲。
  3. 关注点分离：YouTube API改了？只改services/youtube.py。想换输出格式？只改工具。

三、完整的六层架构实践

工具和服务的分离只是一条边界。一个完整的生产级智能体系统需要更多结构。综合多方实践，一个成熟的分层架构通常包含以下六层：

┌─────────────────────────────────────────────────────┐
│  Presentation Layer (表示层)                        │  ← CLI / API网关
├─────────────────────────────────────────────────────┤
│  Agent Layer (智能体层)                            │  ← 仅配置行为
├─────────────────────────────────────────────────────┤
│  Tool Layer (工具层)                               │  ← LLM适配器
├─────────────────────────────────────────────────────┤
│  Service Layer (服务层)                            │  ← 业务逻辑
├─────────────────────────────────────────────────────┤
│  Model/Domain Layer (领域模型层)                   │  ← 数据结构
├─────────────────────────────────────────────────────┤
│  Infrastructure Layer (基础设施层)                  │  ← HTTP/数据库
└─────────────────────────────────────────────────────┘

3.1 各层职责说明

表示层（Presentation Layer）：系统的入口。可以是CLI命令、RESTful API网关或gRPC接口。只负责接收输入、调用下层、返回输出，不包含业务逻辑。在API网关层还可以统一处理鉴权、限流和内容风控。

智能体层（Agent Layer）：仅负责配置智能体的行为。它是一个工厂函数，将模型、工具列表和指令组合成一个可运行的Agent。这一层不包含工具的实现细节，也不包含业务逻辑。

工具层（Tool Layer）：前文详述的LLM适配层。每个工具接受简单参数，调用服务层，返回字符串。

服务层（Service Layer）：真正的业务逻辑所在。包含配置、缓存、错误处理、重试，返回结构化的领域对象。

领域模型层（Model/Domain Layer）：定义核心的数据结构和领域对象。例如VideoResult、TranscriptResult等。这些对象在整个系统中流转，确保数据的一致性和类型安全。

基础设施层（Infrastructure Layer）：最底层，负责与外部系统交互。HTTP客户端、数据库连接池、消息队列等都属于这一层。

3.2 代码落地：一个完整示例

领域模型层：

# models/youtube.py
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class VideoMetadata:
    video_id: str
    title: str
    channel: str
    duration: int

@dataclass
class Transcript:
    full_text: str
    segments: List[dict]
    language: str

@dataclass
class TranscriptResult:
    metadata: VideoMetadata
    transcript: Transcript

基础设施层：

# infra/http_client.py
import httpx

async def fetch_html(url: str, timeout: float = 10.0) -> str:
    """获取HTML内容，带超时和重试"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url, 
            headers={"User-Agent": "Mozilla/5.0"},
            timeout=timeout
        )
        response.raise_for_status()
        return response.text

服务层：

# services/youtube.py
from models.youtube import VideoMetadata, Transcript, TranscriptResult
from infra.http_client import fetch_html

class YouTubeTranscriptFetcher:
    def __init__(self, proxy_url: str | None = None):
        self.proxy_url = proxy_url
    
    async def fetch(self, video_id: str) -> TranscriptResult:
        # 真正的业务逻辑：调用API、处理错误、构造领域对象
        raw = await self._fetch_from_api(video_id)
        metadata = await self._fetch_metadata(video_id)
        return TranscriptResult(
            metadata=metadata,
            transcript=Transcript(
                full_text=self._format(raw),
                segments=raw,
                language=self._detect_language(raw)
            )
        )

工具层：

# tools/youtube.py
from typing import Annotated
from pydantic import Field

async def fetch_video_transcript(
    video_id: Annotated[str, Field(description="YouTube视频ID")]
) -> str:
    """获取YouTube视频字幕，供LLM调用"""
    fetcher = YouTubeTranscriptFetcher()
    result = await fetcher.fetch(video_id)
    return f"标题: {result.metadata.title}\n\n字幕:\n{result.transcript.full_text}"

智能体层：

# agents/search.py
def create_research_agent() -> ChatAgent:
    """工厂函数：创建研究智能体"""
    return ChatAgent(
        name="ResearchAgent",
        instructions=SEARCH_AGENT_PROMPT,  # Prompt作为常量管理
        tools=[fetch_video_transcript, search_youtube_formatted]
    )

表示层（CLI）：

# presentation/cli.py
@click.command()
@click.argument("query")
def search(query: str):
    agent = create_research_agent()
    result = agent.run(query)
    click.echo(result)

这个分层设计的好处是：每一层都可以独立修改和测试。当YouTube API变更时，只需要修改services/youtube.py；当想调整LLM输出格式时，只需要修改工具层；当想换一个Agent框架时，只需要修改智能体层。

四、进阶：API聚合层与模型路由

在更复杂的企业级场景中，业务系统可能需要调用多个不同的大模型（GPT-4、Claude、国产模型等）。此时可以在表示层和服务层之间引入一个LLM API聚合层。

这一层的核心价值是：

  • 统一协议：屏蔽不同模型提供商的API差异，提供统一的OpenAI风格接口
  • 模型路由：根据任务复杂度动态分发请求——复杂逻辑用GPT-4，简单问答用轻量模型，最高可节省90%的API费用
  • 成本控制：统一监控各模型的Token消耗
# services/llm_gateway.py
class LLMGateway:
    def __init__(self):
        self.routers = {
            "complex": OpenAIClient("gpt-4-turbo"),
            "simple": OpenAIClient("gpt-3.5-turbo"),
            "local": LocalLLMClient("qwen2.5:7b")
        }
    
    async def chat(self, messages: list, task_type: str = "simple"):
        # 根据任务类型路由到不同模型
        if task_type == "complex":
            client = self.routers["complex"]
        elif self._is_local_only(messages):
            client = self.routers["local"]
        else:
            client = self.routers["simple"]
        return await client.chat(messages)

结语

大模型应用的分层架构，本质上是用软件工程中沉淀了几十年的最佳实践，去应对AI带来的新挑战。

工具与服务的分离、表现层与逻辑层的解耦、API聚合与模型路由——这些概念在传统后端开发中早已存在，但在AI场景中有了新的表现形式和重要性。分层设计让系统能够独立演进各层能力，使AI应用从“能跑的Demo”成长为“可维护、可测试、可扩展的生产系统”。

核心启示：不要被新框架层出不穷的表象迷惑。无论技术如何迭代，“关注点分离”和“单一职责”这两个原则始终有效。当你的AI系统代码能够清晰地划分边界时，你才真正从“调API的人”变成了“架构师”。

# 架构 # 人工智能 # 大数据 # golang
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

一个 API 入口调用多个大模型：AiiOnly客户端 + CC Switch 打通 Codex AI编程流程

avatar AI编程社区
cover

VSCode Claude Code + MiMo 中转免登录完整配置教程（解决原生登录弹窗 + 模型报错）

avatar AI编程社区
cover

从零开始做一个属于自己的 Skill（保姆级教程）

avatar AI编程社区