anthropology¶
introduce¶
Logfire supports an extra line of code to detect calls to Anthropic.
import anthropic
import logfire
client = anthropic.Anthropic()
logfire.configure()
logfire.instrument_anthropic(client) # (1)!
response = client.messages.create(
max_tokens=1000,
model='claude-3-haiku-20240307',
system='You are a helpful assistant.',
messages=[{'role': 'user', 'content': 'Please write me a limerick about Python logging.'}],
)
print(response.content[0].text)
-
If you do not have access to the client instance, you can detect the
两者。人类学and人类学。AsyncAnthropicclasses by passing a class (for example,logfire.instrument_anthropic(anthropic.Anthropic)) or simply passing no parameters (i.e.,logfire.instrument_anthropic()).
For more information, see instrument_anthropic() API Reference.
With it, you can get:
-
Span around calls to Anthropic that record duration and catch any exceptions that may occur
-
Human-readable display of conversations with agents
-
Details of the response, including the number of tokens used
the methods encompassed¶
The following anthropological methods are covered:
client.messages.createclient.messages.stream(客户端.messages.stream)client.beta.tools.messages.create
All methods cover 人为的方法。人类学 and 人类学。异步人。
streaming response¶
When detecting streaming responses, Logfire creates two spans-one around the initial request and one around the streaming response.
Here we also use Rich's Live and Markdown types to render responses in real time in the terminal.: Dancer:
import anthropic
import logfire
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown
client = anthropic.AsyncAnthropic()
logfire.configure()
logfire.instrument_anthropic(client)
async def main():
console = Console()
with logfire.span('Asking Anthropic to write some code'):
response = client.messages.stream(
max_tokens=1000,
model='claude-3-haiku-20240307',
system='Reply in markdown one.',
messages=[{'role': 'user', 'content': 'Write Python to show a tree of files 🤞.'}],
)
content = ''
with Live('', refresh_per_second=15, console=console) as live:
async with response as stream:
async for chunk in stream:
if chunk.type == 'content_block_delta':
content += chunk.delta.text
live.update(Markdown(content))
if __name__ == '__main__':
import asyncio
asyncio.run(main())
In Logfire it looks like this:
本文总阅读量次