Java 开发者的 Python 快速迁移指南

目标不是从零学习编程，而是把已经掌握的 Java 概念，快速映射到 Python，并最终能够阅读和编写 Agent / Harness 工程代码。

本文默认使用 Python 3.11+。第三方库示例采用 Pydantic v2、HTTPX、pytest 和 uv。

阅读方式

每一节都围绕四件事展开：

Python 写法是什么。
可以类比成哪个 Java 概念。
给出一个可以直接运行的例子。
标出 Java 开发者最容易踩的坑。

不必先把所有语法背完。建议先读第 1～6 章，能够看懂普通 Python；再重点学习第 9、10、12 章，进入 Agent / Harness 开发。

一张表先建立整体映射

Java	Python	说明
`null`	`None`	空值
`true / false`	`True / False`	首字母大写
`&& / \|\| / !`	`and / or / not`	逻辑运算
`List<T>`	`list[T]`	列表
`Map<K, V>`	`dict[K, V]`	字典
`Set<T>`	`set[T]`	集合
`list.add(x)`	`items.append(x)`	追加元素
`list.size()`	`len(items)`	长度
`map.entrySet()`	`mapping.items()`	遍历键值对
`for (X x : xs)`	`for x in xs:`	增强 for
`this`	`self`	当前对象
构造函数	`__init__`	初始化对象
`static`	`@staticmethod`	静态方法
方法重载	默认参数/可选参数	Python 后定义会覆盖先定义
`Object... args`	`*args`	可变位置参数
`Map<String, Object>`	`**kwargs`	可变命名参数
`try-with-resources`	`with`	自动释放资源
`CompletableFuture`	`asyncio` / `async` / `await`	异步任务
`Stream<T>`	生成器 / `yield`	惰性产生数据
Lombok `@Data`	`@dataclass`	数据类
DTO + Jackson + Validation	Pydantic `BaseModel`	数据解析与校验
Maven / Gradle	`uv` / `pip`	依赖和项目管理
`pom.xml`	`pyproject.toml`	项目配置

01. 变量与基础类型

1.1 变量不需要先声明类型

Java 在编译期固定变量类型：

java

int age = 18;
String name = "Tom";

Python 直接赋值：

python

age = 18
name = "Tom"

Python 是动态类型语言，类型属于值，而不是变量名：

python

value = 18
print(type(value))  # <class 'int'>

value = "eighteen"
print(type(value))  # <class 'str'>

这在语法上合法，但工程代码里不建议让同一变量频繁改变含义。

1.2 常用基础类型

Java	Python	示例
`int` / `long`	`int`	`count = 100`
`double`	`float`	`score = 98.5`
`boolean`	`bool`	`enabled = True`
`String`	`str`	`name = "Tom"`
`byte[]`	`bytes`	`data = b"hello"`
`null`	`None`	`result = None`

Python 的 int 不需要区分 int 和 long：

python

small = 10
large = 10**100
print(large)

1.3 `None`：对应 Java 的 `null`

python

result = None

if result is None:
    print("还没有结果")

推荐使用 is None，不要写 result == None。None 是一个单例对象，is 表达的正是“它是不是这个唯一的空值对象”。

1.4 布尔值和逻辑运算

python

enabled = True
deleted = False

if enabled and not deleted:
    print("可以使用")

对应关系：

text

Java &&    → Python and
Java ||    → Python or
Java !     → Python not

注意 True、False 首字母必须大写。

1.5 字符串与 f-string

Java：

java

String message = "name=" + name + ", age=" + age;

Python 推荐 f-string：

python

name = "Tom"
age = 18
message = f"name={name}, age={age}"
print(message)

格式化也可以直接写在 {} 中：

python

price = 12.3456
print(f"价格：{price:.2f}")  # 价格：12.35

多行字符串使用三引号：

python

prompt = """
你是一个代码审查助手。
请找出下面代码的并发问题。
""".strip()

1.6 类型标注 Type Hints

Python 的类型标注主要帮助 IDE、静态检查器和读代码的人，不会自动变成 Java 那样的运行时强制约束。

python

name: str = "Tom"
age: int = 18
scores: list[int] = [90, 95]
metadata: dict[str, str] = {"source": "api"}

函数参数和返回值：

python

def add(a: int, b: int) -> int:
    return a + b

允许为空：

python

def find_user(user_id: str) -> dict[str, object] | None:
    if user_id == "u-1":
        return {"id": "u-1", "name": "Tom"}
    return None

老代码可能写成：

python

from typing import Optional

name: Optional[str] = None

在 Python 3.10+ 中，优先写 str | None。

1.7 `Any` 与 `object`

python

from typing import Any

raw: Any = {"anything": 1}
value: object = "hello"

Any：基本放弃静态类型检查，可以继续调用任意属性。
object：表示任何 Python 对象，但使用前通常要先判断具体类型。

python

def normalize(value: object) -> str:
    if isinstance(value, str):
        return value.strip()
    return str(value)

在 Agent 工程里，不要为了省事到处使用 Any，否则工具参数、Observation 和模型响应的边界会失去约束。

02. 条件判断

2.1 `if / elif / else`

Python 使用缩进表示代码块，没有花括号：

python

score = 85

if score >= 90:
    level = "A"
elif score >= 80:
    level = "B"
else:
    level = "C"

print(level)

通常使用 4 个空格缩进。不要混用 Tab 和空格。

2.2 `in`：包含判断

Java：

java

if (names.contains("Tom")) { ... }

Python：

python

names = ["Tom", "Jack"]

if "Tom" in names:
    print("找到了")

in 也能用于字符串、集合和字典：

python

print("agent" in "agent runtime")       # True
print("admin" in {"admin", "user"})   # True
print("model" in {"model": "gpt"})    # True，字典默认判断 key

2.3 `==` 与 `is`

python

a = [1, 2]
b = [1, 2]

print(a == b)  # True：值相等
print(a is b)  # False：不是同一个对象

Java 类比：

text

Python ==  ≈ Java equals()
Python is  ≈ Java 引用对象的 ==

正确的空值判断：

python

if result is None:
    ...

if result is not None:
    ...

不要用 is 比较字符串和数字。

2.4 Truthy / Falsy

下面这些值在条件中都被视为假：

python

None
False
0
0.0
""
[]
{}
set()

所以 Java 的：

java

if (users != null && !users.isEmpty()) { ... }

在 Python 中常写成：

python

users = ["Tom"]

if users:
    print("有用户")

但要注意：if not value 无法区分 None、0 和空字符串。如果业务上三者含义不同，就必须显式判断：

python

limit = 0

if limit is None:
    limit = 10

2.5 `or` 提供默认值

python

input_name = ""
name = input_name or "default"
print(name)  # default

这会把所有 Falsy 值都替换掉，包括 0、False 和空集合。如果只想替换 None，应写：

python

timeout = 0
effective_timeout = 30 if timeout is None else timeout
print(effective_timeout)  # 0

2.6 三元表达式

Java：

java

String level = age >= 18 ? "adult" : "minor";

Python：

python

age = 18
level = "adult" if age >= 18 else "minor"

阅读顺序是：先看中间的条件，再决定取左边还是右边。

2.7 `match`：结构化分支

Python 3.10+ 支持模式匹配，可以类比增强版 switch：

python

def handle_event(event: dict[str, object]) -> str:
    match event:
        case {"type": "message", "text": str(text)}:
            return f"收到消息：{text}"
        case {"type": "tool_result", "success": True}:
            return "工具执行成功"
        case _:
            return "未知事件"

print(handle_event({"type": "message", "text": "hello"}))

case _ 类似 Java switch 中的 default。

03. 循环

3.1 `for`：遍历可迭代对象

Java 增强 for：

java

for (String name : names) { ... }

Python：

python

names = ["Tom", "Jack", "Bob"]

for name in names:
    print(name)

Python 的 for 不只遍历 List，任何可迭代对象都可以，包括 tuple、set、dict、生成器和文件对象。

3.2 `range`：按次数循环

python

for i in range(5):
    print(i)

输出 0, 1, 2, 3, 4。与切片一样，包含开始值、不包含结束值。

python

print(list(range(2, 6)))     # [2, 3, 4, 5]
print(list(range(0, 10, 2))) # [0, 2, 4, 6, 8]
print(list(range(5, 0, -1))) # [5, 4, 3, 2, 1]

3.3 `while`

python

attempt = 0

while attempt < 3:
    attempt += 1
    print(f"第 {attempt} 次重试")

Python 没有 i++，要写 i += 1。

3.4 `enumerate`：同时获得下标和值

python

users = ["Tom", "Jack"]

for index, user in enumerate(users):
    print(index, user)

从 1 开始编号：

python

for index, user in enumerate(users, start=1):
    print(f"第 {index} 个用户：{user}")

不要为了拿下标写 for i in range(len(users))，除非确实需要通过下标修改列表。

3.5 `zip`：并行遍历多个集合

python

names = ["Tom", "Jack"]
ages = [18, 20]

for name, age in zip(names, ages):
    print(name, age)

默认在最短集合结束时停止：

python

print(list(zip([1, 2, 3], ["a", "b"])))
# [(1, 'a'), (2, 'b')]

如果长度必须一致，Python 3.10+ 可以写：

python

for name, age in zip(names, ages, strict=True):
    print(name, age)

长度不一致会抛出 ValueError。

3.6 `break`、`continue` 与循环的 `else`

python

for number in [1, 3, 6, 7]:
    if number % 2 != 0:
        continue
    print(f"找到偶数：{number}")
    break
else:
    print("没有找到偶数")

循环后的 else 只会在循环没有被 break 中断时执行。这是 Python 特有写法，但可读性有争议；团队不熟悉时，用显式标志变量更清楚。

3.7 列表推导式

Java Stream：

java

users.stream()
    .filter(user -> user.getAge() >= 18)
    .map(User::getName)
    .toList();

Python：

python

users = [
    {"name": "Tom", "age": 18},
    {"name": "Lucy", "age": 16},
]

adult_names = [
    user["name"]
    for user in users
    if user["age"] >= 18
]

print(adult_names)  # ['Tom']

也有集合和字典推导式：

python

squares = {n * n for n in range(5)}
name_by_id = {user["name"]: user["age"] for user in users}

推导式超过两层或包含复杂条件时，普通 for 往往更容易维护。

04. 集合

4.1 List：对应 Java `ArrayList`

python

names: list[str] = ["Tom", "Jack"]

names.append("Bob")          # 末尾追加一个元素
names.extend(["Lucy", "Amy"]) # 追加多个元素
names.insert(1, "Alex")      # 在下标 1 插入

print(names[0])
print(len(names))

常见映射：

Java	Python
`list.add(x)`	`items.append(x)`
`list.addAll(xs)`	`items.extend(xs)`
`list.get(i)`	`items[i]`
`list.set(i, x)`	`items[i] = x`
`list.size()`	`len(items)`
`list.remove(i)`	`items.pop(i)`
`list.remove(value)`	`items.remove(value)`

负数下标从末尾开始：

python

items = [10, 20, 30]
print(items[-1])  # 30
print(items[-2])  # 20

4.2 Dict：对应 Java `HashMap`

python

user: dict[str, object] = {
    "name": "Tom",
    "age": 18,
}

user["enabled"] = True
print(user["name"])

安全读取：

python

nickname = user.get("nickname")
nickname_with_default = user.get("nickname", "anonymous")

区别是：user["missing"] 抛 KeyError，user.get("missing") 返回 None。

遍历：

python

for key, value in user.items():
    print(key, value)

for key in user:
    print(key)

for value in user.values():
    print(value)

合并字典：

python

defaults = {"timeout": 30, "retry": 2}
overrides = {"timeout": 60}
config = defaults | overrides
print(config)  # {'timeout': 60, 'retry': 2}

4.3 Set：去重与成员判断

python

permissions: set[str] = {"read", "write", "read"}
print(permissions)  # 重复的 read 只保留一个

permissions.add("admin")
print("write" in permissions)

集合运算：

python

a = {"read", "write"}
b = {"write", "delete"}

print(a | b)  # 并集
print(a & b)  # 交集
print(a - b)  # 差集

空集合必须写 set()；{} 创建的是空字典。

4.4 Tuple：固定结构的不可变序列

python

point = (120.1, 30.2)
x, y = point
print(x, y)

Tuple 本身不能修改：

python

# point[0] = 121.0  # TypeError

只有一个元素时必须保留逗号：

python

single = (1,)
print(type(single))  # tuple

4.5 切片 Slice

python

nums = [0, 1, 2, 3, 4, 5]

print(nums[1:4])   # [1, 2, 3]
print(nums[:3])    # [0, 1, 2]
print(nums[3:])    # [3, 4, 5]
print(nums[::2])   # [0, 2, 4]
print(nums[::-1])  # [5, 4, 3, 2, 1, 0]

通用格式是 [start:end:step]，同样包含 start、不包含 end。

4.6 解包 Unpacking

python

name, age = ("Tom", 18)
print(name, age)

接收剩余元素：

python

first, *middle, last = [1, 2, 3, 4, 5]
print(first)   # 1
print(middle)  # [2, 3, 4]
print(last)    # 5

交换变量不需要临时变量：

python

a, b = 1, 2
a, b = b, a

4.7 排序

原地修改：

python

numbers = [3, 1, 2]
numbers.sort()
print(numbers)

返回新列表：

python

users = [
    {"name": "Tom", "age": 20},
    {"name": "Lucy", "age": 18},
]

sorted_users = sorted(users, key=lambda user: user["age"])

倒序：

python

sorted_users = sorted(
    users,
    key=lambda user: user["age"],
    reverse=True,
)

4.8 引用、浅拷贝与深拷贝

python

original = [[1], [2]]
same = original
shallow = original.copy()

same.append([3])
print(original)  # [[1], [2], [3]]，因为是同一个对象

shallow[0].append(99)
print(original)  # [[1, 99], [2], [3]]，内部列表仍然共享

完全复制嵌套对象：

python

from copy import deepcopy

independent = deepcopy(original)
independent[0].append(100)

这与 Java 中复制引用、浅克隆和深克隆的差异一致。

05. 函数与参数

5.1 定义函数

python

def add(a: int, b: int) -> int:
    """返回两个整数之和。"""
    return a + b

print(add(1, 2))

没有返回值时通常标注 -> None：

python

def log_message(message: str) -> None:
    print(message)

5.2 返回多个值

python

def parse_user() -> tuple[str, int]:
    return "Tom", 18

name, age = parse_user()

本质上返回了 tuple ("Tom", 18)。如果字段很多、含义复杂，优先使用 dataclass、NamedTuple 或 Pydantic 模型。

5.3 默认参数

python

def query(name: str, limit: int = 10) -> list[str]:
    return [name] * limit

query("Tom")
query("Tom", 3)

这通常替代 Java 的方法重载。

可变默认值陷阱

不要这样写：

python

def add_item(item: str, items: list[str] = []):
    items.append(item)
    return items

默认列表只在函数定义时创建一次，多次调用会共享它。正确写法：

python

def add_item(item: str, items: list[str] | None = None) -> list[str]:
    if items is None:
        items = []
    items.append(item)
    return items

5.4 命名参数

python

def create_user(name: str, age: int, enabled: bool) -> dict[str, object]:
    return {"name": name, "age": age, "enabled": enabled}

user = create_user(
    name="Tom",
    age=18,
    enabled=True,
)

命名参数让 SDK 调用很容易阅读，这也是 Agent SDK 中常见大量参数名的原因。

5.5 仅限命名参数

参数列表中的 * 表示后续参数必须按名字传递：

python

def call_model(prompt: str, *, timeout: float = 30, stream: bool = False) -> str:
    return f"{prompt=}, {timeout=}, {stream=}"

call_model("hello", timeout=10, stream=True)

call_model("hello", 10, True) 会报错。这适合防止多个布尔值、数字参数传错位置。

5.6 `*args`：可变位置参数

python

def total(*numbers: int) -> int:
    return sum(numbers)

print(total(1, 2, 3))

函数内部的 numbers 是 tuple，可类比 Java 的 int... numbers。

5.7 `**kwargs`：可变命名参数

python

def build_config(**kwargs: object) -> dict[str, object]:
    return kwargs

config = build_config(model="gpt", timeout=30)
print(config)

内部得到：

python

{"model": "gpt", "timeout": 30}

它可以类比 Map<String, Object>，但类型过宽，不宜用它替代所有明确参数。

5.8 用 `*` 和 `**` 拆开集合

python

def add(a: int, b: int) -> int:
    return a + b

params = [1, 2]
print(add(*params))  # 等价于 add(1, 2)

python

def create_user(name: str, age: int) -> str:
    return f"{name}:{age}"

user = {"name": "Tom", "age": 18}
print(create_user(**user))

Agent 项目经常使用 ModelConfig(**raw_config) 把字典转换成配置对象。

5.9 Lambda

python

double = lambda x: x * 2
print(double(3))

常见用途是传给 sorted：

python

users = [{"name": "Tom", "age": 20}, {"name": "Lucy", "age": 18}]
users.sort(key=lambda user: user["age"])

Lambda 只适合短小的单表达式。逻辑稍复杂就定义普通函数。

5.10 作用域与 `nonlocal` / `global`

Python 查找变量大致遵循“局部 → 外层函数 → 模块全局 → 内置”的顺序。

python

def make_counter():
    count = 0

    def increment() -> int:
        nonlocal count
        count += 1
        return count

    return increment

counter = make_counter()
print(counter())  # 1
print(counter())  # 2

nonlocal 修改外层函数变量；global 修改模块级变量。工程代码中应谨慎使用全局可变状态，尤其在异步和多进程环境里。

5.11 Python 没有传统方法重载

下面第二个定义会覆盖第一个：

python

def query(name: str):
    return name

def query(name: str, limit: int):
    return [name] * limit

通常改成默认参数：

python

def query(name: str, limit: int | None = None):
    if limit is None:
        return name
    return [name] * limit

也可以用不同函数名，让行为更加明确。

06. 类与对象

6.1 定义类、构造对象与 `self`

python

class User:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

    def greet(self) -> str:
        return f"Hello, {self.name}"

user = User("Tom", 18)
print(user.greet())

创建对象不写 new。
实例方法第一个参数显式写 self。
调用 user.greet() 时不用手动传入 self。

可以把 user.greet() 粗略理解为 User.greet(user)。

6.2 类属性与实例属性

python

class Worker:
    category = "agent"  # 类属性，所有实例共享

    def __init__(self, name: str):
        self.name = name  # 实例属性，每个对象独立

a = Worker("planner")
b = Worker("executor")
print(a.category, b.category)

不要把可变对象随意放在类属性上，否则实例会共享：

python

class BadSession:
    messages = []  # 所有实例共用同一个列表

6.3 `@property`：像字段一样访问的方法

python

class Temperature:
    def __init__(self, celsius: float):
        self._celsius = celsius

    @property
    def fahrenheit(self) -> float:
        return self._celsius * 9 / 5 + 32

temp = Temperature(20)
print(temp.fahrenheit)  # 不需要写括号

它可以类比 Java Getter，但调用方表现为属性访问。

6.4 继承与 `super()`

python

class Tool:
    def __init__(self, name: str):
        self.name = name

    def run(self) -> str:
        raise NotImplementedError


class SearchTool(Tool):
    def __init__(self, endpoint: str):
        super().__init__("search")
        self.endpoint = endpoint

    def run(self) -> str:
        return f"search via {self.endpoint}"

class SearchTool(Tool) 对应 Java 的 extends Tool。

6.5 `@staticmethod`

python

class Ids:
    @staticmethod
    def normalize(raw: str) -> str:
        return raw.strip().lower()

print(Ids.normalize(" U-001 "))

静态方法不接收 self 或 cls。如果函数与类的状态完全无关，也可以直接写成模块函数，通常更简单。

6.6 `@classmethod`

python

class User:
    def __init__(self, name: str):
        self.name = name

    @classmethod
    def from_dict(cls, data: dict[str, str]) -> "User":
        return cls(name=data["name"])

user = User.from_dict({"name": "Tom"})

cls 代表当前类，常用于工厂方法和替代构造函数。子类调用时，cls 会自动变成子类。

6.7 `@dataclass`：接近 Lombok 数据类

python

from dataclasses import dataclass, field


@dataclass
class Task:
    task_id: str
    status: str = "pending"
    tags: list[str] = field(default_factory=list)


task = Task(task_id="t-1")
task.tags.append("coding")
print(task)

它会生成 __init__、__repr__、__eq__ 等方法，可以粗略类比 Lombok @Data 加构造函数。

可变字段要用 default_factory，不要直接写 tags: list[str] = []。

不可变数据对象：

python

@dataclass(frozen=True)
class ModelConfig:
    name: str
    temperature: float = 0.0

6.8 Dunder / Magic Method

__xxx__ 叫 dunder method，定义对象如何参与 Python 语言协议。

python

class MessageBuffer:
    def __init__(self, messages: list[str]):
        self._messages = messages

    def __len__(self) -> int:
        return len(self._messages)

    def __str__(self) -> str:
        return "\n".join(self._messages)

buffer = MessageBuffer(["hello", "world"])
print(len(buffer))
print(buffer)

类比：

text

__str__   → Java toString()
__eq__    → Java equals()
__hash__  → Java hashCode()
__len__   → 支持 len(obj)
__iter__  → 支持 for x in obj

6.9 抽象类与 Protocol

传统接口式约束：

python

from abc import ABC, abstractmethod


class Tool(ABC):
    @abstractmethod
    async def run(self, arguments: dict[str, object]) -> str:
        ...

结构化类型约束：

python

from typing import Protocol


class RunnableTool(Protocol):
    async def run(self, arguments: dict[str, object]) -> str:
        ...

Protocol 更像“只要方法长得一样就算实现”，不要求显式 implements，很适合 Agent 工具、模型客户端和存储适配器的依赖倒置。

07. 异常处理

7.1 `try / except / else / finally`

python

def parse_timeout(raw: str) -> int:
    try:
        value = int(raw)
    except ValueError as exc:
        print(f"不是整数：{exc}")
        return 30
    else:
        print("解析成功")
        return value
    finally:
        print("解析结束")

except 对应 Java catch。
else 只在没有异常时执行。
finally 无论成功失败都会执行。

只捕获能处理的具体异常，不要随意写裸 except:。

7.2 多种异常

python

try:
    value = config["timeout"]
    timeout = int(value)
except (KeyError, ValueError, TypeError) as exc:
    timeout = 30
    print(f"使用默认超时：{exc}")

7.3 `raise`：主动抛异常

python

def validate_temperature(value: float) -> None:
    if not 0 <= value <= 2:
        raise ValueError("temperature 必须在 0 到 2 之间")

Java 的 throw new IllegalArgumentException(...) 可以类比成 Python 的 raise ValueError(...)。

7.4 自定义异常

python

class ToolExecutionError(RuntimeError):
    def __init__(self, tool_name: str, message: str):
        super().__init__(f"工具 {tool_name} 执行失败：{message}")
        self.tool_name = tool_name


raise ToolExecutionError("search", "连接超时")

自定义异常名通常以 Error 结尾。

7.5 异常链 `raise ... from ...`

python

import json


def load_arguments(raw: str) -> dict[str, object]:
    try:
        return json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ToolExecutionError("parser", "参数不是合法 JSON") from exc

这样既向上层提供业务含义，又保留原始异常作为 cause，类似 Java 包装异常时传入原始 Throwable。

08. 资源与上下文管理

8.1 `with`：对应 try-with-resources

python

with open("prompt.txt", "r", encoding="utf-8") as file:
    prompt = file.read()

print(prompt)

离开 with 代码块后会自动关闭文件，即使中间抛出异常。

写文件：

python

with open("result.txt", "w", encoding="utf-8") as file:
    file.write("done\n")

8.2 `pathlib`：更现代的路径 API

python

from pathlib import Path

data_dir = Path("data")
data_dir.mkdir(parents=True, exist_ok=True)

result_file = data_dir / "result.json"
result_file.write_text('{"ok": true}', encoding="utf-8")

print(result_file.read_text(encoding="utf-8"))

相比大量 os.path.join(...)，Path 更接近对象化的 Java NIO Path。

8.3 自定义 Context Manager

使用 contextmanager：

python

from contextlib import contextmanager
from time import perf_counter
from collections.abc import Iterator


@contextmanager
def trace(name: str) -> Iterator[None]:
    start = perf_counter()
    print(f"开始：{name}")
    try:
        yield
    finally:
        elapsed = perf_counter() - start
        print(f"结束：{name}，耗时 {elapsed:.3f}s")


with trace("call-model"):
    sum(range(1000))

yield 前相当于资源初始化，yield 后相当于清理逻辑。

8.4 基于类的 Context Manager

python

class Session:
    def __enter__(self):
        print("打开连接")
        return self

    def query(self) -> str:
        return "result"

    def __exit__(self, exc_type, exc_value, traceback):
        print("关闭连接")


with Session() as session:
    print(session.query())

__exit__ 返回 True 会吞掉异常，通常不要这样做，除非确定异常已经被正确处理。

8.5 `async with`

异步资源使用异步上下文管理器：

python

import httpx


async def fetch(url: str) -> str:
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.text

它常用于异步 HTTP 客户端、数据库连接、流式响应和 Trace 生命周期。

09. 等待、异步与并发

9.1 同步等待与异步等待的区别

同步代码在等待 I/O 时，当前线程通常不能继续处理其他任务：

python

import time


def sync_job() -> None:
    time.sleep(1)
    print("完成")

异步代码在等待网络、磁盘等 I/O 时，可以把执行权交回事件循环：

python

import asyncio


async def async_job() -> None:
    await asyncio.sleep(1)
    print("完成")

await 不是“创建新线程”，而是暂停当前协程，让事件循环先运行其他可执行任务。

9.2 `async def` 返回协程对象

python

async def get_answer() -> str:
    return "42"

coroutine = get_answer()
print(coroutine)  # 还没有真正拿到字符串结果

要在异步函数中使用 await：

python

async def main() -> None:
    answer = await get_answer()
    print(answer)


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

asyncio.run() 通常只在程序最外层调用一次。已经处于事件循环中时，不要再次调用它。

9.3 串行 `await`

python

import asyncio


async def call_tool(name: str, seconds: float) -> str:
    await asyncio.sleep(seconds)
    return f"{name}:done"


async def main() -> None:
    first = await call_tool("search", 1)
    second = await call_tool("database", 1)
    print(first, second)

两个任务按顺序执行，总耗时约 2 秒。await 并不自动让相邻调用并发。

9.4 `create_task`：启动并发任务

python

async def main() -> None:
    search_task = asyncio.create_task(call_tool("search", 1))
    db_task = asyncio.create_task(call_tool("database", 1))

    search_result = await search_task
    db_result = await db_task
    print(search_result, db_result)

任务创建后会被事件循环调度，两次等待可以重叠，总耗时约 1 秒。

不要创建任务后丢掉引用。否则异常可能无人接收，任务生命周期也难以管理。

9.5 `gather`：并发等待多个结果

python

async def main() -> None:
    results = await asyncio.gather(
        call_tool("search", 1),
        call_tool("database", 1),
        call_tool("api", 1),
    )
    print(results)

可类比 CompletableFuture.allOf(...)，但 gather 会按传入顺序返回各个结果。

默认情况下，一个任务抛异常会让 gather 向调用者抛出异常。不要随意使用 return_exceptions=True，否则异常会混在正常结果列表里，容易漏处理。

9.6 `TaskGroup`：结构化并发

Python 3.11+ 推荐用 TaskGroup 管理一组相关任务：

python

async def main() -> None:
    async with asyncio.TaskGroup() as group:
        search_task = group.create_task(call_tool("search", 1))
        db_task = group.create_task(call_tool("database", 1))

    print(search_task.result(), db_task.result())

如果其中一个任务失败，TaskGroup 会取消同组其他未完成任务，并以异常组向外报告，生命周期比散落的 create_task 更可控。

9.7 超时

单个调用超时：

python

async def main() -> None:
    try:
        result = await asyncio.wait_for(
            call_tool("slow-tool", 10),
            timeout=2,
        )
        print(result)
    except TimeoutError:
        print("工具执行超时")

Python 3.11+ 也可以包住一段逻辑：

python

async def main() -> None:
    try:
        async with asyncio.timeout(2):
            result = await call_tool("slow-tool", 10)
            print(result)
    except TimeoutError:
        print("整段操作超时")

生产代码中，超时不仅要抛错，还要明确是否重试、任务是否可幂等、远端操作是否可能已经成功。

9.8 取消任务

python

async def worker() -> None:
    try:
        while True:
            await asyncio.sleep(1)
            print("working")
    except asyncio.CancelledError:
        print("收到取消，执行清理")
        raise


async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(2)
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print("任务已取消")

捕获 CancelledError 后通常要重新 raise，否则上层会误以为任务正常结束。

9.9 在异步程序中运行阻塞函数

错误示例：

python

async def bad() -> None:
    time.sleep(5)  # 阻塞整个事件循环

普通阻塞 I/O 可以放入线程：

python

def read_large_file() -> str:
    time.sleep(1)
    return "data"


async def main() -> None:
    data = await asyncio.to_thread(read_large_file)
    print(data)

CPU 密集型工作不是 asyncio 的优势，应考虑进程池、独立 Worker 或专门计算服务。

9.10 并发限制

Agent 同时调用过多工具或模型，容易触发限流和资源耗尽。可用信号量限制并发：

python

semaphore = asyncio.Semaphore(3)


async def limited_call(name: str) -> str:
    async with semaphore:
        return await call_tool(name, 1)


async def main() -> None:
    results = await asyncio.gather(
        *(limited_call(f"tool-{i}") for i in range(10))
    )
    print(results)

这表示最多同时执行 3 个调用。

10. Iterator、Generator 与流式数据

10.1 Iterable 与 Iterator

Iterable：可以被遍历的对象，能产生 Iterator。
Iterator：保存当前遍历位置，每次 next() 给出下一个值。

python

numbers = [10, 20, 30]
iterator = iter(numbers)

print(next(iterator))  # 10
print(next(iterator))  # 20
print(next(iterator))  # 30

再调用一次会抛 StopIteration。普通 for 循环会自动处理它。

10.2 `yield`：创建生成器

python

from collections.abc import Iterator


def numbers() -> Iterator[int]:
    yield 1
    yield 2
    yield 3


for number in numbers():
    print(number)

调用 numbers() 不会立刻执行完整函数，而是返回生成器；每次取值时，代码运行到下一个 yield 后暂停，并保留局部状态。

10.3 惰性处理大数据

一次构造全部结果：

python

def eager_squares(limit: int) -> list[int]:
    return [n * n for n in range(limit)]

逐个产生：

python

def lazy_squares(limit: int):
    for n in range(limit):
        yield n * n

生成器适合大文件、分页查询、事件流和模型 Token 流，因为不必一次把所有内容装进内存。

10.4 生成器表达式

python

squares = (n * n for n in range(1_000_000))
first_three = [next(squares) for _ in range(3)]
print(first_three)

圆括号生成惰性生成器；方括号生成完整列表。

10.5 `yield from`

python

def child():
    yield "a"
    yield "b"


def parent():
    yield "start"
    yield from child()
    yield "end"


print(list(parent()))
# ['start', 'a', 'b', 'end']

yield from 可以把另一个可迭代对象的内容继续向外输出。

10.6 异步生成器与 `async for`

python

import asyncio
from collections.abc import AsyncIterator


async def token_stream() -> AsyncIterator[str]:
    for token in ["Hello", ",", " world"]:
        await asyncio.sleep(0.1)
        yield token


async def main() -> None:
    async for token in token_stream():
        print(token, end="", flush=True)

这就是 LLM Streaming 常见结构：上游等待异步数据，每到一个 chunk 就向下游产出一个 chunk。

10.7 Generator 不等于完整结果

python

stream = token_stream()
print(stream)  # 异步生成器对象，不是 token 列表

如果需要完整内容，必须消费并聚合：

python

async def collect() -> str:
    chunks = [chunk async for chunk in token_stream()]
    return "".join(chunks)

在 Harness 中要明确接口返回的是“最终结果”还是“可消费事件流”，不要让调用方猜。

11. Python 工程结构

11.1 Module 与 Package

一个 .py 文件就是一个 module。包含 __init__.py 的目录通常作为 package 使用。

text

agent_demo/
├── pyproject.toml
├── src/
│   └── agent_demo/
│       ├── __init__.py
│       ├── main.py
│       ├── models.py
│       └── tools/
│           ├── __init__.py
│           └── search.py
└── tests/
    └── test_models.py

Java 对照：

Java	Python
`.java` 文件	`.py` module
package 目录	Python package
`import com.demo.User`	`from agent_demo.models import User`
Maven 标准目录	`src/` layout

11.2 Import

python

import json
from pathlib import Path
from agent_demo.models import Message

同一 package 内可以相对导入：

python

from .models import Message
from .tools.search import SearchTool

应用代码更推荐清晰的绝对导入。不要使用 from module import *，因为来源不清楚，还可能发生命名覆盖。

11.3 `name == "main"`

python

def main() -> None:
    print("application started")


if __name__ == "__main__":
    main()

直接执行文件时，__name__ 是 "__main__"；被其他模块导入时则不是。因此导入不会意外启动程序。

模块化项目更常从项目根目录执行：

bash

python -m agent_demo.main

11.4 虚拟环境

虚拟环境把不同项目的 Python 依赖隔离开，类似每个 Java 项目有自己解析出的依赖集合。

传统方式：

bash

python -m venv .venv

Windows 激活：

powershell

.venv\Scripts\Activate.ps1

Linux / macOS 激活：

bash

source .venv/bin/activate

使用 uv run 时，很多命令不必手动激活环境。

11.5 `pyproject.toml`

toml

[project]
name = "agent-demo"
version = "0.1.0"
description = "A minimal agent harness"
requires-python = ">=3.11"
dependencies = [
    "httpx>=0.27",
    "pydantic>=2.8",
]

[dependency-groups]
dev = [
    "pytest>=8",
    "pytest-asyncio>=0.24",
    "ruff>=0.6",
    "mypy>=1.11",
]

[tool.pytest.ini_options]
pythonpath = ["src"]
asyncio_mode = "auto"

它可以同时承载项目元数据、依赖和工具配置，类似 pom.xml 加上一部分 Checkstyle、测试插件配置。

11.6 使用 uv 管理项目

创建项目：

bash

uv init agent-demo
cd agent-demo

添加依赖：

bash

uv add pydantic httpx
uv add --dev pytest pytest-asyncio ruff mypy

运行：

bash

uv run python -m agent_demo.main
uv run pytest
uv run ruff check .

同步锁定依赖：

bash

uv sync

uv.lock 应提交到应用项目的 Git 仓库，以便团队和 CI 安装一致版本。

11.7 环境变量与配置

python

import os

api_key = os.environ["MODEL_API_KEY"]       # 缺失时立即抛 KeyError
timeout = int(os.getenv("MODEL_TIMEOUT", "30"))

不要把密钥写进代码、Markdown、日志或 Git。.env 适合本地开发，但需要加入 .gitignore。

复杂配置建议用 Pydantic Settings 等专门配置模型，启动时一次校验，避免运行到一半才发现缺参数。

11.8 常用工程命令

bash

uv run ruff format .
uv run ruff check .
uv run mypy src
uv run pytest -q

对应 Java 开发的直觉：

Java 工具职责	Python 常用选择
Maven / Gradle	uv
Checkstyle / Spotless	Ruff
编译期类型检查	mypy / pyright
JUnit	pytest
SLF4J + Logback	`logging` / structlog

12. Agent / Harness 开发常用 Python

12.1 TypedDict：约束字典结构

python

from typing import TypedDict, Literal


class Message(TypedDict):
    role: Literal["system", "user", "assistant", "tool"]
    content: str


messages: list[Message] = [
    {"role": "user", "content": "上海天气如何？"}
]

TypedDict 仍然是普通字典，只给静态检查器提供结构信息；运行时不会自动校验外部 JSON。

12.2 Literal、Enum 与 Union

python

from enum import Enum
from typing import Literal


FinishReason = Literal["stop", "length", "tool_call"]


class TaskStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

Literal 适合少量固定字符串；Enum 适合需要复用、遍历或携带行为的状态集合。

Union 表示多种可能类型：

python

ToolOutput = str | dict[str, object]

12.3 Pydantic `BaseModel`

可以把 Pydantic 粗略理解为：

text

DTO + Lombok + Jackson + Bean Validation

python

from pydantic import BaseModel, Field


class SearchArguments(BaseModel):
    query: str = Field(min_length=1, max_length=200)
    limit: int = Field(default=5, ge=1, le=20)


arguments = SearchArguments.model_validate({
    "query": "Python asyncio",
    "limit": 10,
})

print(arguments.query)
print(arguments.model_dump())
print(arguments.model_dump_json())

外部数据校验失败会抛 ValidationError：

python

from pydantic import ValidationError

try:
    SearchArguments.model_validate({"query": "", "limit": 100})
except ValidationError as exc:
    print(exc.errors())

常见方法：

Pydantic v2	含义
`model_validate(data)`	从 Python 对象解析并校验
`model_validate_json(raw)`	从 JSON 字符串解析并校验
`model_dump()`	转成 dict
`model_dump_json()`	转成 JSON 字符串
`model_json_schema()`	生成 JSON Schema

最后一项对 Tool Calling 特别重要，因为模型的工具参数定义通常就是 JSON Schema。

12.4 Pydantic 嵌套模型

python

from typing import Literal
from pydantic import BaseModel


class ToolCall(BaseModel):
    id: str
    name: str
    arguments: dict[str, object]


class AssistantMessage(BaseModel):
    role: Literal["assistant"] = "assistant"
    content: str | None = None
    tool_calls: list[ToolCall] = []

这里的 Pydantic 模型字段默认值会被模型框架安全处理，但在普通函数和普通类中仍应遵守“不要使用可变默认值”的规则。为了表达意图更明确，也可以写：

python

from pydantic import Field

tool_calls: list[ToolCall] = Field(default_factory=list)

12.5 自定义字段校验

python

from pydantic import BaseModel, field_validator


class ModelConfig(BaseModel):
    model: str
    temperature: float = 0.0

    @field_validator("model")
    @classmethod
    def model_must_not_be_blank(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("model 不能为空")
        return value

边界数据进入 Harness 时立即校验，比在执行深处遇到模糊的 KeyError 或类型错误更容易定位。

12.6 Decorator：装饰器

装饰器接收一个函数，并返回包装后的函数：

python

from collections.abc import Callable
from functools import wraps
from time import perf_counter
from typing import TypeVar, ParamSpec

P = ParamSpec("P")
R = TypeVar("R")


def timed(func: Callable[P, R]) -> Callable[P, R]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            print(f"{func.__name__} took {perf_counter() - start:.3f}s")

    return wrapper


@timed
def calculate() -> int:
    return sum(range(1000))

可以类比 Java 注解加 AOP，但 Python 装饰器本身就是运行时函数包装，不需要 Spring 容器。

异步函数必须使用异步 wrapper：

python

def async_timed(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = perf_counter()
        try:
            return await func(*args, **kwargs)
        finally:
            print(f"{func.__name__} took {perf_counter() - start:.3f}s")

    return wrapper

12.7 JSON

python

import json

raw = '{"name": "Tom", "age": 18}'
data = json.loads(raw)       # str → Python dict
encoded = json.dumps(data)   # Python dict → str

中文输出：

python

text = json.dumps(
    {"message": "你好"},
    ensure_ascii=False,
    indent=2,
)
print(text)

记忆方法：

text

loads：从 String 加载
dumps：转储成 String

不要直接相信模型返回的 JSON。至少捕获 JSONDecodeError，更推荐解析后再交给 Pydantic 校验。

12.8 使用 HTTPX 调用 HTTP API

异步客户端：

python

import httpx


async def call_api(url: str, payload: dict[str, object]) -> dict[str, object]:
    timeout = httpx.Timeout(30.0, connect=5.0)

    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.post(url, json=payload)
        response.raise_for_status()
        return response.json()

高频调用时不要每次都创建新客户端，应在应用生命周期内复用连接池：

python

class ApiClient:
    def __init__(self) -> None:
        self._client = httpx.AsyncClient(timeout=30)

    async def close(self) -> None:
        await self._client.aclose()

    async def post(self, url: str, payload: dict[str, object]):
        response = await self._client.post(url, json=payload)
        response.raise_for_status()
        return response.json()

请求重试要区分错误：连接失败、429、部分 5xx 可能适合退避重试；参数错误、权限错误通常不应盲目重试。

12.9 pytest

普通测试：

python

def add(a: int, b: int) -> int:
    return a + b


def test_add() -> None:
    assert add(1, 2) == 3

参数化测试：

python

import pytest


@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("1", 1),
        ("10", 10),
        ("-1", -1),
    ],
)
def test_parse_int(raw: str, expected: int) -> None:
    assert int(raw) == expected

异步测试：

python

import pytest


@pytest.mark.asyncio
async def test_call_tool() -> None:
    result = await call_tool("search", 0)
    assert result == "search:done"

测试异常：

python

def test_invalid_temperature() -> None:
    with pytest.raises(ValueError, match="temperature"):
        validate_temperature(3)

12.10 Mock 异步依赖

python

from unittest.mock import AsyncMock


async def ask(client, prompt: str) -> str:
    return await client.complete(prompt)


async def test_ask() -> None:
    client = AsyncMock()
    client.complete.return_value = "answer"

    result = await ask(client, "question")

    assert result == "answer"
    client.complete.assert_awaited_once_with("question")

Agent 测试中应尽量 Mock 模型和外部工具，让单元测试稳定、快速、无需真实密钥。再用少量集成测试验证真实链路。

12.11 Logging

python

import logging

logger = logging.getLogger(__name__)


def execute_tool(tool_name: str, arguments: dict[str, object]) -> str:
    logger.info(
        "tool execution started",
        extra={"tool_name": tool_name},
    )
    try:
        return "ok"
    except Exception:
        logger.exception("tool execution failed")
        raise

启动入口统一配置日志：

python

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)

生产 Agent 日志建议包含 trace_id、turn_id、tool_call_id、工具名、耗时和结果状态。不要记录 API Key、完整用户隐私数据或未经裁剪的超长 Prompt。

12.12 一个最小 Tool Registry

下面把 Protocol、Pydantic、异步和异常边界组合起来：

python

import asyncio
from typing import Protocol
from pydantic import BaseModel, Field, ValidationError


class ToolExecutionError(RuntimeError):
    pass


class SearchArgs(BaseModel):
    query: str = Field(min_length=1)
    limit: int = Field(default=5, ge=1, le=20)


class AsyncTool(Protocol):
    name: str

    async def run(self, raw_arguments: dict[str, object]) -> str:
        ...


class SearchTool:
    name = "search"

    async def run(self, raw_arguments: dict[str, object]) -> str:
        args = SearchArgs.model_validate(raw_arguments)
        await asyncio.sleep(0.1)
        return f"搜索 {args.query!r}，最多返回 {args.limit} 条"


class ToolRegistry:
    def __init__(self, tools: list[AsyncTool]):
        self._tools = {tool.name: tool for tool in tools}

    async def execute(
        self,
        tool_name: str,
        arguments: dict[str, object],
    ) -> str:
        tool = self._tools.get(tool_name)
        if tool is None:
            raise ToolExecutionError(tool_name, "工具不存在")

        try:
            async with asyncio.timeout(10):
                return await tool.run(arguments)
        except ValidationError as exc:
            raise ToolExecutionError(tool_name, f"参数错误：{exc}") from exc
        except TimeoutError as exc:
            raise ToolExecutionError(tool_name, "执行超时") from exc


async def main() -> None:
    registry = ToolRegistry([SearchTool()])
    result = await registry.execute(
        "search",
        {"query": "asyncio TaskGroup", "limit": 3},
    )
    print(result)

这个例子体现了 Harness 中非常重要的边界：

模型只负责给出 tool_name + arguments。
Registry 决定工具是否存在。
Pydantic 校验不可信参数。
Harness 负责超时和异常归一化。
Tool 只实现自己的业务逻辑。

12.13 一个最小 Agent Loop

以下代码不绑定具体模型 SDK，只展示控制流：

python

from dataclasses import dataclass, field
from typing import Literal


@dataclass
class ToolRequest:
    name: str
    arguments: dict[str, object]


@dataclass
class ModelDecision:
    kind: Literal["final", "tool"]
    content: str | None = None
    tool_request: ToolRequest | None = None


@dataclass
class AgentState:
    messages: list[dict[str, object]] = field(default_factory=list)
    step: int = 0


async def run_agent(model, registry: ToolRegistry, user_input: str) -> str:
    state = AgentState(
        messages=[{"role": "user", "content": user_input}]
    )

    max_steps = 8

    while state.step < max_steps:
        state.step += 1
        decision: ModelDecision = await model.decide(state.messages)

        if decision.kind == "final":
            if decision.content is None:
                raise RuntimeError("final decision 缺少 content")
            return decision.content

        request = decision.tool_request
        if request is None:
            raise RuntimeError("tool decision 缺少 tool_request")

        try:
            observation = await registry.execute(
                request.name,
                request.arguments,
            )
        except ToolExecutionError as exc:
            observation = f"ERROR: {exc}"

        state.messages.append({
            "role": "assistant",
            "tool_call": {
                "name": request.name,
                "arguments": request.arguments,
            },
        })
        state.messages.append({
            "role": "tool",
            "name": request.name,
            "content": observation,
        })

    raise RuntimeError(f"Agent 超过最大步数 {max_steps}")

这段代码最需要记住的不是语法，而是控制权：模型提出动作，Harness 校验并执行动作，再把结构化 Observation 放回上下文。max_steps 是防止死循环的最基本 Limit。

13. Java 开发者高频踩坑清单

13.1 缩进就是语法

python

if enabled:
    run_task()  # 必须缩进

格式化工具能减少缩进和样式问题，但代码块层级仍需开发者正确表达。

13.2 没有 `++` 和 `--`

python

count += 1
count -= 1

13.3 `append` 和 `extend` 不一样

python

items = [1, 2]
items.append([3, 4])
print(items)  # [1, 2, [3, 4]]

items = [1, 2]
items.extend([3, 4])
print(items)  # [1, 2, 3, 4]

13.4 可变对象是引用传递效果

严格说，Python 传递的是对象引用的值。可变对象能被函数内部修改：

python

def add_message(messages: list[str]) -> None:
    messages.append("hello")


messages: list[str] = []
add_message(messages)
print(messages)  # ['hello']

重新绑定局部变量不会替换外部变量：

python

def reset(messages: list[str]) -> None:
    messages = []

13.5 不要用 `is` 比较普通值

python

name == "Tom"   # 正确：比较值
name is None    # 正确：检查 None

13.6 `dict.get()` 可能掩盖字段缺失

python

model = config.get("model")

如果 model 是必填项，上面会把配置错误延迟到更深处。更合适的方式是 config["model"] 让它立即失败，或使用 Pydantic 在入口统一校验。

13.7 捕获 `Exception` 后不要静默

错误：

python

try:
    await run_tool()
except Exception:
    pass

至少应该记录上下文并决定重试、转换或继续抛出：

python

try:
    await run_tool()
except TimeoutError as exc:
    logger.warning("tool timed out", exc_info=exc)
    raise

13.8 `await` 不等于并发

python

a = await task_a()
b = await task_b()

这仍然是串行。只有先创建多个 Task、使用 gather 或 TaskGroup，等待时间才可能重叠。

13.9 异步程序不能混入阻塞调用

time.sleep()、同步 HTTP 请求和耗时文件操作都会卡住事件循环。使用对应异步库，或把无法替换的阻塞函数放进 asyncio.to_thread()。

13.10 类型标注不是运行时校验

python

def add(a: int, b: int) -> int:
    return a + b

print(add("1", "2"))  # 运行时仍可能得到 '12'

类型标注需要 mypy / pyright 检查；外部输入需要 Pydantic 等运行时校验。

13.11 列表推导式不要写成谜语

下面虽然能写，但难维护：

python

result = [f(x) for group in groups for x in group if p(x) and q(x)]

业务逻辑复杂时展开成普通循环，并给中间步骤命名。

13.12 不要依赖全局可变状态

Web 服务、测试、并发 Agent 和多 Session 场景都会让全局状态变得危险。把状态明确放在对象、请求上下文或持久化存储中，并标清生命周期。

14. 推荐学习顺序

第一阶段：先能看懂代码

变量、None、类型标注。
条件、Truthy/Falsy、is 与 ==。
for、range、enumerate、推导式。
List、Dict、Tuple、切片和解包。
函数、默认参数、*args、**kwargs。
Class、self、dataclass。

完成标准：能够把一段普通 Python 业务代码翻译成自己熟悉的 Java 逻辑。

第二阶段：进入 Python 工程

Module、Package、Import。
pyproject.toml 和 uv。
pytest、Ruff、mypy。
with、异常处理、日志。
配置与环境变量。

完成标准：能独立创建、运行、测试和检查一个 Python 项目。

第三阶段：进入 Agent / Harness

async / await、Task、TaskGroup、Timeout、Cancellation。
Iterator、Generator、异步流。
Pydantic 与 JSON Schema。
Protocol、依赖注入和 Tool Registry。
Agent Loop、Limit、Observation 和错误归一化。

完成标准：能实现一个有超时、参数校验、工具注册、最大步数和测试的最小 Agent Harness。

15. 最后背下来的核心符号

text

null                         None
true / false                 True / False
&& / || / !                  and / or / not

List<T>                      list[T]
Map<K, V>                    dict[K, V]
Set<T>                       set[T]

list.add(x)                  items.append(x)
list.size()                  len(items)
map.entrySet()               mapping.items()

for (X x : xs)               for x in xs:
index + value                 enumerate(xs)
parallel iteration           zip(xs, ys)

Object... args               *args
Map<String, Object>          **kwargs
nullable String              str | None

this                         self
constructor                  __init__
static                       @staticmethod
factory method               @classmethod
Lombok data object           @dataclass

try-with-resources            with
CompletableFuture / Reactor   async / await
allOf                         gather / TaskGroup
Stream / lazy iterator        generator / yield

DTO + Jackson + Validation    Pydantic BaseModel
JUnit                         pytest
Maven / Gradle                uv
pom.xml                       pyproject.toml

真正进入 Agent / Harness 开发后，最重要的不是追求“Python 写得短”，而是把边界写清楚：外部数据有校验，工具执行有超时，异常可以追踪，并发受到限制，状态有明确归属，每一轮动作都有可观察结果。

Java 开发者的 Python 快速迁移指南 ​

阅读方式 ​

一张表先建立整体映射 ​

01. 变量与基础类型 ​

1.1 变量不需要先声明类型 ​

1.2 常用基础类型 ​

1.3 None：对应 Java 的 null ​

1.4 布尔值和逻辑运算 ​

1.5 字符串与 f-string ​

1.6 类型标注 Type Hints ​

1.7 Any 与 object ​

02. 条件判断 ​

2.1 if / elif / else ​

2.2 in：包含判断 ​

2.3 == 与 is ​

2.4 Truthy / Falsy ​

2.5 or 提供默认值 ​

2.6 三元表达式 ​

2.7 match：结构化分支 ​

03. 循环 ​

3.1 for：遍历可迭代对象 ​

3.2 range：按次数循环 ​

3.3 while ​

3.4 enumerate：同时获得下标和值 ​

3.5 zip：并行遍历多个集合 ​

3.6 break、continue 与循环的 else ​

3.7 列表推导式 ​

04. 集合 ​

4.1 List：对应 Java ArrayList ​

4.2 Dict：对应 Java HashMap ​

4.3 Set：去重与成员判断 ​

4.4 Tuple：固定结构的不可变序列 ​

4.5 切片 Slice ​

4.6 解包 Unpacking ​

4.7 排序 ​

4.8 引用、浅拷贝与深拷贝 ​

05. 函数与参数 ​

5.1 定义函数 ​

5.2 返回多个值 ​

5.3 默认参数 ​

可变默认值陷阱 ​

5.4 命名参数 ​

5.5 仅限命名参数 ​

5.6 *args：可变位置参数 ​

5.7 **kwargs：可变命名参数 ​

5.8 用 * 和 ** 拆开集合 ​

5.9 Lambda ​

5.10 作用域与 nonlocal / global ​

5.11 Python 没有传统方法重载 ​

06. 类与对象 ​

6.1 定义类、构造对象与 self ​

6.2 类属性与实例属性 ​

6.3 @property：像字段一样访问的方法 ​

6.4 继承与 super() ​

6.5 @staticmethod ​

6.6 @classmethod ​

6.7 @dataclass：接近 Lombok 数据类 ​

6.8 Dunder / Magic Method ​

6.9 抽象类与 Protocol ​

07. 异常处理 ​

7.1 try / except / else / finally ​

7.2 多种异常 ​

7.3 raise：主动抛异常 ​

7.4 自定义异常 ​

7.5 异常链 raise ... from ... ​

08. 资源与上下文管理 ​

8.1 with：对应 try-with-resources ​

8.2 pathlib：更现代的路径 API ​

8.3 自定义 Context Manager ​

8.4 基于类的 Context Manager ​

8.5 async with ​

09. 等待、异步与并发 ​

9.1 同步等待与异步等待的区别 ​

9.2 async def 返回协程对象 ​

9.3 串行 await ​

9.4 create_task：启动并发任务 ​

9.5 gather：并发等待多个结果 ​

9.6 TaskGroup：结构化并发 ​

9.7 超时 ​

9.8 取消任务 ​

Java 开发者的 Python 快速迁移指南

阅读方式

一张表先建立整体映射

01. 变量与基础类型

1.1 变量不需要先声明类型

1.2 常用基础类型

1.3 `None`：对应 Java 的 `null`

1.4 布尔值和逻辑运算

1.5 字符串与 f-string

1.6 类型标注 Type Hints

1.7 `Any` 与 `object`

02. 条件判断

2.1 `if / elif / else`

2.2 `in`：包含判断

2.3 `==` 与 `is`

2.4 Truthy / Falsy

2.5 `or` 提供默认值

2.6 三元表达式

2.7 `match`：结构化分支

03. 循环

3.1 `for`：遍历可迭代对象

3.2 `range`：按次数循环

3.3 `while`

3.4 `enumerate`：同时获得下标和值

3.5 `zip`：并行遍历多个集合

3.6 `break`、`continue` 与循环的 `else`

3.7 列表推导式

04. 集合

4.1 List：对应 Java `ArrayList`

4.2 Dict：对应 Java `HashMap`

4.3 Set：去重与成员判断

4.4 Tuple：固定结构的不可变序列

4.5 切片 Slice

4.6 解包 Unpacking

4.7 排序

4.8 引用、浅拷贝与深拷贝

05. 函数与参数

5.1 定义函数

5.2 返回多个值

5.3 默认参数

可变默认值陷阱

5.4 命名参数

5.5 仅限命名参数

5.6 `*args`：可变位置参数

5.7 `**kwargs`：可变命名参数

5.8 用 `*` 和 `**` 拆开集合

5.9 Lambda

5.10 作用域与 `nonlocal` / `global`

5.11 Python 没有传统方法重载

06. 类与对象

6.1 定义类、构造对象与 `self`

6.2 类属性与实例属性

6.3 `@property`：像字段一样访问的方法

6.4 继承与 `super()`

6.5 `@staticmethod`

6.6 `@classmethod`

6.7 `@dataclass`：接近 Lombok 数据类

6.8 Dunder / Magic Method

6.9 抽象类与 Protocol

07. 异常处理

7.1 `try / except / else / finally`

7.2 多种异常

7.3 `raise`：主动抛异常

7.4 自定义异常

7.5 异常链 `raise ... from ...`

08. 资源与上下文管理

8.1 `with`：对应 try-with-resources

8.2 `pathlib`：更现代的路径 API

8.3 自定义 Context Manager

8.4 基于类的 Context Manager

8.5 `async with`

09. 等待、异步与并发

9.1 同步等待与异步等待的区别

9.2 `async def` 返回协程对象

9.3 串行 `await`

9.4 `create_task`：启动并发任务

9.5 `gather`：并发等待多个结果

9.6 `TaskGroup`：结构化并发

9.7 超时

9.8 取消任务