从 Java 到 Python Agent 编程

第 0 章 · 旅途起点

为什么是 Python？为什么是 Agent？

你已经是 Java 开发者，熟悉 Spring Boot、Maven、MyBatis、强类型、编译期检查。Python 完全相反——动态类型、解释执行、缩进即语法。但 Python 在 AI Agent 领域是事实标准：LLM SDK、AI 框架、数据生态全在 Python 侧。

教程导航

第 1 章 · Hello Python

1.1 你已会 80%

把 Java 语法和 Python 语法并排放，你会发现相似之处远超想象：

变量与基本类型

# Python：无类型声明，无分号
name = "Agent Runtime"
port = 8100
is_healthy = True
tokens = ["hello", "world"]  # list = ArrayList
config = {"key": "value"}      # dict = HashMap

// 对应 Java
String name = "Agent Runtime";
int port = 8100;
boolean isHealthy = true;
List<String> tokens = List.of("hello", "world");
Map<String, String> config = Map.of("key", "value");

字符串：f-string 是最强特性

agent_id = "a001"
log_msg = f"Agent {agent_id} 已启动，端口 {port}"
# 输出: "Agent a001 已启动，端口 8100"
# 等价于 Java: String.format("Agent %s 已启动，端口 %d", agentId, port)

if/for/while

if port > 8000:
    print("高危端口")
elif port > 1024:
    print("普通端口")
else:
    print("特权端口")

# for-each（无 for i 语法，但可以用 range）
for token in tokens:
    print(token)

for i in range(5):  # 0, 1, 2, 3, 4
    print(i)

函数定义

def generate_session_id(agent_id: str) -> str:
    """生成唯一的会话 ID"""
    import uuid
    return f"{agent_id}-{uuid.uuid4().hex[:12]}"

1.2 Python 项目结构：从 Maven 到模块包

EAH 的 Python 模块结构，对应你熟悉的 Maven 多模块：

# Maven 多模块（你熟悉的）
src/backend/
├── api/                    # 启动入口
├── core/                   # 实体
├── infrastructure/         # 基础设施
└── admin/                  # 业务逻辑

# Python 模块（对应 EAH）
src/agent-runtime/app/
├── main.py                 # 启动入口（= Application.java）
├── config/                 # 配置（= application.yml）
├── models/                 # 数据模型（= 实体/DTO）
├── core/                   # 核心逻辑（= Service）
├── api/                    # API 路由（= Controller）
├── services/               # 服务（= Service）
└── tests/                  # 测试（= test/java）

# 每个目录下必须有 __init__.py 才能被视为包
# 这相当于 package-info.java

# __init__.py 可以是空文件，只是告诉 Python "这是一个包"
# 文件内容就是该目录下的 __init__.py

导入语法（import = Java import）

# 绝对导入（推荐）
from app.config.settings import settings
from app.models.agent import AgentRunRequest, AgentRunResponse

# 等价于 Java:
// import com.enterprise.agent.hub.config.Settings;
// import com.enterprise.agent.hub.entity.AgentRunRequest;

# 别名导入
from app.api.v1.health import router as health_router

第 2 章 · 数据模型（Pydantic = Lombok + Jackson）

2.1 从 Java 实体到 Pydantic 模型

EAH 的 `AgentRunRequest` 等价于你的 `@Data` 类 + Jackson 序列化：

# Python Pydantic 模型——既是 @Data 又是 @JsonProperty 还是 @Valid
from pydantic import BaseModel, Field
from typing import Optional, Any

class AgentRunRequest(BaseModel):
    """Agent 执行请求——注释即文档"""
    agent_id: str                                # 必填，无默认值
    user_input: str                              # 必填
    system_prompt: str                            # 必填
    model: Optional[str] = None                 # 可选 = null 默认
    temperature: Optional[float] = None          # 可选
    tool_ids: list[str] = Field(default_factory=list)  # 默认为空列表

// 对比 Java @Data 类
@Data
@NoArgsConstructor
@AllArgsConstructor
public class AgentRunRequest {
    @NotNull private String agentId;
    @NotNull private String userInput;
    @NotNull private String systemPrompt;
    private String model;
    private Double temperature;
    @Builder.Default
    private List<String> toolIds = new ArrayList<>();
}

2.2 Pydantic 核心要点

EAH 中的 Pydantic 实战

# 创建请求对象
req = AgentRunRequest(
    agent_id="a001",
    user_input="查询今日数据",
    system_prompt="你是一个数据分析助手",
    instruction_prompt="请执行: {{query}}",
    variables={"query": "SELECT * FROM orders"},
    model="deepseek-chat",
)

# JSON 序列化（发给 Java 后端）
json_str = req.model_dump_json()
# '{"agent_id":"a001","user_input":"查询今日数据","variables":{"query":"SELECT * FROM orders"},"tool_ids":[],"tool_schemas":[]}'

# JSON 反序列化
req2 = AgentRunRequest.model_validate_json(json_str)

# 字段访问（像 Java getter 一样）
print(req.agent_id)  # "a001"

第 3 章 · 配置管理（从 application.yml 到 pydantic-settings）

3.1 Spring @Value → pydantic-settings

EAH 使用 `pydantic-settings` 管理配置，等价于 Spring 的 `@ConfigurationProperties` 或 `application.yml`：

# Python: settings.py —— 对应 application.yml
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """Agent Runtime 全局配置——等价于 @ConfigurationProperties(prefix="agent")"""
    app_name: str = "Agent Runtime"
    app_port: int = 8100

    llm_base_url: str = "https://api.deepseek.com/v1"
    llm_api_key: str = ""       # 敏感字段，从环境变量读取
    llm_default_model: str = "deepseek-chat"

    model_config = {  # 等价于 @ConfigurationProperties 的 prefix
        "env_prefix": "AGENT_",   # 环境变量如 AGENT_LLM_API_KEY
        "env_file": ".env",        # 也支持 .env 文件
    }

# 导出单例（饿汉式）——等价于 Spring 自动注入的单例 Bean
settings = Settings()

在代码中使用

from app.config.settings import settings

# 像使用 @Autowired 的 Bean 一样
payload = {
    "model": settings.llm_default_model,
    "messages": messages,
}

# 等价于 Java:
// @Value("${agent.llm.default-model}")
// private String llmDefaultModel;

# 也可以直接从环境变量覆盖：
# $ export AGENT_LLM_BASE_URL="https://api.openai.com/v1"

依赖管理（requirements.txt = pom.xml）

# requirements.txt —— 等价于 pom.xml 的 <dependencies>
fastapi==0.115.6           # Spring Boot Web 启动器
uvicorn[standard]==0.34.0   # 内嵌 Tomcat（Web 服务器）
pydantic-settings==2.7.1    # @ConfigurationProperties
httpx==0.28.1               # WebClient（异步 HTTP）
jinja2==3.1.5                # Thymeleaf（模板引擎）

第 4 章 · 异步编程（从 ThreadPool 到 async/await）

这是 Python 和 Java 差异最大的领域。但思路完全一样——只是语法不同。

Java 异步 vs Python 协程

# Python 异步函数
async def fetch_data(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.json()

# 调用异步函数
data = await fetch_data("http://api.example.com")

// 等价 Java
public class AsyncService {
    public CompletableFuture<JsonNode> fetchData(String url) {
        return webClient.get().uri(url).retrieve().bodyToMono(JsonNode.class).toFuture();
    }
}
// 调用
JsonNode data = fetchData(url).get();  // 同步阻塞
// 或
fetchData(url).subscribe(data -> process(data));  // 非阻塞回调

EAH 中的异步实战：并行 HTTP 调用

async def chat_completion(
    messages: list[dict],
    model: Optional[str] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    tools: Optional[list[dict]] = None,
) -> tuple[dict, dict]:
    """调用 LLM API"""
    headers = {
        "Authorization": f"Bearer {settings.llm_api_key}",
        "Content-Type": "application/json",
    }
    payload = _build_payload(messages, model, temperature, max_tokens, tools)

    try:
        async with httpx.AsyncClient(timeout=settings.llm_request_timeout) as client:
            resp = await client.post(
                f"{settings.llm_base_url}/chat/completions",
                json=payload, headers=headers,
            )
            resp.raise_for_status()
            data = resp.json()
    except httpx.HTTPStatusError as e:
        raise AgentLLMException(f"LLM 返回 HTTP {e.response.status_code}") from e

    choice = data["choices"][0]
    return choice["message"], data.get("usage", {})

async for：流式处理 AsyncGenerator

# EAH 中的 SSE 流式生成器（等价于 Java Flux<String>）
async def run_agent_stream(request) -> AsyncGenerator[str, None]:
    """生成 SSE 事件流"""
    ...
    while iteration < max_iters:
        yield f"data: {...}\n\n"   # yield = 产生一个值但不结束函数
        ...

# 在 API 路由中消费（对应 Java Flux 的 SSE 端点）
async def event_generator():
    async for sse_event in run_agent_stream(request):
        yield sse_event

return StreamingResponse(event_generator(), media_type="text/event-stream")

第 5 章 · HTTP 客户端（从 WebClient 到 httpx）

EAH 用 `httpx` 做 HTTP 调用。你熟悉的 `WebClient` 用法完全对应：

EAH MCP 引擎：完整的 HTTP 客户端实战

# MCP 协议引擎——通过 HTTP 发现和调用远程工具
async def discover_tools(mcp_server_url: str) -> list[ToolSchema]:
    """从 MCP Server 发现可用工具"""
    async with httpx.AsyncClient(timeout=settings.mcp_connect_timeout) as client:
        resp = await client.post(
            f"{mcp_server_url}/tools/list",
            json={"method": "tools/list"},
        )
        resp.raise_for_status()  # 检查 HTTP 状态码，非 2xx 抛异常
        data = resp.json()       # 自动解析 JSON 响应体
        return [ToolSchema(**t) for t in data.get("tools", [])]
        # 列表推导式（List comprehension）—— Python 最强大的语法之一
        # 等价 Java: data.getTools().stream().map(t -> new ToolSchema(t)).toList()

列表推导式详解

# [表达式 for 变量 in 可迭代对象 if 条件]
# 等价于 Java Stream API

# 1. 基本映射
squares = [x*x for x in range(10)]
# Java: IntStream.range(0, 10).map(x -> x * x).boxed().toList()

# 2. 带过滤
even = [x for x in range(10) if x % 2 == 0]
# Java: IntStream.range(0, 10).filter(x -> x % 2 == 0).boxed().toList()

# 3. 解包字典（EAH 实战）
tools = [ToolSchema(**t) for t in data.get("tools", [])]
# **t 把字典展开为关键字参数：ToolSchema(name=t["name"], description=t["description"], ...)
# 等价 Java: data.getTools().stream().map(t -> new ToolSchema(t.getName(), t.getDesc())).toList()

# 4. 字典推导式
square_map = {x: x*x for x in range(5)}
# {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}

第 6 章 · FastAPI 服务（从 @RestController 到 @router）

EAH 的 API 层用 FastAPI。和 Spring MVC 高度对应：

EAH 的入口文件 —— main.py

"""FastAPI 应用入口——等价于 EnterpriseAgentHubApplication.java"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config.settings import settings
from app.api.v1.health import router as health_router
from app.api.v1.agent import router as agent_router

def create_app() -> FastAPI:
    """创建并配置应用——等价于 @SpringBootApplication + @Bean"""
    app = FastAPI(title=settings.app_name, version="0.1.0")

    # 等价于 SecurityConfig.addCorsMappings()
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # 等价于 @RequestMapping("/v1") + 子路由
    app.include_router(health_router, prefix="/v1")
    app.include_router(agent_router, prefix="/v1")

    return app

# 模块级单例——等价于 Spring IoC 创建的 Bean
app = create_app()

EAH 的 API 端点 —— agent.py

from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from app.models.agent import AgentRunRequest, AgentRunResponse
from app.core.agent_runner import run_agent, run_agent_stream

router = APIRouter()  # 等价于 @RestController 的类级别"分组"

@router.post("/agent/run", response_model=AgentRunResponse)
async def agent_run(request: AgentRunRequest):  # 自动从 JSON body 反序列化+校验
    """同步执行 Agent"""
    try:
        return await run_agent(request)
    except AgentException as e:
        raise HTTPException(status_code=400, detail=e.message)

@router.post("/agent/run-stream")
async def agent_run_stream(request: AgentRunRequest):
    """SSE 流式执行——等价于 Spring 的 Flux<ServerSentEvent>"""
    async def event_generator():
        async for sse_event in run_agent_stream(request):
            yield sse_event

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )

第 7 章 · Jinja2 模板引擎（从 Thymeleaf 到 Jinja2）

EAH 用 Jinja2 渲染 Agent 的 Prompt 模板。等价于你用 Thymeleaf 做服务端模板渲染：

# EAH 的 prompt_renderer.py——Prompt 模板引擎
from jinja2 import Template, StrictUndefined
from app.core.exceptions import AgentValidationException

def render_prompt(template: str, variables: dict[str, str]) -> str:
    """
    将任务指令模板中的 {{变量}} 替换为实际值。
    使用 StrictUndefined：变量未定义时立即抛异常。
    """
    if not variables:
        if "{{" not in template:
            return template

    try:
        return Template(template, undefined=StrictUndefined).render(**variables)
    except Exception as e:
        raise AgentValidationException(f"模板变量未定义: {e}") from e

EAH 中的模板使用

# Java 端定义的 Prompt 模板（存储在数据库）
instruction_prompt = """
请根据用户输入查询数据库。
用户输入：{{query}}
数据表：{{table_name}}

请按以下格式回答：
- 查询结果：...
- 分析结论：...
"""

# Python 端渲染
rendered = render_prompt(instruction_prompt, {
    "query": "SELECT * FROM orders",
    "table_name": "orders",
})
# 渲染结果就是发送给 LLM 的完整指令

第 8 章 · 异常体系（从 @ExceptionHandler 到异常类层次）

EAH 的异常设计——和你熟悉的 Spring 异常体系对应：

"""Agent 异常体系——按错误来源分类，对应不同 HTTP 状态码"""

class AgentException(Exception):      // RuntimeException（基类）
    def __init__(self, message: str, error_code: str = "AGENT_ERROR"):
        self.message = message
        self.error_code = error_code
        super().__init__(message)  # 调用父类构造方法

class AgentValidationException(AgentException):  // 400 Bad Request
    def __init__(self, message: str):
        super().__init__(message, "VALIDATION_ERROR")

class AgentLLMException(AgentException):         // 502 Bad Gateway
    def __init__(self, message: str):
        super().__init__(message, "LLM_ERROR")

// Spring 的等价设计
@ResponseStatus(HttpStatus.BAD_REQUEST)
public class AgentValidationException extends RuntimeException {
    private String errorCode = "VALIDATION_ERROR";
}

@ResponseStatus(HttpStatus.BAD_GATEWAY)
public class AgentLLMException extends RuntimeException { ... }

异常映射（API 层）

# 异常 → HTTP 状态码映射——等价于 @ExceptionHandler 方法
EXCEPTION_STATUS_MAP = {
    AgentValidationException: 400,
    AgentLLMException: 502,
    AgentToolException: 502,
    AgentExecutionException: 500,
}

# isinstance = Java 的 instanceof
def _map_exception_to_status(e: AgentException) -> int:
    for exc_type, status in EXCEPTION_STATUS_MAP.items():
        if isinstance(e, exc_type):  // e instanceof AgentValidationException
            return status
    return 500

第 9 章 · 测试（从 JUnit + Mockito 到 pytest）

EAH 用 pytest 做单元测试。和 JUnit 5 几乎一一对应：

EAH 模型测试（纯数据层）

"""Agent 数据模型测试——验证 Pydantic 模型的序列化与校验"""
import pytest
from app.models.agent import AgentRunRequest, AgentRunResponse

class TestAgentRunRequest:
    """AgentRunRequest 模型测试"""

    def test_minimal_request(self):
        """最小必填字段"""
        req = AgentRunRequest(
            agent_id="a001",
            user_input="你好",
            system_prompt="你是助手",
            instruction_prompt="回答: {{query}}",
        )
        assert req.agent_id == "a001"
        assert req.variables == {}         # assertEquals
        assert req.tool_ids == []          # 默认空列表

    def test_full_request(self):
        """完整字段"""
        req = AgentRunRequest(
            agent_id="a002",
            user_input="查询",
            system_prompt="system",
            instruction_prompt="指令",
            model="deepseek-chat",
            temperature=0.5,
            max_tokens=1024,
        )
        assert req.model == "deepseek-chat"
        assert req.temperature == 0.5

EAH 的异常路径测试

"""模板渲染测试——验证正常路径和异常路径"""
from app.core.prompt_renderer import render_prompt
from app.core.exceptions import AgentValidationException

class TestPromptRenderer:
    def test_render_with_variables(self):
        """正常渲染"""
        result = render_prompt("你好 {{name}}", {"name": "张三"})
        assert result == "你好 张三"

    def test_render_missing_variable(self):
        """缺少模板变量应抛异常——等价于 assertThrows"""
        with pytest.raises(AgentValidationException, match="模板变量未定义"):
            render_prompt("你好 {{name}}", {})
        # with pytest.raises = Java 的 assertThrows(AgentValidationException.class, () -> {...})
        # match = 验证异常消息包含特定字符串

第 10 章 · 组装 Agent：完整的 ReAct 循环

这是核心章节——将前面所有知识串联，实现一个真正的 Agent。EAH 的 `agent_runner.py` 实现了完整的 ReAct（Reasoning + Acting）循环。

10.1 什么是 ReAct 循环

ReAct ≈ 思考 → 行动 → 观察 → 重复 → 得出结论。等价于一个 while 循环：

# Agent = 带 LLM 推理能力的 while(true) 循环
while iteration < max_iterations:
    # 1️⃣ 思考（Think）：LLM 决定下一步做什么
    response = await llm.chat(messages, tools)

    # 2️⃣ 行动（Act）：若 LLM 调用了工具则执行
    if response.has_tool_calls():
        for tool_call in response.tool_calls:
            result = await tool.execute(tool_call)
            messages.add(result)   # 观察结果注入对话
        continue                # 继续循环

    # 3️⃣ 输出（Answer）：LLM 输出最终回答
    return response.content

10.2 EAH 的 ReAct 实现

# EAH 的 agent_runner.py——完整的 ReAct 循环
async def run_agent(request: AgentRunRequest) -> AgentRunResponse:
    start_time = time.time()

    # Step 1: 渲染 Prompt（Jinja2）
    instruction = render_prompt(request.instruction_prompt, request.variables)

    # Step 2: 构建消息列表（相当于 Java 的 List<Message>）
    messages = _build_initial_messages(
        request.system_prompt, instruction, request.user_input, request.history
    )

    # Step 3: 构建工具 Schema（OpenAI function calling 格式）
    tools = _build_tool_schemas(request.tool_schemas) if request.tool_schemas else None

    # Step 4: ReAct 循环
    iteration = 0
    max_iters = request.max_iterations or settings.agent_max_iterations

    while iteration < max_iters:
        iteration += 1

        # 最后一轮不给工具——强制 LLM 输出文字
        current_tools = tools if iteration < max_iters else None

        # 调用 LLM（async/await）
        response_msg, usage = await chat_completion(
            messages=messages,
            model=request.model,
            temperature=request.temperature,
            max_tokens=request.max_tokens,
            tools=current_tools,
        )
        messages.append(response_msg)

        # 判断 LLM 是否调用了工具
        tool_calls = response_msg.get("tool_calls", [])
        if tool_calls:
            # 🔧 执行工具调用
            for tc in tool_calls:
                result = await _execute_tool_call(tc)  # async HTTP 调用
                messages.append(_build_tool_result_message(tc["id"], result["result"]))
            continue  # → 回到 LLM 推理（让 LLM 看工具结果后决定下一步）

        # ✅ LLM 直接输出回答——循环结束
        final_content = response_msg.get("content", "")
        break

    return AgentRunResponse(
        content=final_content,
        iterations=iteration,
        duration_ms=(time.time() - start_time) * 1000,
        success=bool(final_content),
    )

10.3 掌握 Python 后你可以做什么

基于 EAH 的代码模式，你可以独立编码以下 Agent 场景：

1. 简单问答 Agent： `system_prompt` + `user_input` → LLM 直接回复（无工具调用）
2. 数据库查询 Agent： ReAct 循环 + 一个 `execute_sql` 工具 → 自然语言查数据库
3. 监控分析 Agent： ReAct 循环 + `query_prometheus` 工具 → 分析指标根因
4. 多工具 Agent： 多个工具 Schema → LLM 自主选择调用哪个
5. 流式 Agent： SSE 推送中间状态 → 前端实时展示思考过程
6. 带 HITL 的 Agent： 写操作前暂停等待人工审批

10.4 EAH 代码中的 Python 特性实战

第 11 章 · 总结：从 Java 到 Python 的心智模型

核心思维转换

Python 技能树速查

# 你已掌握的核心技能：
# ✔ 变量/类型/list/dict  →  对应 Java 基础类型
# ✔ if/for/while/def     →  对应 Java 流程控制
# ✔ class/继承/异常      →  对应 Java OOP
# ✔ import              →  对应 Java import
# ✔ Pydantic            →  @Data + Jackson + @Valid
# ✔ async/await         →  WebClient + CompletableFuture
# ✔ FastAPI             →  Spring MVC @RestController
# ✔ httpx               →  WebClient
# ✔ pytest              →  JUnit + Mockito + AssertJ

# 下一步需要深入：
# 🔜 装饰器（Decorator）         →  Spring AOP @Around
# 🔜 上下文管理器（with）        →  try-with-resources
# 🔜 生成器（yield）             →  Flux / Iterator
# 🔜 元类（Metaclass）           →  无直接对应（Spring 后处理）

快速参考：Java → Python 对照表