Pydantic AI
可观测性
使用 Langfuse 实现可观测性
Langfuse 介绍
Pydantic 还有一款可观测产品 Pydantic Logfire，不过这是个商业平台。如果考虑需要纯开源的解决方案，可以考虑 Langfuse。PydanticAI 也支持 OpenTelemetry，所以也可以与任何支持 OpenTelemetry 的可观测工具集成。
安装 Langfuse
下载 Langfuse 的 Docker Compose 文件，然后执行 docker compose up，一会儿就会在本地 3000 端口启动。
访问 http://localhost:3000，可以注册一个用户。
然后新建一个组织（Organization），然后再创建一个项目（Project），然后创建这个项目的 API Keys。包括 Public Key 和 Secret Key。
修改程序
应用需要添加 Langfuse 的依赖，执行：
uv add langfuse代码修改为：
from pydantic_ai import Agent, RunContext
import os
import httpx
from pydantic_ai.models.gemini import GeminiModel # Update import to use GeminiModel
from pydantic_ai.providers.google_gla import GoogleGLAProvider
from pydantic import BaseModel
from dataclasses import dataclass
import readability
from markdownify import markdownify as md
from langfuse import get_client, Langfuse
langfuse = Langfuse(
debug=True
)
# Verify connection
if langfuse.auth_check():
print("Langfuse client is authenticated and ready!")
else:
print("Authentication failed. Please check your credentials and host.")
Agent.instrument_all()
proxy = os.getenv('HTTP_PROXY')
client = httpx.AsyncClient(proxy=proxy, timeout=120)
model = GeminiModel(
model_name='gemini-2.0-flash',
provider=GoogleGLAProvider(
http_client=client
)
)
@dataclass
class MyDeps:
url: str
class ArticleMeta(BaseModel):
url: str
summary: str
meta_agent = Agent(
model=model,
retries=3,
deps_type=MyDeps,
output_type=ArticleMeta,
instructions="""Analyze the content and provide structured response with Chinese.""",
instrument=True
)
translator_agent = Agent(
model=model,
retries=3,
instructions="""As a cloud native expert and translator.
Fetch and Translate this article from English to Chinese.
Keep the markdown format intact.
TRANSLATION REQUIREMENTS:
- Do not translate names of non-famous people.
- Do not translate the text in the code block.
- Do not print explanation, just print the translation.
- Ensure the text of link will be translated.
- Translate 'obserablity' into '可观测性'.
- Make sure translate the text into Simplified Chinese.""",
instrument=True
)
@translator_agent.tool_plain
async def fetch_and_convert_to_markdown(url: str) -> str:
"""
Fetches and converts the content of the given URL to markdown format.
Args:
url (str): The URL to fetch content from.
Returns:
str: The content converted to markdown format.
"""
response = await client.get(url) # Fetch the URL content
response.raise_for_status()
original_html = response.text
doc = readability.Document(original_html)
main_html = doc.summary()
markdown = md(main_html, strip=['img', 'script', 'style']) # Convert HTML to markdown
return markdown
async def main():
url = 'https://thenewstack.io/boost-performance-with-react-server-components-and-next-js/'
# print(content)
result = await translator_agent.run(url)
print(result.output)
meta = await meta_agent.run(result.output, deps=MyDeps(url=url))
print(meta)
langfuse.flush()
if __name__ == "__main__":
import asyncio
asyncio.run(main())验证效果
增加以下环境变量：
LANGFUSE_PUBLIC_KEY=pk-lf-2045a344-665e-4232-937e-61f6d1fe4278
LANGFUSE_SECRET_KEY=sk-lf-3c21313d-6553-4d7c-abb5-db1a879c92c2
LANGFUSE_HOST=http://localhost:3000然后执行 uv run trans_with_langfuse.py，因为设置了 debug=True，所以会打印详细信息，执行完之后回到 http://localhost:3000/ 对应项目的 Trace 已经有信息了：
可以看到 Agent 的调用信息都已经记录了下来，原来 translator_agent 并没有执行 tool，而是将 url 传递给了 Gemini，Gemini 获取了网页并转化成了中文。
Langfuse 的更多用法可以参考 Langfuse 的文档。