天气智能体
Pydantic AI 的一个示例，其中包含多个工具，LLM 需要依次调用这些工具来回答问题。
演示内容：
在这个例子中，我们的想法是一个“天气”代理——用户可以询问多个地点的天气，代理将使用 get_lat_lng 工具获取这些地点的纬度和经度，然后使用 get_weather 工具获取这些地点的天气。
运行示例
要正确运行此示例，您可能需要添加两个额外的 API 密钥 （注意：如果缺少任一密钥，代码将回退使用虚拟数据，因此它们不是必需的）
- 来自 tomorrow.io 的天气 API 密钥，通过
WEATHER_API_KEY设置 - 来自 geocode.maps.co 的地理编码 API 密钥，通过
GEO_API_KEY设置
在安装依赖并设置环境变量后，运行
python -m pydantic_ai_examples.weather_agent
uv run -m pydantic_ai_examples.weather_agent
示例代码
weather_agent.py
"""Example of Pydantic AI with multiple tools which the LLM needs to call in turn to answer a question.
In this case the idea is a "weather" agent — the user can ask for the weather in multiple cities,
the agent will use the `get_lat_lng` tool to get the latitude and longitude of the locations, then use
the `get_weather` tool to get the weather.
Run with:
uv run -m pydantic_ai_examples.weather_agent
"""
from __future__ import annotations as _annotations
import asyncio
from dataclasses import dataclass
from typing import Any
import logfire
from httpx import AsyncClient
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
# 'if-token-present' means nothing will be sent (and the example will work) if you don't have logfire configured
logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()
@dataclass
class Deps:
client: AsyncClient
weather_agent = Agent(
'openai:gpt-4.1-mini',
# 'Be concise, reply with one sentence.' is enough for some models (like openai) to use
# the below tools appropriately, but others like anthropic and gemini require a bit more direction.
instructions='Be concise, reply with one sentence.',
deps_type=Deps,
retries=2,
)
class LatLng(BaseModel):
lat: float
lng: float
@weather_agent.tool
async def get_lat_lng(ctx: RunContext[Deps], location_description: str) -> LatLng:
"""Get the latitude and longitude of a location.
Args:
ctx: The context.
location_description: A description of a location.
"""
# NOTE: the response here will be random, and is not related to the location description.
r = await ctx.deps.client.get(
'https://demo-endpoints.pydantic.workers.dev/latlng',
params={'location': location_description},
)
r.raise_for_status()
return LatLng.model_validate_json(r.content)
@weather_agent.tool
async def get_weather(ctx: RunContext[Deps], lat: float, lng: float) -> dict[str, Any]:
"""Get the weather at a location.
Args:
ctx: The context.
lat: Latitude of the location.
lng: Longitude of the location.
"""
# NOTE: the responses here will be random, and are not related to the lat and lng.
temp_response, descr_response = await asyncio.gather(
ctx.deps.client.get(
'https://demo-endpoints.pydantic.workers.dev/number',
params={'min': 10, 'max': 30},
),
ctx.deps.client.get(
'https://demo-endpoints.pydantic.workers.dev/weather',
params={'lat': lat, 'lng': lng},
),
)
temp_response.raise_for_status()
descr_response.raise_for_status()
return {
'temperature': f'{temp_response.text} °C',
'description': descr_response.text,
}
async def main():
async with AsyncClient() as client:
logfire.instrument_httpx(client, capture_all=True)
deps = Deps(client=client)
result = await weather_agent.run(
'What is the weather like in London and in Wiltshire?', deps=deps
)
print('Response:', result.output)
if __name__ == '__main__':
asyncio.run(main())
运行 UI
您可以使用 Gradio 为您的代理构建多轮聊天应用程序，Gradio 是一个完全用 Python 构建 AI Web 应用程序的框架。Gradio 带有内置的聊天组件和代理支持，因此整个 UI 将在一个 Python 文件中实现！
以下是天气代理的 UI 界面
pip install gradio>=5.9.0
python/uv-run -m pydantic_ai_examples.weather_agent_gradio
UI 代码
weather_agent_gradio.py
from __future__ import annotations as _annotations
import json
from httpx import AsyncClient
from pydantic import BaseModel
from pydantic_ai.messages import ToolCallPart, ToolReturnPart
from pydantic_ai_examples.weather_agent import Deps, weather_agent
try:
import gradio as gr
except ImportError as e:
raise ImportError(
'Please install gradio with `pip install gradio`. You must use python>=3.10.'
) from e
TOOL_TO_DISPLAY_NAME = {'get_lat_lng': 'Geocoding API', 'get_weather': 'Weather API'}
client = AsyncClient()
deps = Deps(client=client)
async def stream_from_agent(prompt: str, chatbot: list[dict], past_messages: list):
chatbot.append({'role': 'user', 'content': prompt})
yield gr.Textbox(interactive=False, value=''), chatbot, gr.skip()
async with weather_agent.run_stream(
prompt, deps=deps, message_history=past_messages
) as result:
for message in result.new_messages():
for call in message.parts:
if isinstance(call, ToolCallPart):
call_args = call.args_as_json_str()
metadata = {
'title': f'🛠️ Using {TOOL_TO_DISPLAY_NAME[call.tool_name]}',
}
if call.tool_call_id is not None:
metadata['id'] = call.tool_call_id
gr_message = {
'role': 'assistant',
'content': 'Parameters: ' + call_args,
'metadata': metadata,
}
chatbot.append(gr_message)
if isinstance(call, ToolReturnPart):
for gr_message in chatbot:
if (
gr_message.get('metadata', {}).get('id', '')
== call.tool_call_id
):
if isinstance(call.content, BaseModel):
json_content = call.content.model_dump_json()
else:
json_content = json.dumps(call.content)
gr_message['content'] += f'\nOutput: {json_content}'
yield gr.skip(), chatbot, gr.skip()
chatbot.append({'role': 'assistant', 'content': ''})
async for message in result.stream_text():
chatbot[-1]['content'] = message
yield gr.skip(), chatbot, gr.skip()
past_messages = result.all_messages()
yield gr.Textbox(interactive=True), gr.skip(), past_messages
async def handle_retry(chatbot, past_messages: list, retry_data: gr.RetryData):
new_history = chatbot[: retry_data.index]
previous_prompt = chatbot[retry_data.index]['content']
past_messages = past_messages[: retry_data.index]
async for update in stream_from_agent(previous_prompt, new_history, past_messages):
yield update
def undo(chatbot, past_messages: list, undo_data: gr.UndoData):
new_history = chatbot[: undo_data.index]
past_messages = past_messages[: undo_data.index]
return chatbot[undo_data.index]['content'], new_history, past_messages
def select_data(message: gr.SelectData) -> str:
return message.value['text']
with gr.Blocks() as demo:
gr.HTML(
"""
<div style="display: flex; justify-content: center; align-items: center; gap: 2rem; padding: 1rem; width: 100%">
<img src="https://ai.pydantic.org.cn/img/logo-white.svg" style="max-width: 200px; height: auto">
<div>
<h1 style="margin: 0 0 1rem 0">Weather Assistant</h1>
<h3 style="margin: 0 0 0.5rem 0">
This assistant answer your weather questions.
</h3>
</div>
</div>
"""
)
past_messages = gr.State([])
chatbot = gr.Chatbot(
label='Packing Assistant',
type='messages',
avatar_images=(None, 'https://ai.pydantic.org.cn/img/logo-white.svg'),
examples=[
{'text': 'What is the weather like in Miami?'},
{'text': 'What is the weather like in London?'},
],
)
with gr.Row():
prompt = gr.Textbox(
lines=1,
show_label=False,
placeholder='What is the weather like in New York City?',
)
generation = prompt.submit(
stream_from_agent,
inputs=[prompt, chatbot, past_messages],
outputs=[prompt, chatbot, past_messages],
)
chatbot.example_select(select_data, None, [prompt])
chatbot.retry(
handle_retry, [chatbot, past_messages], [prompt, chatbot, past_messages]
)
chatbot.undo(undo, [chatbot, past_messages], [prompt, chatbot, past_messages])
if __name__ == '__main__':
demo.launch()