多くの人が初めて AI エージェントを作成し、作成後に少しがっかりします。
コードは実際に実行され、モデルは音声を発しますが、その秘密はわずか 2 つのステップで明らかになります。
返されるのは文字列です。 json.loads()そして、今日は JSON を返し、明日は Markdown を返すことを防ぐために、多くの詳細を記述する必要があります。
自分のデータをチェックしてもらいたい場合は、「xxx という関数があります。パラメーターは yyy で、次のように呼び出すことができます。」という説明のプロンプトを作成する必要があります。プロンプトは、作成するにつれて長くなります。
単体テストを作成しようとすると、コードがグローバル変数とハードコーディングされたパスでいっぱいで、テストが不可能であることがわかります。
問題は、モデルが十分にスマートではないということではなく、途中でエンジニアリング上の制約が欠如していることです。
そしてPydantic AI このレベルの作業を行うことができます。これにより、使い慣れた Python 記述メソッド、型アノテーション、Pydantic モデル、デコレータ、およびデータ クラスが大規模なモデルに直接接続され、「出力はこの構造でなければなりません」、「この関数は呼び出せます」、「このデータソースが実行されているときに提供されます」プロンプトのリクエストに依存するのではなく、これらのことがコード レベルで制約されるまで待ちます。
この記事では、ゼロから構築することを試みます個人会計分析アシスタント: 「先月いくら使った、そのお金はどこに消えた?」と尋ねると、請求書を読み取り、前月比を計算し、カテゴリーを分割し、異常な消費を見つけ出し、最後に完全な構造を持つ結論オブジェクトを返します。
LLM はエージェントと同等ではありません。 一度インターフェースを調整して質疑応答することを「大きなモデルを使う」といいます。
Agent 余分な次に何をするかを決めることができる: 作成した関数を呼び出し、結果を取得し、次のステップについて考え続け、最後に検証済みのオブジェクトを渡します。少なくとも 1 つは、音声文字列ジェネレーターです。
**ピダンティック AI**: それは Pydantic チームが作った Agent フレーム。
アイデアはたったの一文です:タイプ システムを使用してエージェントを管理する。
出力モデルを定義すると、スキーマが生成され、検証され、モデルが自動的に応答します。関数を登録すると、それがモデルが理解できる呼び出しの説明に変換されます。依存関係の種類を宣言すると、実行時にデータ ソースがツールに送信されます。
やめるべき悪い習慣。 「まずは解析して試してみる」という暫定コードです。不確実性はフレームワークレベルで排除できるので、手書きをやめてください try/except ポケットに行きましょう。
最終的に構築されるものは次のように描画されます。
リンク全体の中で実際に「AI」に属するのは中央のブロックだけです。左が質問、右がExcelとプッシュ、真ん中が請求書を読み取り、カテゴリを計算し、異常を検出しますそれはすべて本物の Python です。
課金の利点は、データを自分で持っていることです。
Alipay と WeChat Pay は両方とも、取引の詳細をエクスポートし、日付、カテゴリ、金額、支払い方法、および備考を含む表に整理できます (date / category / amount / method / note)を使用することができます。
答えるべき質問は次のとおりです。いくら使ったか、先月より多いか少ないか、主にどのカテゴリーにお金がかかっているか、使ってはいけないものはないか。 この一連のプロセスは固定されていることに注意してください。数値の取得、集計、連鎖率の計算、カテゴリの分割、異常の検出などです。何が変わるかというと、「ユーザーが質問したいのはどの期間で、どのカテゴリに興味があるのか」ということです。
修正されたものは Python に書き込まれ、意図の理解はモデルに引き継がれます。これはこの記事全体の中で最も重要です。
焦らずに、まずは「モデルが話せるかどうか」を確認してください。
python -m venv venv
source venv/bin/activate # Windows 用 venv\Scripts\activate
pip install "pydantic-ai" pandas numpy
export OPENAI_API_KEY="sk-xxxx" # Windows 用 setx OPENAI_API_KEY "sk-xxxx"
最初のエージェントには 2 つの必須フィールドがあります。
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
MODEL = OpenAIChatModel(
"deepseek-ai/DeepSeek-V3.2",
provider=OpenAIProvider(
base_url="
api_key=os.getenv("OPENAI_API_KEY"),
),
)
ledger_agent = Agent(
MODEL,
instructions=("你是一名个人财务分析助手。"),
)
result = agent.run_sync("什么是大语言模型?")
print(result.output)
Silicon Flow で DeepSeek モデルを使用しています。instructions** これはシステムレベルの個人設定であり、実行ごとに有効になります**。
run_sync これは同期バージョンであり、FastAPI に接続すると置き換えられます。 await agent.run(...)API はまったく同じです。
ここに行きます result.output これは文字列であるため使用できますが、フォームに直接入力することはできません。
これが最も収益性の高い手順です: **エージェントを指定**output_type。 Pydantic モデルを使用して必要な構造を記述します。フレームワークの残りの部分はすべて包括的です。JSON スキーマの生成、プロンプトとともに送信、検証のために結果を取得しますが、エラーを報告してモデルに再度応答させるだけです。
この記事のアシスタントが最終的に引き渡すものは次のとおりです。
class MetricChange(BaseModel):
name: str = Field(description="指标名,如 支出总额 / 笔数 / 单笔均值")
current: float
previous: float
change_pct: float = Field(description="环比变化百分比,保留一位小数,单位 %")
class SpendingInsight(BaseModel):
period: str = Field(description="分析区间,形如 2026-09-01 ~ 2026-09-30")
headline: str = Field(description="一句话结论:花超了还是省下来了,把数字带上")
top_categories: list[str] = Field(description="花钱最多的前三个类别,按金额降序")
anomalies: list[str] = Field(default_factory=list, description="异常消费,没有就返回空列表",)
suggestions: list[str] = Field(description="下个月能做的具体动作,不要写套话")
confidence: float = Field(description="结论可信度,0 到 1 之间", ge=0.0, le=1.0)
どのフィールドもハングアップしている Field(description=...)これはコメントではありませんが、モデルに送信されたスキーマに表示されますこのフィールドが何を必要とするかをモデルに明確に伝えます。明確に記述することで、リトライ回数を直接減らすことができます。
このモデルは実はコピー品ですインターフェースコントラクト: 決定後は、Excel で作成しても、グループにプッシュしても、常に同じフィールドのセットが得られます。
構造化された出力があっても、モデルは先月いくら使ったかをまだ知りません。ツールは単に Python 関数を提供するだけです。
まず、呼び出しの完全なサイクルを見てみましょう。
2 つの重要なポイント:ステップ 3 とツールの間で数回の往復が発生する可能性があります。モデルは段階的に情報を入力します。そしてすべてのツール呼び出しはモデルリクエストですツールが壊れれば壊れるほど、お金と時間が早く費やされます。したがって、ツールの設計原則は次のとおりです。十分な情報を一度に提供する。
通常の関数がモデルによってどのように理解されるか:
デコレータ間の違いは 1 つだけです。@agent.tool_plain 実行時コンテキストを必要としない関数の場合。@agent.tool 最初のパラメータは RunContext[...]注入されたものを拾うために使用されます。
@ledger_agent.tool
def get_metrics(ctx: RunContext[LedgerRepo], start: str, end: str,
categories: list[str] | None = None) -> str:
"""查询一段日期内的支出总额、笔数、单笔均值,
并自动与上一个等长周期做环比。
Args:
start: 开始日期,YYYY-MM-DD。
categories: 消费类别列表,不传表示全部。
"""
return json.dumps(ctx.deps.query_metrics(start, end, categories),
ensure_ascii=False)
その段落 docstring はコメントではなく、モデルの呼び出し命令です。漠然と書いてしまうと、タイミングが合わなくなってしまいます。
もう一つの実践的な経験:すべての計算はツールに入れられます。チェーン比率、割合、平均値は Python で計算して返す必要があります。モデルは「数字を理解し、人間の言葉を話す」ことだけを担当します。
この時点でコードは実行できますが、請求書データ ソースはハードコーディングされています。ファイルを変更したり、テストを作成したりするたびに、ツールのコードを変更する必要があります。変更後は、間違いがないことを祈る必要があります。依存関係の注入は、この問題を解決します。エージェントは、必要なタイプと必要なインスタンスを宣言します。これについては実行時に議論されます。
すべてのデータをデータ クラスに収集します。データ クラスは、ユーザーとデータ ソース間のインターフェイスです。
@dataclass
class LedgerRepo:
frame: pd.DataFrame # 换成数据库、换成 API 都一样
def available_dimensions(self) -> dict: ...
def query_metrics(self, start, end, categories=None) -> dict: ...
def breakdown_by_category(self, start, end) -> dict: ...
def top_expenses(self, start, end, limit=5) -> dict: ...
代理人に関する声明 deps_type=LedgerRepo実行時にインスタンスを挿入するだけです。
repo = LedgerRepo(frame=pd.read_csv("bill.csv", parse_dates=["date"]))
result = ledger_agent.run_sync("上个月我花超了吗?", deps=repo)
insight = result.output # 一个 SpendingInsight 实例
print(insight.headline, insight.top_categories)
報酬はテストを作成するとすぐに実現します –エージェントの定義を変更する必要はありません依存関係を直接モックに置き換えます。
with ledger_agent.override(deps=build_mock_repo()):
result = ledger_agent.run_sync("看一下最近花了多少")
モデルコールさえ保存したい場合は、公式があります TestModel / FunctionModelモデルをオーバーライドしてオフラインで実行することもできます。
フレームワークにより、再利用可能な動作が構成可能になります capabilities電話をかけるだけで、サプライヤーを選ぶ必要はありません。
from pydantic_ai.capabilities import Thinking, WebSearch
agent = Agent(
"deepseek-ai/DeepSeek-V3.2",
instructions="你是一名个人财务助手,结论必须有数字支撑。",
capabilities=[Thinking(effort="high"), WebSearch()],
)
Thinking 「なぜ消費が増えたのか」の帰属に適した答えを得る前に、段階的に進めてみましょう。WebSearch 特定の月に追加の花が祝日と一致するかどうかなど、外部の視点を加えます。
これをオンにして、算術演算が Python に引き継がれたというシナリオを確認しますか。毎日オフにすることでコストを節約でき、インターネット上の無関係なコンテンツによる偏りを防ぐことができます。
完全なコード:ledger_agent_demo.py: (アクセスパスワード: 6872)
興味があれば、次のコードで試してみてください。
- まず独自の API キーをインポートします。
export OPENAI_API_KEY="sk-xxxx" - コード内のモデル部分を独自のモデルに置き換えてください (コード内のテストにはシリコンベースのフロー モデルを使用しました)。
MODEL = OpenAIChatModel(
"deepseek-ai/DeepSeek-V3.2",
provider=OpenAIProvider(
base_url="
api_key=os.getenv("OPENAI_API_KEY"),
),
)
- コードはデフォルトでシミュレートされたデータを使用します。独自のデータがある場合は、環境変数にデータ (CSV 形式) のパスを設定してください。
csv_path = os.getenv("LEDGER_CSV")
実行結果(デフォルトモード) python ledger_agent_demo.py):
$ python ledger_agent_demo.py
pydantic-ai v2.54.0 • Python 3.12.12
/ \ agent: ledger_agent • model: openai:deepseek-ai/DeepSeek-V3.2 • output: SpendingInsight • tools: 4
/ \ capabilities: 0
/___.___\
/ | \ observability: off — see every model and tool call live, with cost
/ | \ set it up free with Logfire and a GitHub login:
`---.._|_..---' or use any OpenTelemetry backend:
goes away once observability is on — or PYDANTIC_AI_NO_BANNER=1
== 上个月花超了,总支出 16,864.79 元,比上个月多花 67.1%
-- 区间: 2026-09-01 ~ 2026-09-30
支出总额: 16864.79(环比 67.1%)
笔数: 243.0(环比 37.3%)
单笔均值: 69.4(环比 21.7%)
-- 花钱最多的类别: 购物、餐饮、娱乐
-- 异常消费:
- 2026-09-01 居住类支出 2,439.86 元
-- 原因分析:
- 购物支出 4,825.27 元占总支出 28.61%,比上个月明显增加
- 9月初有大额居住支出 2,439.86 元一次性拉高整体消费
- 餐饮消费 3,870.85 元笔数达 90 笔,单日消费频次较高
-- 下个月怎么做:
- 下个月重点关注购物类消费,把单笔购物控制在 200 元以内
- 餐饮支出过高,尝试工作日午餐自带或降低外卖频次
- 大额居住消费已发生,下个月房租支出会恢复正常
-- 可信度: 0.95
独自のデータを使用する場合は、最初に独自のトランザクション詳細をエクスポートし、リボンに整理してください。 date / category / amount / method / note 5 列の CSV (ファイル名など) bill.csv);
それから LEDGER_CSV=./bill.csv python ledger_agent_demo.py。