大模型应用开发全家桶：LLM+RAG+Agent+MCP+工作流双技术栈实践

大模型应用从 Demo 到落地，最缺的不是模型 API，而是一套能串起 LLM、RAG、Agent、MCP、工作流和工具调用的工程骨架。最近看到很多开发者都在问同一个问题：项目里既要接大模型，又要做知识库问答，还要让模型学会调用内部工具，甚至要支持 FastAPI 和 Spring Boot 两种技术栈，到底该怎么组织代码，才不会越写越乱？

这篇文章准备把一条经过验证的 AI 应用开发主线完整拆开讲清楚。它不绑定某个特定厂商模型，也不局限于单一编程语言，而是围绕 LLM、RAG、Agent、MCP、工作流、工具调用这六个核心模块，分别给出 FastAPI 和 Spring Boot 双版本的设计思路与示例代码。读完后你可以得到一套可直接落地参考的“AI 大模型应用开发全家桶”，知道每个模块解决什么问题、放在项目的哪一层、关键接口怎么定义，以及遇到报错时先查哪里。

先说结论：这套全家桶的核心价值，是帮你把“模型能力”和“业务系统”解耦。模型负责理解和生成语言，业务系统负责权限、数据、事务和流程。如果你正在做知识库问答、智能客服、Copilot、自动化办公或内部流程助手，这篇文章值得收藏后仔细对照实践。

1. 这篇文章真正要解决的问题

过去一年里，AI 大模型应用开发最大的变化不是模型本身变强了，而是应用架构从“单次对话”快速走向“多模块协作”。早期我们写一个 LLM 调用，基本就是构造函数、传 Prompt、拿结果。但现在一个稍微完整一点的应用，往往同时包含以下需求：

• 接一个大语言模型，并支持切换不同厂商、不同模型。
• 把公司内部文档、数据库记录、网页内容做向量化，让模型基于私有知识回答。
• 让模型具备“自主规划”能力，能判断一个问题是否需要调用外部工具。
• 通过 MCP 协议统一暴露工具接口，避免每个应用单独写一套工具 SDK。
• 把多个模型调用、工具调用、条件判断组合成一个可复用、可监控的工作流。
• 同时服务 Java 团队和 Python 团队，让前后端或微服务能够各自以自己的语言接入。

这六个需求单独拎出来都不算难，难的是把它们放在同一个项目里，还要保证代码结构清晰、可测试、可维护。很多人一开始会把所有逻辑都写进一个 service 方法：先调 LLM，再查向量库，再根据结果调工具，最后拼 Prompt 再调一次。表面上看功能能跑，但一旦模型返回格式变化、第三方 API 超时、或者需要增加新的工具，整个方法就会迅速膨胀，最终变成无法维护的“面条代码”。

所以，这篇文章真正要解决的，不是“怎么调一次 API”，而是“如何从零搭建一套可扩展的 AI 应用工程骨架”。它强调几个原则：

1. 每个大模型能力模块都是独立组件，可以单独测试。
2. 对外暴露统一接口，内部实现可替换。
3. Python 版和 Java 版保持相同的模块划分，方便团队跨语言协作。
4. 先跑通最小闭环，再逐步接入 RAG、Agent、MCP 和工作流。

如果你是刚进入 AI 应用开发的新手，可以按文章顺序逐步实现；如果你已经做过简单 LLM 调用，可以直接跳到你最关心的模块，比如 RAG 或 MCP。无论哪种情况，这套全家桶的设计思路都能帮你避免“代码堆砌”的坑。

2. 全家桶全景：六大模块的定位与关系

在动手写代码之前，先用一张概念图理解整套系统。这里不使用复杂图例，而是用表格列出每个模块的核心职责、输入输出和典型场景。

它们的关系可以这样理解：LLM 是所有模块的“大脑底座”；RAG 为大脑提供外部知识；Agent 让大脑学会拆解任务；MCP 为 Agent 提供统一工具访问协议；工作流则把一组可重复的调用编排成标准化流程；工具调用是真正落地执行的动作。

很多初学者会混淆 Agent 和工作流。Agent 偏“动态规划”，每一步都可能不同，模型自己决定下一步调什么；工作流偏“静态编排”，流程是预先画好的，每一步固定执行。真实项目里经常两者结合：流程负责兜底和标准化部分，Agent 负责需要灵活判断的部分。

在架构分层上，我建议至少分四层：

• 接口层：Controller/API 路由，暴露 HTTP 接口。
• 应用层：编排 LLM、RAG、Agent、工作流等核心服务。
• 领域层：定义模型请求、消息、文档、工具、Agent 状态等对象。
• 基础层：封装 HTTP 客户端、向量存储、数据库、日志等。

FastAPI 版天然适合这种分层，Python 类型提示配合 Pydantic 让数据模型很清晰。Spring Boot 版则借助 Spring 的 IoC 容器把各个 Bean 注入到应用层。两条技术栈虽然语法不同，但模块边界可以一一对应，这也是这套全家桶能“双版本”落地的关键。

3. 环境准备与前置条件

在开始搭建全家桶前，先把环境确认好。这里不写死具体版本号，因为 AI 工具链更新非常快，你只需要保证下面这些组件在本地能用即可。

• 操作系统：Windows/macOS/Linux 都可以。
• Python 版本：建议 3.10 及以上，便于使用较新的类型语法。
• Java 版本：建议 17 及以上，许多 Spring Boot 3 的依赖要求 17 起步。
• 包管理器：Python 使用 pip 或 poetry；Java 使用 Maven 或 Gradle。
• 模型 API：准备一个可用的 LLM API Key，可以是 OpenAI 兼容格式的，也可以是国内大模型服务商提供的 Key。
• 向量数据库：可以选择 Chroma、Milvus、Qdrant 或 Elasticsearch，先从本地轻量方案开始。
• 开发工具：VS Code 配合 Python 插件，或 IntelliJ IDEA 配合 Spring 插件。

这里真正容易踩坑的地方是你对“模型 API”的选择。很多初学者以为只能用一个固定厂商的 SDK，实际现在绝大多数模型服务都提供 OpenAI 兼容接口，也就是说只要你的代码实现了 OpenAI Chat Completion 格式，就能适配大部分平台。因此，建议在全家桶里封装一个 LLMClient 接口，底层用 HTTP 方式调用，而不是在业务代码里绑定某个厂商 SDK。

另外，本地要做 RAG，还需要准备一些示例文档。可以用一本开源电子书、几份 Markdown 说明文档或者数据库导出的文本。核心目的是验证“检索—增强—生成”的链路是否跑通，不必追求大规模数据集。

4. FastAPI 版基础骨架：LLM 接入与统一接口

先用 FastAPI 实现最小闭环：调用一个 OpenAI 兼容的大模型，完成一次对话。这样做的好处是先把网络层、配置层和数据模型层铺好，后续所有模块都复用这套基础。

4.1 项目结构建议

code复制ai-family-python/
├── app/
│   ├── api/
│   │   └── chat.py
│   ├── core/
│   │   └── config.py
│   ├── domain/
│   │   ├── message.py
│   │   └── response.py
│   ├── infra/
│   │   ├── llm/
│   │   │   ├── base.py
│   │   │   └── openai_client.py
│   ├── application/
│   │   └── chat_service.py
│   └── main.py
├── requirements.txt
└── .env

这个结构把领域层、基础设施层、应用层分开。domain 目录放消息和返回结构，infra 目录放模型厂商实现，application 目录放业务编排逻辑，api 目录只负责 HTTP 暴露。

4.2 配置文件

env复制# .env
OPENAI_API_KEY=sk-your-key
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

python复制# app/core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    openai_api_key: str
    openai_base_url: str = "https://api.openai.com/v1"
    openai_model: str = "gpt-4o-mini"
    embedding_model: str = "text-embedding-3-small"
    vector_store_path: str = "./data/chroma"

    class Config:
        env_file = ".env"

settings = Settings()

4.3 统一消息模型

python复制# app/domain/message.py
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field

class ChatMessage(BaseModel):
    role: str
    content: str
    name: Optional[str] = None
    function_call: Optional[Dict[str, Any]] = None

class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    temperature: float = Field(default=0.7, ge=0, le=2)
    max_tokens: Optional[int] = None
    stream: bool = False

class ChatResponse(BaseModel):
    message: ChatMessage
    usage: Optional[Dict[str, int]] = None

这里定义了两个核心模型。ChatMessage 表示对话消息，ChatRequest 表示请求体，ChatResponse 统一返回给前端。把消息模型独立出来，后续 Agent 和 RAG 才能复用。

4.4 抽象 LLM 客户端

python复制# app/infra/llm/base.py
from abc import ABC, abstractmethod
from typing import List

from app.domain.message import ChatMessage, ChatResponse

class BaseLLMClient(ABC):
    @abstractmethod
    async def chat(self, messages: List[ChatMessage], temperature: float = 0.7,
                   max_tokens: int = None, stream: bool = False) -> ChatResponse:
        pass

python复制# app/infra/llm/openai_client.py
import httpx
from app.core.config import settings
from app.domain.message import ChatMessage, ChatResponse
from app.infra.llm.base import BaseLLMClient

class OpenAIClient(BaseLLMClient):
    def __init__(self):
        self.api_key = settings.openai_api_key
        self.base_url = settings.openai_base_url
        self.model = settings.openai_model

    async def chat(self, messages, temperature=0.7, max_tokens=None, stream=False):
        payload = {
            "model": self.model,
            "messages": [msg.model_dump(exclude_none=True) for msg in messages],
            "temperature": temperature,
        }
        if max_tokens:
            payload["max_tokens"] = max_tokens
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        async with httpx.AsyncClient(base_url=self.base_url) as client:
            resp = await client.post("/chat/completions", json=payload, headers=headers)
            resp.raise_for_status()
            data = resp.json()
        choice = data["choices"][0]["message"]
        return ChatResponse(
            message=ChatMessage(role=choice["role"], content=choice.get("content")),
            usage=data.get("usage"),
        )

BaseLLMClient 是一个抽象接口。以后要接入其他模型，只要实现这个接口即可，应用层完全不需要改动。httpx.AsyncClient 是异步 HTTP 客户端，适合 FastAPI 这类异步框架。

4.5 应用层服务

python复制# app/application/chat_service.py
from typing import List
from app.domain.message import ChatMessage, ChatResponse
from app.infra.llm.base import BaseLLMClient

class ChatService:
    def __init__(self, llm_client: BaseLLMClient):
        self.llm_client = llm_client

    async def chat(self, messages: List[ChatMessage], temperature: float = 0.7):
        return await self.llm_client.chat(messages, temperature=temperature)

4.6 API 路由和启动入口

python复制# app/api/chat.py
from fastapi import APIRouter, Depends

from app.domain.message import ChatRequest, ChatResponse, ChatMessage
from app.application.chat_service import ChatService
from app.infra.llm.openai_client import OpenAIClient

router = APIRouter(prefix="/chat", tags=["chat"])

def get_chat_service():
    return ChatService(OpenAIClient())

@router.post("/completions", response_model=ChatResponse)
async def chat_completions(request: ChatRequest, service: ChatService = Depends(get_chat_service)):
    response = await service.chat(request.messages, request.temperature)
    return response

python复制# app/main.py
from fastapi import FastAPI
from app.api.chat import router as chat_router

app = FastAPI(title="AI Family Python")
app.include_router(chat_router)

@app.get("/health")
async def health():
    return {"status": "ok"}

bash复制pip install fastapi uvicorn pydantic-settings httpx python-dotenv
uvicorn app.main:app --reload --port 8000

验证方式：

bash复制curl -X POST http://localhost:8000/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"用一句话介绍什么是RAG"}]}'

如果一切正常，你会得到一个包含 message.content 和 usage 的 JSON。到这里，全家桶的第一个模块已经跑通。后面所有模块都会围绕这个 ChatService 扩展。

5. Spring Boot 版基础骨架：LLM 接入与统一接口

Java 团队同样需要这套统一抽象。Spring Boot 的做法和 FastAPI 类似，只是把依赖注入交给 Spring 管理。下面是一个完整的工程骨架。

5.1 项目依赖

xml复制<!-- pom.xml -->
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-validation</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-configuration-processor</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
</dependencies>

5.2 配置项

properties复制# src/main/resources/application.yml
ai:
  llm:
    api-key: sk-your-key
    base-url: https://api.openai.com/v1
    model: gpt-4o-mini

java复制// src/main/java/com/example/aifamily/config/LLMProperties.java
@Data
@ConfigurationProperties(prefix = "ai.llm")
public class LLMProperties {
    private String apiKey;
    private String baseUrl;
    private String model;
}

java复制// src/main/java/com/example/aifamily/config/AppConfig.java
@Configuration
@EnableConfigurationProperties(LLMProperties.class)
public class AppConfig {

    @Bean
    public WebClient llmWebClient(LLMProperties props) {
        return WebClient.builder()
                .baseUrl(props.getBaseUrl())
                .defaultHeader(HttpHeaders.CONTENT_TYPE, "application/json")
                .build();
    }
}

5.3 定义消息模型和请求响应

java复制// src/main/java/com/example/aifamily/domain/ChatMessage.java
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ChatMessage {
    private String role;
    private String content;
}

java复制// src/main/java/com/example/aifamily/domain/ChatRequest.java
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ChatRequest {
    private List<ChatMessage> messages;
    private Double temperature;
    private Integer maxTokens;
}

java复制// src/main/java/com/example/aifamily/domain/ChatResponse.java
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ChatResponse {
    private ChatMessage message;
    private Map<String, Integer> usage;
}

5.4 定义 LLM 抽象接口和 OpenAI 实现

java复制// src/main/java/com/example/aifamily/infra/llm/BaseLLMClient.java
public interface BaseLLMClient {
    Mono<ChatResponse> chat(List<ChatMessage> messages, Double temperature, Integer maxTokens);
}

java复制// src/main/java/com/example/aifamily/infra/llm/OpenAIClient.java
@Service
@RequiredArgsConstructor
public class OpenAIClient implements BaseLLMClient {

    private final WebClient webClient;
    private final LLMProperties props;

    @Override
    public Mono<ChatResponse> chat(List<ChatMessage> messages, Double temperature, Integer maxTokens) {
        Map<String, Object> payload = new HashMap<>();
        payload.put("model", props.getModel());
        payload.put("messages", messages);
        payload.put("temperature", temperature != null ? temperature : 0.7);
        if (maxTokens != null) {
            payload.put("max_tokens", maxTokens);
        }

        return webClient.post()
                .uri("/chat/completions")
                .header(HttpHeaders.AUTHORIZATION, "Bearer " + props.getApiKey())
                .bodyValue(payload)
                .retrieve()
                .bodyToMono(JsonNode.class)
                .map(data -> {
                    JsonNode choice = data.path("choices").get(0);
                    JsonNode msg = choice.path("message");
                    ChatResponse response = new ChatResponse();
                    response.setMessage(ChatMessage.builder()
                            .role(msg.path("role").asText())
                            .content(msg.path("content").asText())
                            .build());
                    if (data.has("usage")) {
                        response.setUsage(convertUsage(data.path("usage")));
                    }
                    return response;
                });
    }

    private Map<String, Integer> convertUsage(JsonNode usageNode) {
        Map<String, Integer> usage = new HashMap<>();
        usage.put("prompt_tokens", usageNode.path("prompt_tokens").asInt());
        usage.put("completion_tokens", usageNode.path("completion_tokens").asInt());
        usage.put("total_tokens", usageNode.path("total_tokens").asInt());
        return usage;
    }
}

5.5 应用层和 Controller

java复制// src/main/java/com/example/aifamily/application/ChatService.java
@Service
@RequiredArgsConstructor
public class ChatService {

    private final BaseLLMClient llmClient;

    public Mono<ChatResponse> chat(List<ChatMessage> messages, Double temperature, Integer maxTokens) {
        return llmClient.chat(messages, temperature, maxTokens);
    }
}

java复制// src/main/java/com/example/aifamily/api/ChatController.java
@RestController
@RequestMapping("/chat")
@RequiredArgsConstructor
public class ChatController {

    private final ChatService chatService;

    @PostMapping("/completions")
    public Mono<ChatResponse> completions(@RequestBody @Valid ChatRequest request) {
        return chatService.chat(request.getMessages(), request.getTemperature(), request.getMaxTokens());
    }
}

启动后同样访问 POST /chat/completions，请求体与 FastAPI 版保持一致。这个双版本实践表明：只要模块边界定义清楚，语言只是实现手段，架构思想可以复用。

6. RAG 检索增强生成：如何让模型“知道”私有知识

LLM 接入只是第一步。真实业务场景里，模型不可能知道你公司内部的规章制度、产品文档、历史工单。RAG 解决的核心问题，就是把外部知识注入到大模型生成过程中，而不是重新训练模型。RAG 的本质是“先检索、再回答”。

RAG 流程一般分两步：

1. 离线索引：将文档切分、嵌入向量、存入向量数据库。
2. 在线问答：将用户问题嵌入向量，检索相似片段，拼入 Prompt，再让模型生成。

这一节先给出 FastAPI 版的关键实现，再用 Spring Boot 版做一个简单对比。两者思路一致，差异主要在 API 风格和数据模型上。

6.1 文档切分与向量化

切分是 RAG 最容易忽略却影响最大的环节。切得太粗，上下文会混入无关内容；切得太细，语义不完整。常见策略是：按章节标题切分，设定每块 300 到 500 字，块与块之间保留 50 字重叠。重叠是为了避免关键信息刚好被切断。

python复制# app/infra/rag/doc_processor.py
from typing import List
from langchain_text_splitters import RecursiveCharacterTextSplitter

def split_document(content: str, chunk_size: int = 500, chunk_overlap: int = 50) -> List[str]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "；", "，"]
    )
    return splitter.split_text(content)

6.2 向量存储接口

为了不绑定具体向量数据库，抽象一个 VectorStore 接口，提供 add_documents 和 similarity_search 方法。这样你可以先本地用 Chroma，后续再换 Milvus 或 Qdrant。

python复制# app/infra/vector/base.py
from abc import ABC, abstractmethod
from typing import List

class VectorStore(ABC):
    @abstractmethod
    def add_texts(self, texts: List[str], metadatas: List[dict] = None):
        pass

    @abstractmethod
    def similarity_search(self, query: str, k: int = 4) -> List[dict]:
        pass

python复制# app/infra/vector/chroma_store.py
import chromadb
from app.core.config import settings
from app.infra.vector.base import VectorStore

class ChromaStore(VectorStore):
    def __init__(self):
        self.client = chromadb.PersistentClient(path=settings.vector_store_path)
        self.collection = self.client.get_or_create_collection("docs")

    def add_texts(self, texts, metadatas=None):
        ids = [f"doc_{i}" for i in range(len(texts))]
        self.collection.upsert(
            documents=texts,
            ids=ids,
            metadatas=metadatas or [{} for _ in texts]
        )

    def similarity_search(self, query, k=4):
        result = self.collection.query(query_texts=[query], n_results=k)
        docs = result.get("documents", [[]])[0]
        metas = result.get("metadatas", [[]])[0]
        return [{"content": doc, "metadata": meta or {}} for doc, meta in zip(docs, metas)]

注意，upsert 的效果是如果 doc_{i} 已存在则覆盖。生产环境建议用文档哈希或唯一业务 ID 作为主键，否则重复索引会污染结果。

6.3 RAG 服务：检索 + 拼接 + 生成

python复制# app/application/rag_service.py
from app.domain.message import ChatMessage, ChatResponse
from app.infra.llm.base import BaseLLMClient
from app.infra.vector.base import VectorStore

class RagService:
    def __init__(self, llm_client: BaseLLMClient, vector_store: VectorStore):
        self.llm_client = llm_client
        self.vector_store = vector_store

    def build_prompt(self, query: str, contexts: List[str]) -> str:
        context_text = "\n\n".join([f"[片段{i+1}] {c}" for i, c in enumerate(contexts)])
        prompt = f"""请根据以下知识片段回答用户问题。
如果知识片段中没有答案，请直接说明“根据现有资料无法回答”，不要编造。

知识片段：
{context_text}

用户问题：{query}
"""
        return prompt

    async def ask(self, query: str, k: int = 4):
        docs = self.vector_store.similarity_search(query, k=k)
        contexts = [doc["content"] for doc in docs]
        prompt = self.build_prompt(query, contexts)
        messages = [ChatMessage(role="system", content="你是企业知识库助手。"),
                    ChatMessage(role="user", content=prompt)]
        response = await self.llm_client.chat(messages, temperature=0.3)
        return {
            "answer": response.message.content,
            "sources": contexts,
        }

这里把温度设置为 0.3，让模型更多依赖检索内容而不是自由发挥，减少幻觉。返回的 sources 可以帮助前端展示引用来源，这也是 RAG 在交付时很重要的一环。

6.4 新增 RAG 路由

python复制# app/api/rag.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel

from app.application.rag_service import RagService
from app.infra.llm.openai_client import OpenAIClient
from app.infra.vector.chroma_store import ChromaStore

router = APIRouter(prefix="/rag", tags=["rag"])

class AskRequest(BaseModel):
    query: str
    k: int = 4

def get_rag_service():
    return RagService(OpenAIClient(), ChromaStore())

@router.post("/ask")
async def ask(request: AskRequest, service: RagService = Depends(get_rag_service)):
    return await service.ask(request.query, request.k)

6.5 Spring Boot 版的 RAG 关键差异

Java 端的 RAG 实现逻辑相同，但依赖不同。通常用 Spring AI 中的 VectorStore 抽象，也可以自己封装嵌入模型。下面给一个简化示例，用来演示“先检索，再生成”的编排顺序。

java复制// src/main/java/com/example/aifamily/application/RagService.java
@Service
@RequiredArgsConstructor
public class RagService {

    private final BaseLLMClient llmClient;
    private final SimpleVectorStore vectorStore;

    public Mono<RagResult> ask(String query, int k) {
        List<Document> docs = vectorStore.similaritySearch(query, k);
        String context = docs.stream()
                .map(doc -> "[片段] " + doc.getContent())
                .reduce((a, b) -> a + "\n\n" + b)
                .orElse("");
        String prompt = buildPrompt(query, context);

        List<ChatMessage> messages = List.of(
                ChatMessage.builder().role("system").content("你是企业知识库助手。").build(),
                ChatMessage.builder().role("user").content(prompt).build()
        );

        return llmClient.chat(messages, 0.3, null)
                .map(resp -> new RagResult(resp.getMessage().getContent(), docs));
    }
}

这里没有引入太重的框架，核心是让大家看到 RAG 的编排逻辑和 LLM 模块是解耦的。向量检索返回 List<Document>，应用层负责拼 Prompt，LLM 客户端负责生成。这套接口在双技术栈中保持一致。

7. Agent 与工具调用：让模型学会动手做事

如果说 RAG 解决的是“知识不足”，Agent 解决的是“能力不足”。模型本身只能生成文本，但通过工具调用，它可以查订单、发通知、更新数据库。Agent 的核心循环是：感知任务 → 决定动作 → 执行工具 → 观察结果 → 再次决定，直到任务完成。

7.1 工具定义

先定义一个统一的工具接口。工具必须有两部分：函数本体和一个描述元数据。描述元数据给模型看，函数本体给程序调用。

python复制# app/domain/tool.py
from typing import Callable, Any, Optional
from pydantic import BaseModel, Field

class ToolParameter(BaseModel):
    name: str
    type: str = "string"
    description: str = ""
    required: bool = True

class Tool(BaseModel):
    name: str
    description: str
    parameters: list[ToolParameter] = Field(default_factory=list)
    function: Optional[Callable] = None

    async def execute(self, **kwargs) -> Any:
        if self.function is None:
            raise ValueError(f"Tool {self.name} has no function bound")
        return await self.function(**kwargs)

7.2 实现两个示例工具

python复制# app/tools/weather_tool.py
async def get_weather(city: str) -> str:
    """模拟查询天气，真实场景应接入天气 API"""
    return f"{city}今日晴，气温 24℃，东南风 2 级。"

python复制# app/tools/order_tool.py
async def get_order_status(order_id: str) -> str:
    """模拟查询订单，真实场景应查询数据库"""
    return f"订单 {order_id} 已发货，预计 2025-08-01 送达。"

python复制# app/tools/registry.py
from app.domain.tool import Tool
from app.tools.weather_tool import get_weather
from app.tools.order_tool import get_order_status

def get_default_tools():
    return [
        Tool(name="get_weather",
             description="根据城市名查询天气",
             parameters=[{"name": "city", "type": "string", "description": "城市名，如北京", "required": True}],
             function=get_weather),
        Tool(name="get_order_status",
             description="根据订单号查询订单状态",
             parameters=[{"name": "order_id", "type": "string", "description": "订单号", "required": True}],
             function=get_order_status),
    ]

7.3 Agent 循环核心逻辑

Agent 的难点在于如何让模型输出“工具调用指令”。主流方案有两种：

1. 使用模型原生 function calling 能力。
2. 让模型输出 JSON，再用代码解析 JSON 并执行工具。

本文选择第二种方案，因为它与厂商解耦，任何支持指令遵循的模型都能用。缺点是需要更严格的异常处理，但作为教学示例，它更适合看清 Agent 的本质。

python复制# app/application/agent_service.py
import json
from typing import List

from app.domain.message import ChatMessage
from app.domain.tool import Tool
from app.infra.llm.base import BaseLLMClient

class AgentService:
    def __init__(self, llm_client: BaseLLMClient, tools: List[Tool]):
        self.llm_client = llm_client
        self.tools = {tool.name: tool for tool in tools}

    def describe_tools(self) -> str:
        lines = []
        for tool in self.tools.values():
            params_desc = "; ".join(
                [f"{p.name}: {p.description}" for p in tool.parameters]
            )
            lines.append(f"- {tool.name}({params_desc}): {tool.description}")
        return "\n".join(lines)

    def build_system_prompt(self) -> str:
        tool_desc = self.describe_tools()
        return f"""你是一个智能助手，可以调用以下工具完成任务。

{tool_desc}

当需要调用工具时，必须只输出以下 JSON 格式：
{{"tool": "工具名", "arguments": {{"参数名": "参数值"}}}}

其他情况，正常回答用户。"""

    async def run(self, user_query: str, max_steps: int = 5) -> dict:
        messages = [
            ChatMessage(role="system", content=self.build_system_prompt()),
            ChatMessage(role="user", content=user_query),
        ]

        for step in range(max_steps):
            response = await self.llm_client.chat(messages, temperature=0.2)
            content = response.message.content.strip()

            if content.startswith("{"):
                try:
                    action = json.loads(content)
                except json.JSONDecodeError:
                    messages.append(ChatMessage(role="assistant", content=content))
                    messages.append(ChatMessage(role="user", content="请输出有效 JSON。"))
                    continue

                tool = self.tools.get(action["tool"])
                if not tool:
                    messages.append(ChatMessage(role="assistant", content=content))
                    messages.append(ChatMessage(role="user", content=f"工具 {action['tool']} 不存在"))
                    continue

                args = action.get("arguments", {})
                result = await tool.execute(**args)
                messages.append(ChatMessage(role="assistant", content=content))
                messages.append(ChatMessage(role="user", content=f"工具执行结果：{result}"))
                continue

            return {
                "final_answer": content,
                "steps": step + 1,
                "history": [m.model_dump() for m in messages],
            }

        return {"final_answer": "达到最大步骤数，任务未完成。", "steps": max_steps}

这段代码的核心价值是展示了 Agent 的“循环”本质。它没有直接调用一次模型就返回，而是把模型输出当作“指令”，执行完工具后再把结果喂给模型。这一步会让很多人突然理解 Agent 和普通聊天的区别：普通聊天是一次问答，Agent 是多轮问答加工具执行的推进。

7.4 Spring Boot 版工具调用示例

Java 版本同样可以做工具注册和调用，只是动态执行函数时需要借助接口而不是直接传 Callable。这里可以设计一个 ToolExecutor 接口。

java复制// src/main/java/com/example/aifamily/domain/Tool.java
@Data
@Builder
public class Tool {
    private String name;
    private String description;
    private List<ToolParameter> parameters;
    private ToolExecutor executor;
}

java复制// src/main/java/com/example/aifamily/domain/ToolExecutor.java
@FunctionalInterface
public interface ToolExecutor {
    String execute(Map<String, Object> arguments);
}

java复制// src/main/java/com/example/aifamily/application/AgentService.java
@Service
@RequiredArgsConstructor
public class AgentService {

    private final BaseLLMClient llmClient;
    private final List<Tool> tools;

    public Mono<Map<String, Object>> run(String userQuery, int maxSteps) {
        List<ChatMessage> messages = new ArrayList<>();
        messages.add(ChatMessage.builder().role("system").content(buildSystemPrompt()).build());
        messages.add(ChatMessage.builder().role("user").content(userQuery).build());

        return executeLoop(messages, maxSteps);
    }

    private Mono<Map<String, Object>> executeLoop(List<ChatMessage> messages, int remainingSteps) {
        if (remainingSteps <= 0) {
            return Mono.just(Map.of("final_answer", "达到最大步骤数"));
        }

        return llmClient.chat(messages, 0.2, null)
                .flatMap(response -> {
                    String content = response.getMessage().getContent().trim();
                    if (!content.startsWith("{")) {
                        return Mono.just(Map.of("final_answer", content));
                    }
                    try {
                        ObjectMapper mapper = new ObjectMapper();
                        JsonNode action = mapper.readTree(content);
                        String toolName = action.get("tool").asText();
                        JsonNode argsNode = action.get("arguments");
                        Map<String, Object> args = mapper.convertValue(argsNode, Map.class);

                        Tool tool = tools.stream()
                                .filter(t -> t.getName().equals(toolName))
                                .findFirst()
                                .orElse(null);

                        if (tool == null) {
                            messages.add(ChatMessage.builder().role("assistant").content(content).build());
                            messages.add(ChatMessage.builder().role("user").content("工具不存在").build());
                            return executeLoop(messages, remainingSteps - 1);
                        }

                        String result = tool.getExecutor().execute(args);
                        messages.add(ChatMessage.builder().role("assistant").content(content).build());
                        messages.add(ChatMessage.builder().role("user").content("工具执行结果：" + result).build());
                        return executeLoop(messages, remainingSteps - 1);
                    } catch (Exception e) {
                        messages.add(ChatMessage.builder().role("assistant").content(content).build());
                        messages.add(ChatMessage.builder().role("user").content("无法解析工具调用，请重新输出").build());
                        return executeLoop(messages, remainingSteps - 1);
                    }
                });
    }
}

8. MCP：工具接入的标准化协议

当系统里的工具越来越多，你会发现每个 Agent 都要自己定义工具格式，不同项目无法复用。MCP（Model Context Protocol，模型上下文协议）就是为了解决这个问题而出现的。它定义了一个标准：模型客户端如何发现可用工具、如何调用工具、如何获取工具结果。

MCP 的意义不在于引入一个新框架，而在于让工具接入变成“配置式”。你可以把一个天气服务、数据库查询服务、GitHub 操作服务分别封装成 MCP Server，然后多个 Agent 共用一套工具发现和调用机制。

8.1 MCP Python 版最小实现

这里使用 FastMCP 库，它能用装饰器快速暴露工具。

python复制# mcp_server.py
from fastmcp import FastMCP

mcp = FastMCP("CompanyTools")

@mcp.tool()
def query_employee(name: str) -> str:
    """根据员工姓名查询在职状态和部门"""
    # 实际项目中这里会查询 HR 系统
    return f"{name}，研发部，在职"

@mcp.tool()
def send_message(phone: str, content: str) -> str:
    """给指定手机号发送短信通知"""
    # 实际项目中接入短信网关
    return f"短信已发送至 {phone}：{content}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

bash复制pip install fastmcp
python mcp_server.py

运行后，这个 MCP Server 会通过标准输入输出等待调用请求。Agent 端通过配置即可发现这两个工具，而不需要在自己的代码里 import 具体的业务函数。

8.2 Agent 接入 MCP 工具

在实际 AGI 应用中，Agent 启动时需加载配置并生成可用工具列表。以 Python 为例：

python复制# app/application/mcp_agent_service.py
import asyncio
from typing import List

from app.domain.tool import Tool

async def load_mcp_tools(mcp_config: dict) -> List[Tool]:
    # 简化实现：实际应建立连接并拉取工具列表
    tools = []
    # 假设通过 MCP 客户端发现了一个 query_employee 工具
    tools.append(Tool(
        name="query_employee",
        description="根据员工姓名查询在职状态和部门",
        parameters=[{"name": "name", "type": "string", "description": "员工姓名"}],
        function=lambda name: f"模拟：{name}，研发部，在职",
    ))
    return tools

更完整的做法是启动 MCP Server 并调用工具发现接口，但这里的主旨是理解 MCP 对工具调用的标准化作用：工具不再绑定在代码里，而是运行在独立的进程中。

8.3 Spring Boot 版 MCP 注意事项

Java 生态中也有 Spring AI 对 MCP 的支持，但如果你不想引入重型依赖，也可以直接用 HTTP 方式实现一个简单的 MCP HTTP Server。用 POST /mcp/tools/call 接收工具名和参数，内部路由到对应 Service。这本质上就是“工具网关”。这种实现的好处是：Java 服务不需要理解 Python 的内部实现，只需遵守同一套 HTTP 协议，非常符合微服务场景。

设计的核心是收敛工具入口。所有工具都通过一个 Controller 暴露，并加上调用日志、权限校验和限流。这样即使工具注册在 Python 端，Java 侧也能通过 HTTP 调用同一个 MCP Server，达成双技术栈共存。

9. 工作流：从单次调用到流程编排

Agent 擅长动态决策，但真实业务里的很多流程是固定的，比如“收到工单→判断分类→调用知识库→生成草稿→人工审核→发送回复”。如果每次都让 Agent 自由发挥，结果不可控。工作流就是把固定的步骤定义成模板，让系统按顺序或条件执行。

9.1 工作流节点设计

工作流可以抽象成三种节点：

• 开始节点：接收输入变量。
• 处理节点：调用 LLM、RAG、工具或代码。
• 条件节点：根据前一步结果决定走哪个分支。
• 结束节点：返回最终结果。

先用 Python 实现一个极简工作流引擎，支持按顺序执行节点。

python复制# app/infra/workflow/engine.py
from typing import Any, Dict, Callable

class WorkflowNode:
    def __init__(self, name: str, func: Callable, depends: list[str] = None):
        self.name = name
        self.func = func
        self.depends = depends or []

class WorkflowEngine:
    def __init__(self):
        self.nodes = {}

    def add_node(self, node: WorkflowNode):
        self.nodes[node.name] = node

    async def run(self, start_node: str, input_data: Dict[str, Any]) -> Dict[str, Any]:
        results = {}
        visited = set()

        async def execute(node_name: str):
            if node_name in visited:
                return
            visited.add(node_name)
            node = self.nodes[node_name]
            kwargs = {"input_data": input_data, "results": results}
            result = await node.func(**kwargs)
            results[node.name] = result

        await execute(start_node)
        return results

9.2 用工作流串联 RAG 和 Agent

让流程体现实际价值，可以构建一个“智能客服工作流”：先查询知识库，再决定是否需要调用 Agent。这个工作流既包含固定步骤，也在中间引入条件判断，更贴近真实场景。

python复制# app/application/workflow_service.py
from app.application.rag_service import RagService
from app.application.agent_service import AgentService
from app.infra.workflow.engine import WorkflowEngine, WorkflowNode

async def create_customer_service_flow(rag: RagService, agent: AgentService):
    async def step_extract( input_data, results):
        query = input_data["query"]
        # 这里可以接意图识别模型，简化处理为直接输出
        return {"intent": input_data.get("intent", "general")}

    async def step_rag( input_data, results):
        query = input_data["query"]
        result = await rag.ask(query, k=3)
        return result

    async def step_agent( input_data, results):
        query = input_data["query"]
        result = await agent.run(query, max_steps=3)
        return result

    async def step_decide( input_data, results):
        rag_answer = results.get("step_rag", {}).get("answer", "")
        # 简单规则：如果 RAG 回答中包含“无法回答”，转 Agent
        if "无法回答" in rag_answer:
            agent_result = await step_agent(input_data, results)
            return {"final_answer": agent_result["final_answer"]}
        return {"final_answer": rag_answer}

    engine = WorkflowEngine()
    engine.add_node(WorkflowNode("step_extract", step_extract))
    engine.add_node(WorkflowNode("step_rag", step_rag))
    engine.add_node(WorkflowNode("step_decide", step_decide))
    return engine

这种方式的好处是：业务逻辑可预测，每一步都有结果记录，便于日志追踪。相比纯 Agent 的自由发挥，带固定工作流的场景更受甲方欢迎，因为可解释、可审计。

9.3 Spring Boot 工作流实现要点

Java 版本不需要全部自己写，可以引入流程引擎，比如 Flowable、Camunda。但如果只想在内存中做简单编排，完全可以用 Map<String, Function> + CompletableFuture 实现。建议先定义节点接口：

java复制public interface FlowNode {
    Mono<Map<String, Object>> execute(Map<String, Object> inputData, Map<String, Object> results);
}

然后把 RAG 和 Agent 的 Service 包成节点，放入 FlowEngine。这套思想和 Python 版完全一致。重点是：不要在业务 Controller 里直接写一大堆 if-else 调用，将流程节点化之后，扩展新步骤只需要新增节点，不再修改原方法。

10. 运行验证与效果检查清单

写到这里，你已经拥有了全家桶的全部核心模块。但代码写出来不等于能上线，接下来给出一套可操作的验证路径。

如果某一步失败，优先检查以下方向：

1. 配置是否正确，尤其是 API_KEY、BASE_URL 和 MODEL。
2. 模型是否支持当前使用的方式，比如工具调用 JSON 输出要求模型有较强的指令遵循能力。
3. 依赖版本是否冲突，Python 使用 pip freeze 检查，Java 使用 Maven/Gradle 依赖树检查。
4. 向量数据库是否真的写入了数据，很多 RAG 失败是因为索引阶段没有调用成功。

下面用一个表格汇总常见问题：

11. 全家桶落地的工程建议与安全注意

把 Demo 变成生产系统，光有功能远远不够。从工程角度看，下面几条值得认真对待。

11.1 配置中心化和密钥隔离

一定不要把 API Key 直接写在代码里。FastAPI 项目可以使用 .env 和 Pydantic Settings；Spring Boot 项目可以使用配置文件加环境变量覆盖。生产环境建议接入配置中心，比如 Nacos、Apollo 等，或者在 CI/CD 中注入环境变量。密钥要遵循最小权限原则，不同服务使用不同 Key，便于在模型平台上审计调用来源。

11.2 统一的结构化日志

大模型应用的排查难度比普通后端高很多，因为你无法确定问题是出现在 Prompt 环节、模型环节还是工具环节。建议在所有关键入口打印结构化日志，记录：

• 请求 ID。
• 用户 ID。
• 模型名称和参数。
• Prompt 内容。
• 模型返回内容。
• 工具调用参数和结果。
• 消耗 Token 数和耗时。

FastAPI 可以通过中间件实现，Spring Boot 则可以用拦截器或 AOP。日志只记录业务必要信息，不得泄露用户隐私和敏感业务数据。这里要特别强调，在生产环境一定要遵守数据安全合规，对任意数据库查询、文件读取、消息发送类工具，都要先做权限校验并保留审计记录。

11.3 请求级超时与重试策略

模型服务经常出现网络抖动。不要无限制等待，也不要无脑重试。推荐方案：

• HTTP 请求设置连接超时和读取超时，比如连接 5 秒，读取 60 秒。
• 对 429、5xx 错误使用指数退避重试，最多 3 次。
• 对超时类错误不重试，直接返回降级结果，避免重复消耗 Token。

11.4 用评测集做回归控制

AI 应用最怕“这周调好，下周模型升级就坏了”。建议建立一个小规模回归测试集，比如 50 条常见问题，定期跑一遍全家桶，人工或 LLM 评估结果质量。每次修改 Prompt、调整分块参数或换模型后都执行评测，确保整体效果没有退化。

11.5 生产环境的路由与灰度

如果内部系统中有多套模型、多个向量库，建议在架构中加入路由层。全局配置一个默认路由，不同业务域走不同模型组合。新版本模型先灰度一小部分流量，观察效果再全量切换，规避模型服务不可控带来的业务风险。

12. 双版本统一架构的核心套路总结

走到这里，你会发现 FastAPI 版和 Spring Boot 版长的其实非常像。它们都采用了相同的思想：

• 用接口隔离模型厂商。
• 用独立模块封装 RAG、Agent、MCP 和工作流。
• 用统一数据模型标准化消息、工具、流程。
• 把业务编排放在应用层，避免 Controller 膨胀。

这四种套路完全可以复制到其他语言。即使将来要用 Go 或 Node.js 实现，也可以沿用这套分模块 + 接口抽象 + 依赖注入的思路。团队在协作时，可以让 Java 组负责流程和工具网关，Python 组负责模型编排和知识库索引，两端通过 MCP 协议互通。

如果你现在正在做一个真正的 AI 应用项目，建议从最小闭环开始：先打通 LLM 接入，再逐步扩展 RAG 和 Agent。只有当你意识到工具调用会越来越多、流程越来越固定时，再引入 MCP 和工作流也不迟。这套全家桶不是一个必须一次装完的重量级框架，而是一组可以按需选用的工程组件。

今天写到的代码示例都是教学级的最小实现，离生产仍有距离，但主干路径已经清晰了。建议你把每个模块单独跑一遍，然后再尝试将两个模块串联起来。比如先让 Agent 调用 RAG，再让工作流调度 Agent，你会发现所有模块之间的耦合点都非常少，这正是这套设计的价值所在。