import os
# os.environ['ANTHROPIC_LOG'] = 'debug'
Claudette’s source
This is the ‘literate’ source code for Claudette. You can view the fully rendered version of the notebook here, or you can clone the git repo and run the interactive notebook in Jupyter. The notebook is converted the Python module claudette/core.py using nbdev. The goal of this source code is to both create the Python module, and also to teach the reader how it is created, without assuming much existing knowledge about Claude’s API.
Most of the time you’ll see that we write some source code first, and then a description or discussion of it afterwards.
Setup
To print every HTTP request and response in full, uncomment the above line. This functionality is provided by Anthropic’s SDK.
If you’re reading the rendered version of this notebook, you’ll see an “Exported source” collapsible widget below. If you’re reading the source notebook directly, you’ll see #| exports
at the top of the cell. These show that this piece of code will be exported into the python module that this notebook creates. No other code will be included – any other code in this notebook is just for demonstration, documentation, and testing.
You can toggle expanding/collapsing the source code of all exported sections by using the </> Code
menu in the top right of the rendered notebook page.
Exported source
= {
model_types # Anthropic
'claude-3-opus-20240229': 'opus',
'claude-3-5-sonnet-20240620': 'sonnet',
'claude-3-haiku-20240307': 'haiku',
# AWS
'anthropic.claude-3-opus-20240229-v1:0': 'opus',
'anthropic.claude-3-5-sonnet-20240620-v1:0': 'sonnet',
'anthropic.claude-3-sonnet-20240229-v1:0': 'sonnet',
'anthropic.claude-3-haiku-20240307-v1:0': 'haiku',
# Google
'claude-3-opus@20240229': 'opus',
'claude-3-5-sonnet@20240620': 'sonnet',
'claude-3-sonnet@20240229': 'sonnet',
'claude-3-haiku@20240307': 'haiku',
}
= list(model_types) all_models
These are the current versions and prices of Anthropic’s models at the time of writing.
= models[1] model
For examples, we’ll use Sonnet 3.5, since it’s awesome.
Antropic SDK
= Anthropic() cli
This is what Anthropic’s SDK provides for interacting with Python. To use it, pass it a list of messages, with content and a role. The roles should alternate between user and assistant.
After the code below you’ll see an indented section with an orange vertical line on the left. This is used to show the result of running the code above. Because the code is running in a Jupyter Notebook, we don’t have to use print
to display results, we can just type the expression directly, as we do with r
here.
= {'role': 'user', 'content': "I'm Jeremy"}
m = cli.messages.create(messages=[m], model=model, max_tokens=100)
r r
Hello Jeremy! It’s nice to meet you. How can I assist you today? Feel free to ask me any questions or let me know if you need help with anything.
- id:
msg_01JGTX1KNKpS3W7KogJVmSRX
- content:
[{'text': "Hello Jeremy! It's nice to meet you. How can I assist you today? Feel free to ask me any questions or let me know if you need help with anything.", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 10, 'output_tokens': 38}
Formatting output
That output is pretty long and hard to read, so let’s clean it up. We’ll start by pulling out the Content
part of the message. To do that, we’re going to write our first function which will be included to the claudette/core.py
module.
This is the first exported public function or class we’re creating (the previous export was of a variable). In the rendered version of the notebook for these you’ll see 4 things, in this order (unless the symbol starts with a single _
, which indicates it’s private):
- The signature (with the symbol name as a heading, with a horizontal rule above)
- A table of paramater docs (if provided)
- The doc string (in italics).
- The source code (in a collapsible “Exported source” block)
After that, we generally provide a bit more detail on what we’ve created, and why, along with a sample usage.
find_block
find_block (r:collections.abc.Mapping, blk_type:type=<class 'anthropic.types.text_block.TextBlock'>)
Find the first block of type blk_type
in r.content
.
Type | Default | Details | |
---|---|---|---|
r | Mapping | The message to look in | |
blk_type | type | TextBlock | The type of block to find |
Exported source
def find_block(r:abc.Mapping, # The message to look in
type=TextBlock # The type of block to find
blk_type:
):"Find the first block of type `blk_type` in `r.content`."
return first(o for o in r.content if isinstance(o,blk_type))
This makes it easier to grab the needed parts of Claude’s responses, which can include multiple pieces of content. By default, we look for the first text block. That will generally have the content we want to display.
find_block(r)
TextBlock(text="Hello Jeremy! It's nice to meet you. How can I assist you today? Feel free to ask me any questions or let me know if you need help with anything.", type='text')
contents
contents (r)
Helper to get the contents from Claude response r
.
Exported source
def contents(r):
"Helper to get the contents from Claude response `r`."
= find_block(r)
blk if not blk and r.content: blk = r.content[0]
return blk.text.strip() if hasattr(blk,'text') else str(blk)
For display purposes, we often just want to show the text itself.
contents(r)
"Hello Jeremy! It's nice to meet you. How can I assist you today? Feel free to ask me any questions or let me know if you need help with anything."
Exported source
@patch
def _repr_markdown_(self:(Message)):
= '\n- '.join(f'{k}: `{v}`' for k,v in self.model_dump().items())
det = re.sub(r'\$', '$', contents(self)) # escape `$` for jupyter latex
cts return f"""{cts}
<details>
- {det}
</details>"""
Jupyter looks for a _repr_markdown_
method in displayed objects; we add this in order to display just the content text, and collapse full details into a hideable section. Note that patch
is from fastcore, and is used to add (or replace) functionality in an existing class. We pass the class(es) that we want to patch as type annotations to self
. In this case, _repr_markdown_
is being added to Anthropic’s Message
class, so when we display the message now we just see the contents, and the details are hidden away in a collapsible details block.
r
Hello Jeremy! It’s nice to meet you. How can I assist you today? Feel free to ask me any questions or let me know if you need help with anything.
- id:
msg_01JGTX1KNKpS3W7KogJVmSRX
- content:
[{'text': "Hello Jeremy! It's nice to meet you. How can I assist you today? Feel free to ask me any questions or let me know if you need help with anything.", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 10, 'output_tokens': 38}
One key part of the response is the usage
key, which tells us how many tokens we used by returning a Usage
object.
We’ll add some helpers to make things a bit cleaner for creating and formatting these objects.
r.usage
In: 10; Out: 38; Cache create: 0; Cache read: 0; Total: 48
usage
usage (inp=0, out=0, cache_create=0, cache_read=0)
Slightly more concise version of Usage
.
Type | Default | Details | |
---|---|---|---|
inp | int | 0 | input tokens |
out | int | 0 | Output tokens |
cache_create | int | 0 | Cache creation tokens |
cache_read | int | 0 | Cache read tokens |
Exported source
def usage(inp=0, # input tokens
=0, # Output tokens
out=0, # Cache creation tokens
cache_create=0 # Cache read tokens
cache_read
):"Slightly more concise version of `Usage`."
return Usage(input_tokens=inp, output_tokens=out, cache_creation_input_tokens=cache_create, cache_read_input_tokens=cache_read)
The constructor provided by Anthropic is rather verbose, so we clean it up a bit, using a lowercase version of the name.
5) usage(
In: 5; Out: 0; Cache create: 0; Cache read: 0; Total: 5
Usage.total
Usage.total ()
Exported source
@patch(as_prop=True)
def total(self:Usage): return self.input_tokens+self.output_tokens+getattr(self, "cache_creation_input_tokens",0)+getattr(self, "cache_read_input_tokens",0)
Adding a total
property to Usage
makes it easier to see how many tokens we’ve used up altogether.
5,1).total usage(
6
Usage.__repr__
Usage.__repr__ ()
Return repr(self).
Exported source
@patch
def __repr__(self:Usage): return f'In: {self.input_tokens}; Out: {self.output_tokens}; Cache create: {getattr(self, "cache_creation_input_tokens",0)}; Cache read: {getattr(self, "cache_read_input_tokens",0)}; Total: {self.total}'
In python, patching __repr__
lets us change how an object is displayed. (More generally, methods starting and ending in __
in Python are called dunder
methods, and have some magic
behavior – such as, in this case, changing how an object is displayed.)
5) usage(
In: 5; Out: 0; Cache create: 0; Cache read: 0; Total: 5
Usage.__add__
Usage.__add__ (b)
Add together each of input_tokens
and output_tokens
Exported source
@patch
def __add__(self:Usage, b):
"Add together each of `input_tokens` and `output_tokens`"
return usage(self.input_tokens+b.input_tokens, self.output_tokens+b.output_tokens, getattr(self,'cache_creation_input_tokens',0)+getattr(b,'cache_creation_input_tokens',0), getattr(self,'cache_read_input_tokens',0)+getattr(b,'cache_read_input_tokens',0))
And, patching __add__
lets +
work on a Usage
object.
+r.usage r.usage
In: 20; Out: 76; Cache create: 0; Cache read: 0; Total: 96
Creating messages
Creating correctly formatted dict
s from scratch every time isn’t very handy, so next up we’ll add helpers for this.
def mk_msg(content, role='user', **kw):
return dict(role=role, content=content, **kw)
We make things a bit more convenient by writing a function to create a message for us.
You may have noticed that we didn’t export the mk_msg
function (i.e. there’s no “Exported source” block around it). That’s because we’ll need more functionality in our final version than this version has – so we’ll be defining a more complete version later. Rather than refactoring/editing in notebooks, often it’s helpful to simply gradually build up complexity by re-defining a symbol.
= "I'm Jeremy"
prompt = mk_msg(prompt)
m m
{'role': 'user', 'content': "I'm Jeremy"}
= cli.messages.create(messages=[m], model=model, max_tokens=100)
r r
Hello Jeremy! It’s nice to meet you. How are you doing today? Is there anything I can help you with or any questions you’d like to ask?
- id:
msg_01FPMrkmWuYfacN8xDnBXKUY
- content:
[{'text': "Hello Jeremy! It's nice to meet you. How are you doing today? Is there anything I can help you with or any questions you'd like to ask?", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 10, 'output_tokens': 36}
mk_msgs
mk_msgs (msgs:list, **kw)
Helper to set ‘assistant’ role on alternate messages.
Exported source
def mk_msgs(msgs:list, **kw):
"Helper to set 'assistant' role on alternate messages."
if isinstance(msgs,str): msgs=[msgs]
return [mk_msg(o, ('user','assistant')[i%2], **kw) for i,o in enumerate(msgs)]
LLMs, including Claude, don’t actually have state, but instead dialogs are created by passing back all previous prompts and responses every time. With Claude, they always alternate user and assistant. Therefore we create a function to make it easier to build up these dialog lists.
But to do so, we need to update mk_msg
so that we can’t only pass a str
as content
, but can also pass a dict
or an object with a content
attr, since these are both types of message that Claude can create. To do so, we check for a content
key or attr, and use it if found.
Exported source
def _str_if_needed(o):
if isinstance(o, (list,tuple,abc.Mapping,L)) or hasattr(o, '__pydantic_serializer__'): return o
return str(o)
def mk_msg(content, role='user', **kw):
"Helper to create a `dict` appropriate for a Claude message. `kw` are added as key/value pairs to the message"
if hasattr(content, 'content'): content,role = content.content,content.role
if isinstance(content, abc.Mapping): content=content['content']
return dict(role=role, content=_str_if_needed(content), **kw)
= mk_msgs([prompt, r, 'I forgot my name. Can you remind me please?'])
msgs msgs
[{'role': 'user', 'content': "I'm Jeremy"},
{'role': 'assistant',
'content': [TextBlock(text="Hello Jeremy! It's nice to meet you. How are you doing today? Is there anything I can help you with or any questions you'd like to ask?", type='text')]},
{'role': 'user', 'content': 'I forgot my name. Can you remind me please?'}]
Now, if we pass this list of messages to Claude, the model treats it as a conversation to respond to.
=msgs, model=model, max_tokens=200) cli.messages.create(messages
Of course! You just told me that your name is Jeremy.
- id:
msg_014ntKw7Mvfws4BHjjsyYLPn
- content:
[{'text': 'Of course! You just told me that your name is Jeremy.', 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 60, 'output_tokens': 16}
Client
Client
Client (model, cli=None, log=False)
Basic Anthropic messages client.
Exported source
class Client:
def __init__(self, model, cli=None, log=False):
"Basic Anthropic messages client."
self.model,self.use = model,usage()
self.log = [] if log else None
self.c = (cli or Anthropic(default_headers={'anthropic-beta': 'prompt-caching-2024-07-31'}))
We’ll create a simple Client
for Anthropic
which tracks usage stores the model to use. We don’t add any methods right away – instead we’ll use patch
for that so we can add and document them incrementally.
= Client(model)
c c.use
In: 0; Out: 0; Cache create: 0; Cache read: 0; Total: 0
Exported source
@patch
def _r(self:Client, r:Message, prefill=''):
"Store the result of the message and accrue total usage."
if prefill:
= find_block(r)
blk = prefill + (blk.text or '')
blk.text self.result = r
self.use += r.usage
self.stop_reason = r.stop_reason
self.stop_sequence = r.stop_sequence
return r
We use a _
prefix on private methods, but we document them here in the interests of literate source code.
_r
will be used each time we get a new result, to track usage and also to keep the result available for later.
c._r(r) c.use
In: 10; Out: 36; Cache create: 0; Cache read: 0; Total: 46
Whereas OpenAI’s models use a stream
parameter for streaming, Anthropic’s use a separate method. We implement Anthropic’s approach in a private method, and then use a stream
parameter in __call__
for consistency:
Exported source
@patch
def _log(self:Client, final, prefill, msgs, maxtok=None, sp=None, temp=None, stream=None, stop=None, **kwargs):
self._r(final, prefill)
if self.log is not None: self.log.append({
"msgs": msgs, "prefill": prefill, **kwargs,
"msgs": msgs, "prefill": prefill, "maxtok": maxtok, "sp": sp, "temp": temp, "stream": stream, "stop": stop, **kwargs,
"result": self.result, "use": self.use, "stop_reason": self.stop_reason, "stop_sequence": self.stop_sequence
})return self.result
Exported source
@patch
def _stream(self:Client, msgs:list, prefill='', **kwargs):
with self.c.messages.stream(model=self.model, messages=mk_msgs(msgs), **kwargs) as s:
if prefill: yield(prefill)
yield from s.text_stream
self._log(s.get_final_message(), prefill, msgs, **kwargs)
Claude supports adding an extra assistant
message at the end, which contains the prefill – i.e. the text we want Claude to assume the response starts with. However Claude doesn’t actually repeat that in the response, so for convenience we add it.
Exported source
@patch
def _precall(self:Client, msgs, prefill, stop, kwargs):
= [prefill.strip()] if prefill else []
pref if not isinstance(msgs,list): msgs = [msgs]
if stop is not None:
if not isinstance(stop, (list)): stop = [stop]
"stop_sequences"] = stop
kwargs[= mk_msgs(msgs+pref)
msgs return msgs
@patch
@delegates(messages.Messages.create)
def __call__(self:Client,
list, # List of messages in the dialog
msgs:='', # The system prompt
sp=0, # Temperature
temp=4096, # Maximum tokens
maxtok='', # Optional prefill to pass to Claude as start of its response
prefillbool=False, # Stream response?
stream:=None, # Stop sequence
stop**kwargs):
"Make a call to Claude."
= self._precall(msgs, prefill, stop, kwargs)
msgs if stream: return self._stream(msgs, prefill=prefill, max_tokens=maxtok, system=sp, temperature=temp, **kwargs)
= self.c.messages.create(
res =self.model, messages=msgs, max_tokens=maxtok, system=sp, temperature=temp, **kwargs)
modelreturn self._log(res, prefill, msgs, maxtok, sp, temp, stream=stream, **kwargs)
Defining __call__
let’s us use an object like a function (i.e it’s callable). We use it as a small wrapper over messages.create
. However we’re not exporting this version just yet – we have some additions we’ll make in a moment…
= Client(model, log=True)
c c.use
In: 0; Out: 0; Cache create: 0; Cache read: 0; Total: 0
= models[-1] c.model
'Hi') c(
Hello! How can I assist you today?
- id:
msg_01GQLdNJ4BT92shxxq9JAiUE
- content:
[{'text': 'Hello! How can I assist you today?', 'type': 'text'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 8, 'output_tokens': 12, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
c.use
In: 8; Out: 12; Cache create: 0; Cache read: 0; Total: 20
Let’s try out prefill:
= "Concisely, what is the meaning of life?"
q = 'According to Douglas Adams,' pref
=pref) c(q, prefill
According to Douglas Adams, “The answer to the ultimate question of life, the universe, and everything is 42.”
- id:
msg_019ULPf3arEfv9yqdEm8oZat
- content:
[{'text': 'According to Douglas Adams, "The answer to the ultimate question of life, the universe, and everything is 42."', 'type': 'text'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 24, 'output_tokens': 23, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
We can pass stream=True
to stream the response back incrementally:
for o in c('Hi', stream=True): print(o, end='')
Hello! How can I assist you today?
c.use
In: 40; Out: 47; Cache create: 0; Cache read: 0; Total: 87
for o in c(q, prefill=pref, stream=True): print(o, end='')
According to Douglas Adams, "The answer to the ultimate question of life, the universe, and everything is 42."
c.use
In: 64; Out: 70; Cache create: 0; Cache read: 0; Total: 134
Pass a stop seauence if you want claude to stop generating text when it encounters it.
"Count from 1 to 10", stop="5") c(
1, 2, 3, 4,
- id:
msg_013PPywqnu3H1zL7eG6M9BsF
- content:
[{'text': '1, 2, 3, 4, ', 'type': 'text'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
stop_sequence
- stop_sequence:
5
- type:
message
- usage:
{'input_tokens': 15, 'output_tokens': 14, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
This also works with streaming, and you can pass more than one stop sequence:
for o in c("Count from 1 to 10", stop=["2", "yellow"], stream=True): print(o, end='')
print(c.stop_reason, c.stop_sequence)
1, stop_sequence 2
You can check the logs:
-1] c.log[
{'msgs': [{'role': 'user', 'content': 'Count from 1 to 10'}],
'prefill': '',
'max_tokens': 4096,
'system': '',
'temperature': 0,
'stop_sequences': ['2', 'yellow'],
'maxtok': None,
'sp': None,
'temp': None,
'stream': None,
'stop': None,
'result': Message(id='msg_01K76r1E9Wh8fM8CkmotxDQs', content=[TextBlock(text='1, ', type='text')], model='claude-3-haiku-20240307', role='assistant', stop_reason='stop_sequence', stop_sequence='2', type='message', usage=In: 15; Out: 5; Cache create: 0; Cache read: 0; Total: 20),
'use': In: 94; Out: 89; Cache create: 0; Cache read: 0; Total: 183,
'stop_reason': 'stop_sequence',
'stop_sequence': '2'}
Tool use
Let’s now add tool use (aka function calling).
mk_tool_choice
mk_tool_choice (choose:Union[str,bool,NoneType])
Create a tool_choice
dict that’s ‘auto’ if choose
is None
, ‘any’ if it is True, or ‘tool’ otherwise
Exported source
def mk_tool_choice(choose:Union[str,bool,None])->dict:
"Create a `tool_choice` dict that's 'auto' if `choose` is `None`, 'any' if it is True, or 'tool' otherwise"
return {"type": "tool", "name": choose} if isinstance(choose,str) else {'type':'any'} if choose else {'type':'auto'}
print(mk_tool_choice('sums'))
print(mk_tool_choice(True))
print(mk_tool_choice(None))
{'type': 'tool', 'name': 'sums'}
{'type': 'any'}
{'type': 'auto'}
Claude can be forced to use a particular tool, or select from a specific list of tools, or decide for itself when to use a tool. If you want to force a tool (or force choosing from a list), include a tool_choice
param with a dict from mk_tool_choice
.
For testing, we need a function that Claude can call; we’ll write a simple function that adds numbers together, and will tell us when it’s being called:
def sums(
int, # First thing to sum
a:int=1 # Second thing to sum
b:-> int: # The sum of the inputs
) "Adds a + b."
print(f"Finding the sum of {a} and {b}")
return a + b
= 604542,6458932
a,b = f"What is {a}+{b}?"
pr = "You are a summing expert." sp
Claudette can autogenerate a schema thanks to the toolslm
library. We’ll force the use of the tool using the function we created earlier.
=[get_schema(sums)]
tools= mk_tool_choice('sums') choice
We’ll start a dialog with Claude now. We’ll store the messages of our dialog in msgs
. The first message will be our prompt pr
, and we’ll pass our tools
schema.
= mk_msgs(pr)
msgs = c(msgs, sp=sp, tools=tools, tool_choice=choice)
r r
ToolUseBlock(id=‘toolu_01KDmzXp4XfPBT5e4pRe551A’, input={‘a’: 604542, ‘b’: 6458932}, name=‘sums’, type=‘tool_use’)
- id:
msg_01VeFngZj3FcjgoMLepXAWqS
- content:
[{'id': 'toolu_01KDmzXp4XfPBT5e4pRe551A', 'input': {'a': 604542, 'b': 6458932}, 'name': 'sums', 'type': 'tool_use'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
tool_use
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 493, 'output_tokens': 53, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
When Claude decides that it should use a tool, it passes back a ToolUseBlock
with the name of the tool to call, and the params to use.
We don’t want to allow it to call just any possible function (that would be a security disaster!) so we create a namespace – that is, a dictionary of allowable function names to call.
Exported source
def _mk_ns(*funcs:list[callable]) -> dict[str,callable]:
"Create a `dict` of name to function in `funcs`, to use as a namespace"
return {f.__name__:f for f in funcs}
= _mk_ns(sums)
ns ns
{'sums': <function __main__.sums(a: int, b: int = 1) -> int>}
mk_funcres
mk_funcres (tuid, res)
Given tool use id and the tool result, create a tool_result response.
Exported source
def call_func(fc:ToolUseBlock, # Tool use block from Claude's message
=None, # Namespace to search for tools, defaults to `globals()`
ns:Optional[abc.Mapping]=None # Object to search for tools
obj:Optional
):"Call the function in the tool response `tr`, using namespace `ns`."
if ns is None: ns=globals()
if not isinstance(ns, abc.Mapping): ns = _mk_ns(*ns)
= getattr(obj, fc.name, None)
func if not func: func = ns[fc.name]
return func(**fc.input)
def mk_funcres(tuid, res):
"Given tool use id and the tool result, create a tool_result response."
return dict(type="tool_result", tool_use_id=tuid, content=str(res))
call_func
call_func (fc:anthropic.types.tool_use_block.ToolUseBlock, ns:Optional[collections.abc.Mapping]=None, obj:Optional=None)
Call the function in the tool response tr
, using namespace ns
.
Type | Default | Details | |
---|---|---|---|
fc | ToolUseBlock | Tool use block from Claude’s message | |
ns | Optional | None | Namespace to search for tools, defaults to globals() |
obj | Optional | None | Object to search for tools |
We can now use the function requested by Claude. We look it up in ns
, and pass in the provided parameters.
= find_block(r, ToolUseBlock)
fc = mk_funcres(fc.id, call_func(fc, ns=ns))
res res
Finding the sum of 604542 and 6458932
{'type': 'tool_result',
'tool_use_id': 'toolu_01KDmzXp4XfPBT5e4pRe551A',
'content': '7063474'}
mk_toolres
mk_toolres (r:collections.abc.Mapping, ns:Optional[collections.abc.Mapping]=None, obj:Optional=None)
Create a tool_result
message from response r
.
Type | Default | Details | |
---|---|---|---|
r | Mapping | Tool use request response from Claude | |
ns | Optional | None | Namespace to search for tools |
obj | Optional | None | Class to search for tools |
Exported source
def mk_toolres(
# Tool use request response from Claude
r:abc.Mapping, =None, # Namespace to search for tools
ns:Optional[abc.Mapping]=None # Class to search for tools
obj:Optional
):"Create a `tool_result` message from response `r`."
= getattr(r, 'content', [])
cts = [mk_msg(r)]
res = [mk_funcres(o.id, call_func(o, ns=ns, obj=obj)) for o in cts if isinstance(o,ToolUseBlock)]
tcs if tcs: res.append(mk_msg(tcs))
return res
In order to tell Claude the result of the tool call, we pass back the tool use assistant request and the tool_result
response.
= mk_toolres(r, ns=ns)
tr tr
Finding the sum of 604542 and 6458932
[{'role': 'assistant',
'content': [ToolUseBlock(id='toolu_01KDmzXp4XfPBT5e4pRe551A', input={'a': 604542, 'b': 6458932}, name='sums', type='tool_use')]},
{'role': 'user',
'content': [{'type': 'tool_result',
'tool_use_id': 'toolu_01KDmzXp4XfPBT5e4pRe551A',
'content': '7063474'}]}]
We add this to our dialog, and now Claude has all the information it needs to answer our question.
+= tr
msgs =sp, tools=tools)) contents(c(msgs, sp
'The sum of 604542 and 6458932 is 7063474.'
This works with methods as well – in this case, use the object itself for ns
:
class Dummy:
def sums(
self,
int, # First thing to sum
a:int=1 # Second thing to sum
b:-> int: # The sum of the inputs
) "Adds a + b."
print(f"Finding the sum of {a} and {b}")
return a + b
= [get_schema(Dummy.sums)]
tools = Dummy()
o = c(pr, sp=sp, tools=tools, tool_choice=choice)
r = mk_toolres(r, obj=o)
tr += tr
msgs =sp, tools=tools)) contents(c(msgs, sp
Finding the sum of 604542 and 6458932
'The sum of 604,542 and 6,458,932 is 7,063,474.'
Client.__call__
Client.__call__ (msgs:list, sp='', temp=0, maxtok=4096, prefill='', stream:bool=False, stop=None, tools:Optional[list]=None, tool_choice:Optional[dict]=None, metadata:MetadataParam|NotGiven=NOT_GIVEN, stop_sequences:List[str]|NotGiven=NOT_GIVEN, system:Unio n[str,Iterable[TextBlockParam]]|NotGiven=NOT_GIVEN, temperature:float|NotGiven=NOT_GIVEN, top_k:int|NotGiven=NOT_GIVEN, top_p:float|NotGiven=NOT_GIVEN, extra_headers:Headers|None=None, extra_query:Query|None=None, extra_body:Body|None=None, timeout:float|httpx.Timeout|None|NotGiven=NOT_GIVEN)
Make a call to Claude.
Type | Default | Details | |
---|---|---|---|
msgs | list | List of messages in the dialog | |
sp | str | The system prompt | |
temp | int | 0 | Temperature |
maxtok | int | 4096 | Maximum tokens |
prefill | str | Optional prefill to pass to Claude as start of its response | |
stream | bool | False | Stream response? |
stop | NoneType | None | Stop sequence |
tools | Optional | None | List of tools to make available to Claude |
tool_choice | Optional | None | Optionally force use of some tool |
metadata | MetadataParam | NotGiven | NOT_GIVEN | |
stop_sequences | List[str] | NotGiven | NOT_GIVEN | |
system | Union[str, Iterable[TextBlockParam]] | NotGiven | NOT_GIVEN | |
temperature | float | NotGiven | NOT_GIVEN | |
top_k | int | NotGiven | NOT_GIVEN | |
top_p | float | NotGiven | NOT_GIVEN | |
extra_headers | Headers | None | None | |
extra_query | Query | None | None | |
extra_body | Body | None | None | |
timeout | float | httpx.Timeout | None | NotGiven | NOT_GIVEN |
Exported source
@patch
@delegates(messages.Messages.create)
def __call__(self:Client,
list, # List of messages in the dialog
msgs:='', # The system prompt
sp=0, # Temperature
temp=4096, # Maximum tokens
maxtok='', # Optional prefill to pass to Claude as start of its response
prefillbool=False, # Stream response?
stream:=None, # Stop sequence
stoplist]=None, # List of tools to make available to Claude
tools:Optional[dict]=None, # Optionally force use of some tool
tool_choice:Optional[**kwargs):
"Make a call to Claude."
if tools: kwargs['tools'] = [get_schema(o) for o in listify(tools)]
if tool_choice: kwargs['tool_choice'] = mk_tool_choice(tool_choice)
= self._precall(msgs, prefill, stop, kwargs)
msgs if stream: return self._stream(msgs, prefill=prefill, max_tokens=maxtok, system=sp, temperature=temp, **kwargs)
= self.c.messages.create(model=self.model, messages=msgs, max_tokens=maxtok, system=sp, temperature=temp, **kwargs)
res return self._log(res, prefill, msgs, maxtok, sp, temp, stream=stream, stop=stop, **kwargs)
= c(pr, sp=sp, tools=sums, tool_choice=sums)
r r
ToolUseBlock(id=‘toolu_01BivK3vyqdKSM9pAWQqWfp7’, input={‘a’: 604542, ‘b’: 6458932}, name=‘sums’, type=‘tool_use’)
- id:
msg_017toeJm3hbSP4iLPYgi6YDs
- content:
[{'id': 'toolu_01BivK3vyqdKSM9pAWQqWfp7', 'input': {'a': 604542, 'b': 6458932}, 'name': 'sums', 'type': 'tool_use'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
tool_use
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 489, 'output_tokens': 57, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
= mk_toolres(r) tr
Finding the sum of 604542 and 6458932
Client.structured
Client.structured (msgs:list, tools:Optional[list]=None, obj:Optional=None, ns:Optional[collections.abc.Mapping]=None, sp='', temp=0, maxtok=4096, prefill='', stream:bool=False, stop=None, tool_choice:Optional[dict]=None, metadata:MetadataParam|NotGiven=NOT_GIVEN, stop_sequences:List[str]|NotGiven=NOT_GIVEN, system:Un ion[str,Iterable[TextBlockParam]]|NotGiven=NOT_GIVEN, temperature:float|NotGiven=NOT_GIVEN, top_k:int|NotGiven=NOT_GIVEN, top_p:float|NotGiven=NOT_GIVEN, extra_headers:Headers|None=None, extra_query:Query|None=None, extra_body:Body|None=None, timeout:float|httpx.Timeout|None|NotGiven=NOT_GIVEN)
Return the value of all tool calls (generally used for structured outputs)
Type | Default | Details | |
---|---|---|---|
msgs | list | List of messages in the dialog | |
tools | Optional | None | List of tools to make available to Claude |
obj | Optional | None | Class to search for tools |
ns | Optional | None | Namespace to search for tools |
sp | str | The system prompt | |
temp | int | 0 | Temperature |
maxtok | int | 4096 | Maximum tokens |
prefill | str | Optional prefill to pass to Claude as start of its response | |
stream | bool | False | Stream response? |
stop | NoneType | None | Stop sequence |
tool_choice | Optional | None | Optionally force use of some tool |
metadata | MetadataParam | NotGiven | NOT_GIVEN | |
stop_sequences | List[str] | NotGiven | NOT_GIVEN | |
system | Union[str, Iterable[TextBlockParam]] | NotGiven | NOT_GIVEN | |
temperature | float | NotGiven | NOT_GIVEN | |
top_k | int | NotGiven | NOT_GIVEN | |
top_p | float | NotGiven | NOT_GIVEN | |
extra_headers | Headers | None | None | |
extra_query | Query | None | None | |
extra_body | Body | None | None | |
timeout | float | httpx.Timeout | None | NotGiven | NOT_GIVEN |
Exported source
@patch
@delegates(Client.__call__)
def structured(self:Client,
list, # List of messages in the dialog
msgs:list]=None, # List of tools to make available to Claude
tools:Optional[=None, # Class to search for tools
obj:Optional=None, # Namespace to search for tools
ns:Optional[abc.Mapping]**kwargs):
"Return the value of all tool calls (generally used for structured outputs)"
= listify(tools)
tools = self(msgs, tools=tools, tool_choice=tools, **kwargs)
res if ns is None: ns=tools
= getattr(res, 'content', [])
cts = [call_func(o, ns=ns, obj=obj) for o in cts if isinstance(o,ToolUseBlock)]
tcs return tcs
Anthropic’s API does not support response formats directly, so instead we provide a structured
method to use tool calling to achieve the same result. The result of the tool is not passed back to Claude in this case, but instead is returned directly to the user.
=[sums]) c.structured(pr, tools
Finding the sum of 604542 and 6458932
[7063474]
Chat
Rather than manually adding the responses to a dialog, we’ll create a simple Chat
class to do that for us, each time we make a request. We’ll also store the system prompt and tools here, to avoid passing them every time.
Chat
Chat (model:Optional[str]=None, cli:Optional[__main__.Client]=None, sp='', tools:Optional[list]=None, temp=0, cont_pr:Optional[str]=None)
Anthropic chat client.
Type | Default | Details | |
---|---|---|---|
model | Optional | None | Model to use (leave empty if passing cli ) |
cli | Optional | None | Client to use (leave empty if passing model ) |
sp | str | Optional system prompt | |
tools | Optional | None | List of tools to make available to Claude |
temp | int | 0 | Temperature |
cont_pr | Optional | None | User prompt to continue an assistant response: assistant,[user:“…”],assistant |
Exported source
class Chat:
def __init__(self,
str]=None, # Model to use (leave empty if passing `cli`)
model:Optional[=None, # Client to use (leave empty if passing `model`)
cli:Optional[Client]='', # Optional system prompt
splist]=None, # List of tools to make available to Claude
tools:Optional[=0, # Temperature
tempstr]=None): # User prompt to continue an assistant response: assistant,[user:"..."],assistant
cont_pr:Optional["Anthropic chat client."
assert model or cli
assert cont_pr != "", "cont_pr may not be an empty string"
self.c = (cli or Client(model))
self.h,self.sp,self.tools,self.cont_pr,self.temp = [],sp,tools,cont_pr,temp
@property
def use(self): return self.c.use
The class stores the Client
that will provide the responses in c
, and a history of messages in h
.
= "Never mention what tools you use."
sp = Chat(model, sp=sp)
chat chat.c.use, chat.h
(In: 0; Out: 0; Cache create: 0; Cache read: 0; Total: 0, [])
We’ve shown the token usage but we really care about is pricing. Let’s extract the latest pricing from Anthropic into a pricing
dict.
We’ll patch Usage
to enable it compute the cost given pricing.
Usage.cost
Usage.cost (costs:tuple)
Exported source
@patch
def cost(self:Usage, costs:tuple) -> float:
= getattr(self, "cache_creation_input_tokens",0), getattr(self, "cache_read_input_tokens",0)
cache_w, cache_r return sum([self.input_tokens * costs[0] + self.output_tokens * costs[1] + cache_w * costs[2] + cache_r * costs[3]]) / 1e6
chat.c.use.cost(pricing[model_types[chat.c.model]])
0.0
This is clunky. Let’s add cost
as a property for the Chat
class. It will pass in the appropriate prices for the current model to the usage cost calculator.
Chat.cost
Chat.cost ()
Exported source
@patch(as_prop=True)
def cost(self: Chat) -> float: return self.c.use.cost(pricing[model_types[self.c.model]])
chat.cost
0.0
Chat.__call__
Chat.__call__ (pr=None, temp=None, maxtok=4096, stream=False, prefill='', tool_choice:Optional[dict]=None, **kw)
Call self as a function.
Type | Default | Details | |
---|---|---|---|
pr | NoneType | None | Prompt / message |
temp | NoneType | None | Temperature |
maxtok | int | 4096 | Maximum tokens |
stream | bool | False | Stream response? |
prefill | str | Optional prefill to pass to Claude as start of its response | |
tool_choice | Optional | None | Optionally force use of some tool |
kw |
Exported source
@patch
def _stream(self:Chat, res):
yield from res
self.h += mk_toolres(self.c.result, ns=self.tools, obj=self)
Exported source
@patch
def _post_pr(self:Chat, pr, prev_role):
if pr is None and prev_role == 'assistant':
if self.cont_pr is None:
raise ValueError("Prompt must be given after assistant completion, or use `self.cont_pr`.")
= self.cont_pr # No user prompt, keep the chain
pr if pr: self.h.append(mk_msg(pr))
Exported source
@patch
def _append_pr(self:Chat,
=None, # Prompt / message
pr
):= nested_idx(self.h, -1, 'role') if self.h else 'assistant' # First message should be 'user'
prev_role if pr and prev_role == 'user': self() # already user request pending
self._post_pr(pr, prev_role)
Exported source
@patch
def __call__(self:Chat,
=None, # Prompt / message
pr=None, # Temperature
temp=4096, # Maximum tokens
maxtok=False, # Stream response?
stream='', # Optional prefill to pass to Claude as start of its response
prefilldict]=None, # Optionally force use of some tool
tool_choice:Optional[**kw):
if temp is None: temp=self.temp
self._append_pr(pr)
= self.c(self.h, stream=stream, prefill=prefill, sp=self.sp, temp=temp, maxtok=maxtok,
res =self.tools, tool_choice=tool_choice,**kw)
toolsif stream: return self._stream(res)
self.h += mk_toolres(self.c.result, ns=self.tools, obj=self)
return res
The __call__
method just passes the request along to the Client
, but rather than just passing in this one prompt, it appends it to the history and passes it all along. As a result, we now have state!
= Chat(model, sp=sp) chat
"I'm Jeremy")
chat("What's my name?") chat(
Your name is Jeremy, as you mentioned in your previous message.
- id:
msg_01Lxa5M7S1cjCBTQtaZfWWCQ
- content:
[{'text': 'Your name is Jeremy, as you mentioned in your previous message.', 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 64, 'output_tokens': 16, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
chat.use, chat.cost
(In: 81; Out: 55; Cache create: 0; Cache read: 0; Total: 136, 0.001068)
Let’s try out prefill too:
= "Concisely, what is the meaning of life?"
q = 'According to Douglas Adams,' pref
=pref) chat(q, prefill
According to Douglas Adams, the meaning of life is 42. More seriously, there’s no universally agreed upon answer. Common philosophical perspectives include:
- Finding personal fulfillment
- Serving others
- Pursuing happiness
- Creating meaning through our choices
- Experiencing and appreciating existence
Ultimately, many believe each individual must determine their own life’s meaning.
- id:
msg_012Z1zkix1fb1B7fHWZQpMoF
- content:
[{'text': "According to Douglas Adams, the meaning of life is 42. More seriously, there's no universally agreed upon answer. Common philosophical perspectives include:\n\n1. Finding personal fulfillment\n2. Serving others\n3. Pursuing happiness\n4. Creating meaning through our choices\n5. Experiencing and appreciating existence\n\nUltimately, many believe each individual must determine their own life's meaning.", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 100, 'output_tokens': 82, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
By default messages must be in user, assistant, user format. If this isn’t followed (aka calling chat()
without a user message) it will error out:
try: chat()
except ValueError as e: print("Error:", e)
Error: Prompt must be given after assistant completion, or use `self.cont_pr`.
Setting cont_pr
allows a “default prompt” to be specified when a prompt isn’t specified. Usually used to prompt the model to continue.
= "keep going..."
chat.cont_pr chat()
Continuing on the topic of life’s meaning:
- Achieving self-actualization
- Leaving a positive legacy
- Connecting with others and forming relationships
- Exploring and understanding the universe
- Evolving as a species
- Overcoming challenges and growing
- Finding balance between various aspects of life
- Expressing creativity and individuality
- Pursuing knowledge and wisdom
- Living in harmony with nature
These perspectives often overlap and can be combined in various ways. Some argue that the absence of an inherent meaning allows for the freedom to create our own purpose.
- id:
msg_012qPDMHoEpaT9RDDu4UyYKq
- content:
[{'text': "Continuing on the topic of life's meaning:\n\n6. Achieving self-actualization\n7. Leaving a positive legacy\n8. Connecting with others and forming relationships\n9. Exploring and understanding the universe\n10. Evolving as a species\n11. Overcoming challenges and growing\n12. Finding balance between various aspects of life\n13. Expressing creativity and individuality\n14. Pursuing knowledge and wisdom\n15. Living in harmony with nature\n\nThese perspectives often overlap and can be combined in various ways. Some argue that the absence of an inherent meaning allows for the freedom to create our own purpose.", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 188, 'output_tokens': 135, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
We can also use streaming:
= Chat(model, sp=sp)
chat for o in chat("I'm Jeremy", stream=True): print(o, end='')
Hello Jeremy! It's nice to meet you. How are you doing today? Is there anything in particular you'd like to chat about or any questions you have?
for o in chat(q, prefill=pref, stream=True): print(o, end='')
According to Douglas Adams, the meaning of life is 42. More seriously, there's no universally agreed upon answer. Common philosophical perspectives include:
1. Finding personal fulfillment
2. Serving others or a higher purpose
3. Experiencing and creating love and happiness
4. Pursuing knowledge and understanding
5. Leaving a positive legacy
Ultimately, many believe each individual must determine their own meaning.
Chat tool use
We automagically get streamlined tool use as well:
= f"What is {a}+{b}?"
pr pr
'What is 604542+6458932?'
= Chat(model, sp=sp, tools=[sums])
chat = chat(pr)
r r
Finding the sum of 604542 and 6458932
To answer this question, I can use the “sums” function to add these two numbers together. Let me do that for you.
- id:
msg_01Y6nic6azHsQUHgnu92UDTu
- content:
[{'text': 'To answer this question, I can use the "sums" function to add these two numbers together. Let me do that for you.', 'type': 'text'}, {'id': 'toolu_01DrdxEJ2KuqCPRKmKjgCsyZ', 'input': {'a': 604542, 'b': 6458932}, 'name': 'sums', 'type': 'tool_use'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
tool_use
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 428, 'output_tokens': 101, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
Now we need to send this result to Claude—calling the object with no parameters tells it to return the tool result to Claude:
chat()
The sum of 604542 and 6458932 is 7063474.
- id:
msg_01JNastkUht4sTSZ17ayuqdd
- content:
[{'text': 'The sum of 604542 and 6458932 is 7063474.', 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 543, 'output_tokens': 23, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
It should be correct, because it actually used our Python function to do the addition. Let’s check:
+b a
7063474
Images
Claude can handle image data as well. As everyone knows, when testing image APIs you have to use a cute puppy.
# Image is Cute_dog.jpg from Wikimedia
= Path('samples/puppy.jpg')
fn =fn, width=200) display.Image(filename
= fn.read_bytes() img
Exported source
def _add_cache(d, cache):
"Optionally add cache control"
if cache: d["cache_control"] = {"type": "ephemeral"}
return d
Claude supports context caching by adding a cache_control
header, so we provide an option to enable that.
img_msg
img_msg (data:bytes, cache=False)
Convert image data
into an encoded dict
Exported source
def img_msg(data:bytes, cache=False)->dict:
"Convert image `data` into an encoded `dict`"
= base64.b64encode(data).decode("utf-8")
img = mimetypes.types_map['.'+imghdr.what(None, h=data)]
mtype = dict(type="base64", media_type=mtype, data=img)
r return _add_cache({"type": "image", "source": r}, cache)
Anthropic have documented the particular dict
structure that expect image data to be in, so we have a little function to create that for us.
text_msg
text_msg (s:str, cache=False)
Convert s
to a text message
Exported source
def text_msg(s:str, cache=False)->dict:
"Convert `s` to a text message"
return _add_cache({"type": "text", "text": s}, cache)
A Claude message can be a list of image and text parts. So we’ve also created a helper for making the text parts.
= "In brief, what color flowers are in this image?"
q = mk_msg([img_msg(img), text_msg(q)]) msg
c([msg])
The image contains purple or lavender-colored flowers, which appear to be daisies or a similar type of flower.
- id:
msg_011KHngSPHFC2P8TCGxB3LGf
- content:
[{'text': 'The image contains purple or lavender-colored flowers, which appear to be daisies or a similar type of flower.', 'type': 'text'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 110, 'output_tokens': 28, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
Exported source
def _mk_content(src, cache=False):
"Create appropriate content data structure based on type of content"
if isinstance(src,str): return text_msg(src, cache=cache)
if isinstance(src,bytes): return img_msg(src, cache=cache)
if isinstance(src, abc.Mapping): return {k:_str_if_needed(v) for k,v in src.items()}
return _str_if_needed(src)
There’s not need to manually choose the type of message, since we figure that out from the data of the source data.
'Hi') _mk_content(
{'type': 'text', 'text': 'Hi'}
mk_msg
mk_msg (content, role='user', cache=False, **kw)
Helper to create a dict
appropriate for a Claude message. kw
are added as key/value pairs to the message
Type | Default | Details | |
---|---|---|---|
content | A string, list, or dict containing the contents of the message | ||
role | str | user | Must be ‘user’ or ‘assistant’ |
cache | bool | False | |
kw |
Exported source
def mk_msg(content, # A string, list, or dict containing the contents of the message
='user', # Must be 'user' or 'assistant'
role=False,
cache**kw):
"Helper to create a `dict` appropriate for a Claude message. `kw` are added as key/value pairs to the message"
if hasattr(content, 'content'): content,role = content.content,content.role
if isinstance(content, abc.Mapping): content=content.get('content', content)
if not isinstance(content, list): content=[content]
= [_mk_content(o, cache if islast else False) for islast,o in loop_last(content)] if content else '.'
content return dict2obj(dict(role=role, content=content, **kw), list_func=list)
'hi', 'there'], cache=True) mk_msg([
{ 'content': [ {'text': 'hi', 'type': 'text'},
{ 'cache_control': {'type': 'ephemeral'},
'text': 'there',
'type': 'text'}],
'role': 'user'}
= mk_msg(['hi', 'there'], cache=True) m
When we construct a message, we now use _mk_content
to create the appropriate parts. Since a dialog contains multiple messages, and a message can contain multiple content parts, to pass a single message with multiple parts we have to use a list containing a single list:
c([[img, q]])
The image contains purple or lavender-colored flowers, which appear to be daisies or a similar type of flower.
- id:
msg_01FDxZ8umYNK4yPSuUcFqNoE
- content:
[{'text': 'The image contains purple or lavender-colored flowers, which appear to be daisies or a similar type of flower.', 'type': 'text'}]
- model:
claude-3-haiku-20240307
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 110, 'output_tokens': 28, 'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0}
As promised (much!) earlier, we’ve now finally completed our definition of mk_msg
, and this version is the one we export to the Python module.
Third party providers
Amazon Bedrock
These are Amazon’s current Claude models:
models_aws
['anthropic.claude-3-opus-20240229-v1:0',
'anthropic.claude-3-5-sonnet-20240620-v1:0',
'anthropic.claude-3-sonnet-20240229-v1:0',
'anthropic.claude-3-haiku-20240307-v1:0']
anthropic
at version 0.34.2 seems not to install boto3
as a dependency. You may need to do a pip install boto3
or the creation of the Client
below fails.
Provided boto3
is installed, we otherwise don’t need any extra code to support Amazon Bedrock – we just have to set up the approach client:
= AnthropicBedrock(
ab =os.environ['AWS_ACCESS_KEY'],
aws_access_key=os.environ['AWS_SECRET_KEY'],
aws_secret_key
)= Client(models_aws[-1], ab) client
= Chat(cli=client) chat
"I'm Jeremy") chat(
Hello Jeremy! It’s nice to meet you. How can I assist you today? Is there anything specific you’d like to talk about or any questions you have?
- id:
msg_bdrk_01MwjVA5hwyfob3w4vdsqpnU
- content:
[{'text': "Hello Jeremy! It's nice to meet you. How can I assist you today? Is there anything specific you'd like to talk about or any questions you have?", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 10, 'output_tokens': 36}
Google Vertex
models_goog
['claude-3-opus@20240229',
'claude-3-5-sonnet@20240620',
'claude-3-sonnet@20240229',
'claude-3-haiku@20240307']
from anthropic import AnthropicVertex
import google.auth
= google.auth.default()[1]
project_id = "us-east5"
region = AnthropicVertex(project_id=project_id, region=region)
gv = Client(models_goog[-1], gv) client
= Chat(cli=client) chat
"I'm Jeremy") chat(
Hello Jeremy! It’s nice to meet you. How can I assist you today? Is there anything specific you’d like to talk about or any questions you have?
- id:
msg_vrtx_01PFtHewPDe35yShy7vecp5q
- content:
[{'text': "Hello Jeremy! It's nice to meet you. How can I assist you today? Is there anything specific you'd like to talk about or any questions you have?", 'type': 'text'}]
- model:
claude-3-5-sonnet-20240620
- role:
assistant
- stop_reason:
end_turn
- stop_sequence:
None
- type:
message
- usage:
{'input_tokens': 10, 'output_tokens': 36}