0-pydanticai-comm-agent.py 代码说明

这份脚本是一个很小、但很完整的 Agent 例子。它想让同学们看懂三件事：

• 什么叫“agent 不是只会聊天”
• 什么叫“模型可以调用工具”
• 什么叫“普通 Python 函数也可以变成 agent 工具”

这份代码做的事情很简单：

1. 读取一个通信题目
2. 把题目交给本地 Ollama 上的 qwen3.5:0.8b
3. 当模型需要计算时，调用本地工具函数
4. 最后输出一个中文分析结论

所以它不是一个“大模型训练脚本”，而是一个“让模型学会调用工具做事”的脚本。

先看整体结构

这份脚本可以分成 6 块：

1. 导入依赖
2. 定义默认题目
3. 定义几个辅助函数
4. 写一个 Ollama 适配层
5. 用 PydanticAI 定义 agent 和工具
6. 写命令行入口 main()

你可以把它理解成：

• 前半段在解决“怎么接本地模型”
• 后半段在解决“怎么把工具挂给 agent”

第一部分：导入依赖

from __future__ import annotations

这一行的作用是：

• 让类型注解延迟求值
• 写 Agent[None, str]、list[dict[str, Any]] 这种类型时更方便

对初学者来说，你可以先简单理解成：

• 这是为了让后面的类型标注写得更自然

下面这些是 Python 标准库：

import argparse
import json
import math
import os
import sys
import urllib.error
import urllib.request
from typing import Any

它们分别做什么：

• argparse
  • 处理命令行参数
• json
  • 把 Python 字典变成 JSON
  • 也把 JSON 响应解析回 Python
• math
  • 做对数、幂、香农容量公式等数学计算
• os
  • 处理环境变量
• sys
  • 把错误信息打印到终端错误输出
• urllib.request
  • 向本地 Ollama 发送 HTTP 请求
• urllib.error
  • 处理 HTTP 错误和网络错误
• Any
  • 表示“这里的类型可以比较宽”

下面这些是 PydanticAI 相关导入：

from pydantic_ai import Agent
from pydantic_ai.messages import (
    ModelMessage,
    ModelRequest,
    ModelResponse,
    RetryPromptPart,
    SystemPromptPart,
    TextPart,
    ToolCallPart,
    ToolReturnPart,
    UserPromptPart,
)
from pydantic_ai.models.function import AgentInfo, FunctionModel

这里最重要的是 3 个概念：

• Agent
  • 这是主角
  • 你可以把它理解成“一个会和模型、工具、上下文一起工作”的对象
• FunctionModel
  • 它表示“模型由一个本地函数来驱动”
  • 这里非常关键，因为我们不是直接接 OpenAI 官方 API，而是自己把请求转给本地 Ollama
• ModelRequest / ModelResponse
  • 这些类表示 agent 和模型之间传来传去的消息

第二部分：默认题目 DEFAULT_PROMPT

DEFAULT_PROMPT = (
    "请分析一个 2.4 GHz 校园无线链路..."
)

这一段不是普通文本而已，它其实在做两件事：

1. 给模型一个明确任务
2. 明确要求“必须调用工具”

里面的条件包括：

• 频率 2.4 GHz
• 距离 1.2 km
• 发射功率 20 dBm
• 发射和接收天线增益各 2 dBi
• 其他损耗 3 dB
• 带宽 20 MHz
• 噪声系数 5 dB

后面还要求模型输出：

• 路径损耗
• 接收功率
• 噪声底
• SNR
• Shannon 容量
• 调制建议

为什么这里要写得这么明确？

因为 agent 不是“你心里想什么，它都能猜到”。

如果 prompt 写得含糊：

• 模型可能不调工具
• 也可能少算一些量
• 还可能直接凭感觉编一个答案

第三部分：辅助函数

clear_proxy_env()

def clear_proxy_env() -> None:

这个函数的作用是：

• 清理系统里的代理环境变量

里面删掉了这些变量：

"http_proxy",
"https_proxy",
"HTTP_PROXY",
"HTTPS_PROXY",
"all_proxy",
"ALL_PROXY",

为什么要删？

因为我们要访问的是本地：

http://localhost:11434/api/chat

如果电脑上开了代理，Python 请求有时会把本地请求也错误地走代理，结果就会：

• 连不上本地 Ollama
• 或者报奇怪的网络错误

所以这一步是一个很实用的“避坑动作”。

debug_log(label, payload)

def debug_log(label: str, payload: Any) -> None:

这个函数是调试开关。

只有当环境变量：

AGENT_DEMO_DEBUG=1

时，它才会打印额外信息。

它主要用来打印：

• 发给 Ollama 的请求
• Ollama 返回的响应

这对于调试 agent 特别重要，因为 agent 出问题时，最常见的两个问题就是：

• 工具 schema 发错了
• 模型返回的 tool_calls 长得不对

所以这个函数其实是在告诉同学们：

• agent 开发不是只写 prompt
• 你还得会看请求和响应

to_text(value)

def to_text(value: Any) -> str:

这个函数的作用是：

• 把任意工具结果统一变成文本

逻辑很简单：

if isinstance(value, str):
    return value
return json.dumps(value, ensure_ascii=False)

意思是：

• 如果本来就是字符串，直接返回
• 如果是字典、列表、数字等，就转成 JSON 字符串

为什么要做这一步？

因为发给 Ollama 的消息内容，最终都要是文本形式。

比如工具返回：

{
  "path_loss_db": 101.638,
  "snr_db": 15.352
}

就要先转成一段 JSON 文本，模型才能“看到”这个结果。

第四部分：把 PydanticAI 的消息翻译成 Ollama 能懂的格式

build_ollama_messages(messages)

def build_ollama_messages(messages: list[ModelMessage]) -> list[dict[str, Any]]:

这是整份脚本里最核心的函数之一。

它在做的事是：

• 把 PydanticAI 内部的消息对象
• 翻译成 Ollama /api/chat 需要的消息列表

你可以把它理解成：

• PydanticAI 说的是一种“内部语言”
• Ollama 说的是另一种“接口语言”
• 这个函数就是“翻译器”

先创建一个空列表

ollama_messages: list[dict[str, Any]] = []

意思是：

• 准备一个列表
• 后面把每条消息都一条条塞进去

这里的类型：

• list[...]
  • 表示列表
• dict[str, Any]
  • 表示每个元素是一个字典
  • 键是字符串
  • 值类型可以比较灵活

遍历每一条消息

for message in messages:

意思是：

• messages 里可能有很多条消息
• 例如 system prompt、用户问题、工具返回、模型回应
• 我们把它们一条条拿出来处理

如果它是 ModelRequest

if isinstance(message, ModelRequest):

isinstance(x, y) 的意思是：

• 判断 x 是不是 y 这个类型

这里判断的是：

• 当前这条消息是不是“发给模型的请求”

如果是请求，就继续拆开看里面每个部分 part。

SystemPromptPart

if isinstance(part, SystemPromptPart):
    ollama_messages.append({"role": "system", "content": part.content})

意思是：

• 如果这是系统提示词
• 就翻译成 Ollama 里的：

{"role": "system", "content": "..."}

这里的 append(...) 表示：

• 把一条新消息加到列表末尾

UserPromptPart

elif isinstance(part, UserPromptPart):
    ollama_messages.append({"role": "user", "content": to_text(part.content)})

意思是：

• 如果这是用户输入
• 就翻译成 role = user

这里专门用了：

to_text(part.content)

因为用户输入有时不一定是单纯字符串，所以先统一转成文本更稳。

ToolReturnPart

elif isinstance(part, ToolReturnPart):
    ollama_messages.append(
        {
            "role": "tool",
            "tool_name": part.tool_name,
            "content": to_text(part.content),
        }
    )

这段特别重要。

它表示：

• 工具已经执行完了
• 现在要把工具结果再送回模型

这里会告诉模型：

• 这是一个 tool 消息
• 是哪个工具返回的
• 返回内容是什么

为什么 agent 能继续多轮工作？

就是因为：

• 模型先发起 tool_call
• Python 真去执行工具
• 然后再把结果作为 tool 消息喂回模型

RetryPromptPart

elif isinstance(part, RetryPromptPart):
    ollama_messages.append({"role": "user", "content": to_text(part.content)})

这个部分表示：

• 工具参数验证失败了
• 或者模型需要重新来一次

这时 PydanticAI 会生成一个“重试提示”，再送给模型。

对初学者来说，这说明一个很重要的点：

• agent 不一定一次就成功
• 框架会帮你做纠错和重试

处理 ModelResponse

elif isinstance(message, ModelResponse):

这表示：

• 当前不是“发给模型的消息”
• 而是“模型已经返回的消息”

这时我们要把模型返回里的两类东西提取出来：

• 普通文本
• 工具调用

收集文本块

text_chunks: list[str] = []
tool_calls: list[dict[str, Any]] = []

这两个列表分别存：

• 文本内容
• 工具调用信息

然后遍历每个 part：

if isinstance(part, TextPart):
    text_chunks.append(part.content)

意思是：

• 如果模型说了一段普通文字
• 就先存起来

收集工具调用

elif isinstance(part, ToolCallPart):

这表示：

• 模型不是在直接回答
• 而是在说“请帮我调用某个工具”

然后会把它转成：

{
  "id": part.tool_call_id,
  "type": "function",
  "function": {
      "name": part.tool_name,
      "arguments": part.args if part.args is not None else {},
  },
}

这里几项分别是什么意思：

• id
  • 这次工具调用的编号
• type
  • 这里固定是 function
• name
  • 要调用哪个工具
• arguments
  • 这个工具的参数

这就是“模型发起工具调用”的核心格式。

把 assistant 消息重新组起来

if text_chunks or tool_calls:

意思是：

• 只要这条模型回应里有文本或有工具调用
• 我们就把它组装成一条 assistant 消息

item: dict[str, Any] = {
    "role": "assistant",
    "content": "\n".join(chunk for chunk in text_chunks if chunk).strip(),
}

这里：

• "\n".join(...)
  • 把多段文字用换行拼起来
• .strip()
  • 去掉首尾空白

如果有工具调用，再加上：

item["tool_calls"] = tool_calls

最后：

ollama_messages.append(item)

这样一条完整的 assistant 消息就构建好了。

第五部分：把工具 schema 发给 AI

build_ollama_tools(agent_info)

def build_ollama_tools(agent_info: AgentInfo) -> list[dict[str, Any]]:

这个函数的作用是：

• 把 PydanticAI 里的工具定义
• 转成 Ollama 能识别的 tool schema

agent_info.function_tools 里面装的，就是当前 agent 注册的所有工具。

循环里最关键的是：

"name": tool.name,
"description": tool.description or "",
"parameters": tool.parameters_json_schema,

它们分别表示：

• 工具名
• 工具描述
• 参数的 JSON Schema

这一步很重要，因为模型能不能正确调用工具，很大程度取决于：

• 工具名是不是清楚
• 描述是不是清楚
• 参数 schema 是不是准确

第六部分：真正请求本地 Ollama

call_ollama_chat(...)

def call_ollama_chat(
    model_name: str,
    messages: list[dict[str, Any]],
    tools: list[dict[str, Any]] | None,
    host: str,
) -> dict[str, Any]:

这个函数就是：

• 把请求真正发到本地 Ollama

参数解释：

• model_name
  • 模型名，例如 qwen3.5:0.8b
• messages
  • 前面翻译好的消息列表
• tools
  • 翻译好的工具 schema
• host
  • 本地接口地址，这里默认是：
  • http://localhost:11434/api/chat

构造 payload

payload: dict[str, Any] = {
    "model": model_name,
    "stream": False,
    "messages": messages,
}

这里：

• model
  • 用哪个模型
• stream: False
  • 表示不走流式返回
  • 而是等完整结果一次性返回
• messages
  • 对话历史

如果有工具，再补上：

if tools:
    payload["tools"] = tools

这一步是让模型知道：

• 你现在不仅能回答
• 还可以调这些工具

构造 HTTP 请求

request = urllib.request.Request(
    host,
    data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
    headers={"Content-Type": "application/json"},
    method="POST",
)

逐项解释：

• host
  • 请求地址
• data=...
  • 请求体
• json.dumps(payload, ensure_ascii=False)
  • 把 Python 字典转成 JSON 字符串
  • ensure_ascii=False 表示中文不要转成 \\uXXXX
• .encode("utf-8")
  • 把字符串转成字节
  • 因为网络发送时要发字节
• headers={"Content-Type": "application/json"}
  • 告诉服务端，这是一段 JSON
• method="POST"
  • 说明这是 POST 请求

真正发送请求

clear_proxy_env()
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(request, timeout=120) as response:
    result = json.loads(response.read().decode("utf-8"))

这几行很值得认真看。

clear_proxy_env()

先清理代理，避免本地请求被代理污染。

ProxyHandler({})

这里更进一步，显式告诉 Python：

• 这次请求不要走任何代理

timeout=120

表示最多等 120 秒。

为什么这里要等比较久？

因为本地小模型做工具调用时，有时会思考一阵子，不像普通 HTTP 接口那么快。

response.read().decode("utf-8")

表示：

1. 把服务端返回的原始字节读出来
2. 按 utf-8 解码成字符串

再用：

json.loads(...)

把 JSON 字符串变回 Python 字典。

第七部分：做一个 PydanticAI 的模型适配层

make_ollama_function_model(...)

def make_ollama_function_model(model_name: str, host: str) -> FunctionModel:

这是整份代码最有“agent 框架味道”的部分。

它的作用是：

• 让 PydanticAI 把本地 Ollama /api/chat 当成一个模型来用

为什么需要它？

因为你本机这里的 Ollama：

• /api/chat 是可用的
• 但 /v1 的 OpenAI 兼容接口不稳定

所以我们没有走最省事的 OpenAI provider，而是自己写了一个薄适配层。

内部函数 run_ollama(...)

def run_ollama(messages: list[ModelMessage], agent_info: AgentInfo) -> ModelResponse:

这个函数就是：

• PydanticAI 真正调用的“本地模型函数”

它接收：

• messages
  • agent 当前的消息历史
• agent_info
  • 当前 agent 的工具信息、配置等

第一步：

response = call_ollama_chat(
    model_name=model_name,
    messages=build_ollama_messages(messages),
    tools=build_ollama_tools(agent_info),
    host=host,
)

这是把所有前面写好的小函数串起来：

• 先翻译消息
• 再翻译工具
• 最后发请求给本地 Ollama

从 Ollama 响应里抽取内容

message = response.get("message", {})
parts: list[TextPart | ToolCallPart] = []

这里：

• response.get("message", {})
  • 从响应字典里取出 message
  • 如果没有，就给一个空字典

然后准备一个 parts 列表，后面往里面放：

• 文本
• 工具调用

处理普通文本

content = (message.get("content") or "").strip()
if content:
    parts.append(TextPart(content=content))

意思是：

• 先拿出模型返回的文本
• 如果不为空，就包成 TextPart

TextPart 是 PydanticAI 认识的一种响应片段。

处理工具调用

for tool_call in message.get("tool_calls", []):

如果模型返回里有 tool_calls，就一条条处理。

核心是：

ToolCallPart(
    tool_name=function.get("name", ""),
    args=function.get("arguments", {}),
    tool_call_id=tool_call.get("id", ""),
)

这一步是在告诉 PydanticAI：

• 模型现在不是只想说话
• 它想调用一个工具

然后 PydanticAI 就会：

1. 按 schema 验证参数
2. 真的执行工具函数
3. 再把结果喂回模型

这就是 agent 工作流的关键闭环。

为什么要补一个“兜底文本”

if not parts:
    parts.append(TextPart(content="模型没有返回内容。"))

这是一个防御性写法。

意思是：

• 如果模型既没返回文本，也没返回工具调用
• 至少别让程序直接崩掉

虽然这不是理想情况，但它能帮助我们快速发现问题。

返回 FunctionModel

return FunctionModel(function=run_ollama, model_name=f"ollama:{model_name}")

这一行的意思是：

• 用 run_ollama 这个本地函数
• 包装出一个 PydanticAI 能识别的模型对象

这样后面 Agent(...) 就能直接拿它当模型用了。

第八部分：真正定义 agent

build_agent(...)

def build_agent(model_name: str, host: str) -> Agent[None, str]:

这个函数负责：

• 创建模型
• 创建 agent
• 注册工具

返回类型：

Agent[None, str]

可以先简单理解成：

• 这个 agent 不额外依赖复杂上下文对象
• 最终输出是字符串

先创建模型

model = make_ollama_function_model(model_name=model_name, host=host)

这一步就是：

• 把前面写好的本地 Ollama 适配层接进来

再创建 agent

agent = Agent(
    model=model,
    system_prompt=(...),
    output_type=str,
)

这几个参数很重要：

model=model

• 指定用哪个模型

system_prompt=(...)

• 指定系统提示词
• 决定 agent 的角色和行为规范

这里面写了几条关键规则：

• 你是通信系统助教
• 遇到可计算问题，优先调用工具
• 尽量减少不必要轮次
• 优先使用 analyze_wireless_link 和 suggest_modulation
• 总结时必须严格使用工具返回数值

这说明：

• agent 的行为并不只是代码决定
• system prompt 也是“程序的一部分”

output_type=str

• 表示最后希望输出是普通字符串

第九部分：注册工具

@agent.tool_plain

这个装饰器非常关键。

当你看到：

@agent.tool_plain
def analyze_wireless_link(...):

它的意思是：

• 把下面这个 Python 函数注册成 agent 可以调用的工具

也就是说：

• 这本来只是一个普通函数
• 加了这个装饰器以后，模型就能“请求调用它”

这就是为什么说：

• PydanticAI 特别适合教学
• 因为 tool 看起来就像普通 Python 函数

工具一：analyze_wireless_link(...)

def analyze_wireless_link(
    frequency_ghz: float,
    distance_km: float,
    tx_power_dbm: float,
    tx_gain_dbi: float,
    rx_gain_dbi: float,
    other_loss_db: float,
    bandwidth_mhz: float,
    noise_figure_db: float,
) -> dict[str, float]:

这个工具把一整段链路分析打包在一起。

为什么这样设计？

因为前面测试时发现：

• 如果拆成太多小工具
• 小模型容易把中间量记错
• 也容易多绕几轮

所以这里故意把：

• 路径损耗
• 接收功率
• 噪声底
• SNR
• Shannon 容量

合并成一次工具调用。

这也是工程里很真实的设计思路：

• 工具不是越细越好
• 而是要适合模型稳定调用

逐行看链路分析公式

路径损耗

path_loss = 92.45 + 20 * math.log10(frequency_ghz) + 20 * math.log10(distance_km)

这是一条自由空间路径损耗 FSPL 的常见公式。

这里：

• 92.45
  • 是常数项
• math.log10(...)
  • 是以 10 为底的对数
• 20 * log10(f)
  • 频率越高，路径损耗越大
• 20 * log10(d)
  • 距离越远，路径损耗越大

接收功率

received = tx_power_dbm + tx_gain_dbi + rx_gain_dbi - path_loss - other_loss_db

这就是最基础的链路预算思路：

• 发射功率
• 加上收发两端增益
• 减去路径损耗
• 再减去其他损耗

噪声底

bandwidth_hz = bandwidth_mhz * 1_000_000
noise = -174 + 10 * math.log10(bandwidth_hz) + noise_figure_db

这里：

• bandwidth_mhz * 1_000_000
  • 把 MHz 转成 Hz
• -174
  • 热噪声底的常用近似值，单位是 dBm/Hz
• 10 * log10(bandwidth_hz)
  • 带宽越大，噪声总量越高
• noise_figure_db
  • 接收机噪声系数，会进一步抬高噪声底

SNR

snr_value = received - noise

就是：

• 接收信号功率减去噪声功率

Shannon 容量

capacity_bps = bandwidth_hz * math.log2(1 + 10 ** (snr_value / 10))

这是 Shannon 容量公式：

C = Blog₂(1 + SNR)

这里注意：

• snr_value 现在还是 dB
• 所以要先变成线性值：

10 ** (snr_value / 10)

返回值

return {
    "path_loss_db": round(path_loss, 3),
    "received_power_dbm": round(received, 3),
    "noise_floor_dbm": round(noise, 3),
    "snr_db": round(snr_value, 3),
    "shannon_capacity_mbps": round(capacity_bps / 1_000_000, 3),
}

这里有两个值得注意的点：

1. round(..., 3)
   • 保留 3 位小数
   • 让输出更整齐
2. capacity_bps / 1_000_000
   • 把 bps 转成 Mbps

所以这个工具最后返回的是一个字典，里面装着全部关键指标。

工具二：suggest_modulation(snr_db_value)

def suggest_modulation(snr_db_value: float) -> str:

这个工具比较简单：

• 输入一个 SNR
• 输出一个比较粗略的调制建议

逻辑是 if / elif / return：

if snr_db_value < 6:
    ...
if snr_db_value < 12:
    ...
if snr_db_value < 18:
    ...
if snr_db_value < 24:
    ...
return ...

意思就是：

• 根据信噪比分段
• 给出一个工程上比较容易懂的建议

这里不是在做很严谨的通信标准判决，而是在做课堂演示用的“粗粒度建议工具”。

这也很重要，因为同学们要知道：

• tool 不一定总是特别复杂
• 很多 agent 工具本来就是“小而明确”的函数

第十部分：命令行入口 main()

创建参数解析器

parser = argparse.ArgumentParser(
    description="Run a minimal PydanticAI + Ollama communications agent."
)

这一步是在做命令行接口。

它让你可以在终端里这样运行：

python3 0-pydanticai-comm-agent.py --model qwen3.5:0.8b

--model

parser.add_argument(
    "--model",
    default="qwen3.5:0.8b",
    help="Local Ollama model name, default: qwen3.5:0.8b",
)

含义：

• 参数名：--model
• 默认值：qwen3.5:0.8b

如果不写这个参数，就默认用这个模型。

--host

parser.add_argument(
    "--host",
    default="http://localhost:11434/api/chat",
    help="Local Ollama chat endpoint, default: http://localhost:11434/api/chat",
)

含义：

• 指定本地 Ollama 接口地址
• 默认就是本机 11434 端口

--prompt

parser.add_argument(
    "--prompt",
    default=DEFAULT_PROMPT,
    help="User prompt for the communications agent",
)

含义：

• 允许你在命令行里换一个题目
• 如果不写，就用默认题目

例如：

python3 0-pydanticai-comm-agent.py --prompt "请分析一个 5.8 GHz 的无线链路..."

真正运行 agent

agent = build_agent(model_name=args.model, host=args.host)
result = agent.run_sync(args.prompt)

这两行是主流程：

1. 先创建 agent
2. 再把题目交给它执行

这里的：

run_sync(...)

表示同步运行。

对初学者来说，可以先简单理解成：

• 直接跑
• 等结果出来

异常处理

except urllib.error.HTTPError as exc:

这个表示：

• 服务器返回了 HTTP 错误

比如：

• 地址没写对
• 请求格式不对

except urllib.error.URLError as exc:

这个表示：

• 网络层面没连上

比如：

• 本地 Ollama 没启动
• 端口不对

except TimeoutError:

表示：

• 等太久了
• 本地模型没有在规定时间内返回

这几段异常处理的意义是：

• 别让脚本一出问题就只抛一长串难懂报错
• 而是给学生一个更能看懂的失败原因

输出结果并退出

print(result.output)
return 0

这里：

• result.output
  • 就是 agent 最后的文本输出
• return 0
  • 表示程序正常结束

最后这句：

if __name__ == "__main__":
    raise SystemExit(main())

是 Python 脚本的标准写法，意思是：

• 如果这个文件是被直接运行的
• 就执行 main()

这份脚本最该带走什么

这份代码最想让同学理解的，不是通信公式本身，而是 agent 的结构：

1. 模型不是直接调用工具的
   • 中间要有 tool schema
   • 也要有消息格式转换
2. tool 本质上就是普通 Python 函数
   • 只是被 @agent.tool_plain 注册了
3. agent 的关键不是“多聪明”
   • 而是“什么时候该调工具”
   • “工具结果怎么再送回模型”
4. 本地模型也能做 agent
   • 不一定非要云端 API
   • 这里就是 Ollama + qwen3.5:0.8b
5. 工程上要做适配层
   • 真实世界里，框架和模型接口不一定天然对齐
   • 你经常需要自己写一点“翻译代码”

如果你继续往下升级

这份脚本后面可以继续升级成：

• 支持多轮追问
• 支持结构化输出模型
• 支持更多通信工具
• 接入标准文档或链路预算表
• 加入 memory
• 加入日志、追踪和评测

这就是为什么这份脚本虽然小，但已经很像一个真正 agent 系统的雏形。