Tuesday, June 16, 2009

Have Conversations with @AmyIris

The Amy Iris Project is a project to build a better bot. The project is underway, and utilizes the ALICE chat-bot system as a base.

As a very rudimentary demonstration of the Amy Iris System, I've connected the @amyiris Twitter account to the API. Source code is below. So she will reply to Direct Messages, by sending back a Direct Message, if you follow the instructions below.

Here are screen shots of a dialog that I was having with @amyiris while testing. Unfortunately, Twitter doesn't organize this as a back-and-forth dialog, but you get the idea. First, here are the Direct Messages I sent:





And Amy Iris dutifully wrote back to me:






Amy Iris isn't all that well trained at the moment. She will respond mostly with ALICE answers, but it'll give you a chance to play with her API and converse with her. At the time of this publication, she is programmed to use ALICE answers, as well as code snippets that I have discussed in other blog posts: the Best Buy store look-up and the language translation snippets. This is just a rudimentary demo.

To activate @amyiris so that she responds to your direct messages, do the following:
  1. Follow @amyiris (so that you two can communicate). If you already follow @amyiris, you can skip this step.
  2. Send a public Tweet that contain the words "talk to @amyiris". Amy Iris is looking for those words, using Twitter search, and will follow you as soon as she notices (if she doesn't already follow you). You must do step 2, for Amy Iris to answer you in step 3, even if she already follows you.
  3. Send a Direct Message to Amy Iris (example: DM amyiris Hi Amy Iris!)
  4. She will send you a DM response back. Converse as many times as you want.
  5. When you are finished, simply stop conversing with her. Or you leave the test group by "checking out". Simply send a public tweet that contain the words "check out @amyiris". Stupidly clever, eh?

Here are some links to make this process easier:

Click here to JOIN the test group, and talk to @amyiris (this is step 2, above)
Click here to CHECK OUT of the test group. (or just quit sending her DM's.) (this is step 5, above)


Here are some things to talk to @amyiris about:
  • Ask Amy Iris to translate something to spanish or to french. (Example: "Say chair in spanish")
  • Ask Amy Iris where a Best Buy store is in a nearby city or zip code. (Example: "Where is a Best Buy store near 45249?")
  • Or just talk to her. Most conversation will be handled by the ALICE interface, so the responses might sound a bit nonsensical at the moment.

Notes:

  • She's dumb right now - mostly ALICE, plus the translator and Best Buy Store Locator.
  • I've temporarily disabled the code submission portion of the API for now, so you can't make her smarter (yet).
  • Right now she shares one session for all Twitter users. So if one twitter user tells her that his name is Fred, and another user asks what his name is, Amy Iris will think it's Fred.
  • I use Mike Verdone's Python Twitter Tools as my Twitter API. I like this MUCH better than the old Python-Twitter version I was using earlier. You can follow Mike's installation instructions if you want to be thorough. But I think really all you need are two files: api.py and twitter_globals.py, both of which you can view on github at the links provided. Personally, I have those two files, plus a tiny __init__.py file, stored in a folder called twitter, right in the directory with my project, and I am set to go.
  • I heavily borrowed from Mike Verdone's PTT, to create my API. So don't accuse me of ripping it off - I admit it! His GREAT code is licensed to permit this.
  • So that I don't hammer the Twitter servers too hard, I have some delays in the code. If people are actively conversing, I have a 2 second delay in the loop. But if she's fairly idle, then I increase the delay. First to 4 seconds, then 8, then 16, 32, then 40 seconds (max). So if no one's talking to her, you may have an average 20 second delay (which should be fine).

Here's my code:



import time, sys
import amyirisapi
import twitter as twitterapi # this is the Mike Verdone's PTT

# Delays, so we don't hammer Twitter's servers, especially when idle
MIN_PAUSE_INTERVAL = 2.0 # 2 second pause at a minimum
MAX_PAUSE_INTERVAL = 40.0 # 40 second pause, max
PAUSE_INTERVAL_MULTIPIER = 2.0# keep doubling the interval, from min,
# up to max, if nothing's happening
pause_interval = MIN_PAUSE_INTERVAL

# Search Strings
SIGNUP = "talk to" # to start chatting with @amyiris
CANCEL = "check out" # to check out of the test group

SEARCH_TWEET = "@amyiris"



########################################################
# #
# CUSTOMIZE THIS SECTION FOR YOUR USE #
# #
twitter_username="amyiris" #
twitter_password=open("pwd.txt","r").read() #
# #
# Feel free to change to: twitter_password="yourpass" #
# I put mine in a file to hide it from you! #
# #
# END OF CUSTOM SECTION #
# #
########################################################



# twitter api docs request that you put something meaningful into
# the agent string, so they can more easily track usage.
agent = twitter_username + " agent messing with PTT"


# This section initializes a bunch of stuff so I can
# restart the bot later, and pick up where I left off

testers = {} # a dictionary of testers
search_since = 0L # tracks the highest message id searched
dm_since = 0L # tracks the highest Direct Message received

try: #should fail the first time, if the file doesn't exist
chat_track = open("chat_track.txt","r")
testers = eval(chat_track.readline())
search_since = long(chat_track.readline())
dm_since = long(chat_track.readline())
chat_track.close()
except:
pass

# instantiate the APIs for Twitter, Twitter Search, and Amy Iris
twitter = twitterapi.Twitter(twitter_username,twitter_password,agent=agent)
twitter_search = twitterapi.Twitter(twitter_username,twitter_password,agent=agent,
domain="search.twitter.com")
amyiris = amyirisapi.AmyIris()


while True: # loop forever (obviously)

something_happened = False # tracking whether anyone's interacting with
# us on Twitter; start loop at False

# Search for public tweets requesting interaction
try:
search_results = twitter_search.search(q="",
phrase=SEARCH_TWEET, since_id=search_since)
search_since = search_results['max_id']
if search_results['results']:
something_happened=True

for tweet in reversed(search_results['results']):
if SIGNUP in tweet['text']:
print tweet['from_user'],"wants to talk."
try:
if twitter.friendships.exists(user_a=tweet['from_user'],
user_b=twitter_username):
try: #try to follow them so I can receive DMs
twitter.friendships.create(screen_name=tweet['from_user'])
except twitterapi.TwitterError:
pass # hopefully just because we already follow them,
# but possibly because of network / twitter issues
testers[tweet['from_user']]=True
except twitterapi.TwitterError:
pass
elif CANCEL in tweet['text']:
testers[tweet['from_user']]=False
print tweet['from_user'],"finished talking."

except twitterapi.TwitterError:

pass # probably a 403 - Forbidden from Twitter search, due to rate throttling.
# possibly a fail whale or network issue

# Now process and reply to Direct Messages to me

try:
direct_messages = twitter.direct_messages(since_id=dm_since)
if direct_messages:
dm_since=direct_messages[0]['id']
something_happened=True

for dm in reversed(direct_messages):
try:
# Make sure this is a tester first
if testers.get(dm['sender']['screen_name'],False):
amyresponse = amyiris.textin.submit(textin=dm['text'])
twitter.direct_messages.new(
user=dm['sender_id'],
text=amyresponse)
print dm['sender_id'],dm['text']
print amyresponse
except:
pass

except twitterapi.TwitterError:
pass


# Store our placeholder in the file, in case we want to restart

chat_track = open("chat_track.txt","w")
chat_track.write(repr(testers)+"\n")
chat_track.write(str(search_since)+"\n")
chat_track.write(str(dm_since)+"\n")
chat_track.close()


# Pause for a bit, to be kind to Twitter's servers

if something_happened:
pause_interval = MIN_PAUSE_INTERVAL
else:
pause_interval = min(pause_interval * PAUSE_INTERVAL_MULTIPIER,
MAX_PAUSE_INTERVAL)
time.sleep(pause_interval)



And here is the API file (ver 1.0) so that you can build your own interface to Amy Iris. Name this file amyirisapi.py:




# This is Amy Iris API version 1.0, Licensed to all via an MIT license:
#
# Much of the code adapted from Python Twitter Toolkit
# Thank You, Mike Verdone! http://mike.verdone.ca/twitter/
#
# Copyright (c) 2009 Jerry Felix
#
# Portions of this software are
# Copyright (c) 2008 Mike Verdone
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.




DOMAIN = "api.amyiris.com"
POST_ACTIONS = ['submit','register','new','post']

from base64 import b64encode
from urllib import urlencode

import urllib2

from exceptions import Exception

def _py26OrGreater():
import sys
return sys.hexversion > 0x20600f0

if _py26OrGreater():
import json
else:
import simplejson as json

class AmyIrisError(Exception):
"""
Exception thrown by the AmyIris object when there is an
error interacting with amyiris.com.
"""
pass

class AmyIrisCall(object):
def __init__(
self, username, password, format, domain, uri="", agent=None):
self.username = username
self.password = password
self.format = format
self.domain = domain
self.uri = uri
self.agent = agent
def __getattr__(self, k):
try:
return object.__getattr__(self, k)
except AttributeError:
return AmyIrisCall(
self.username, self.password, self.format, self.domain,
self.uri + "/" + k, self.agent)
def __call__(self, **kwargs):
uri = self.uri
method = "GET"
for action in POST_ACTIONS:
if self.uri.endswith(action):
method = "POST"
if (self.agent):
kwargs["source"] = self.agent
break

id = kwargs.pop('id', None)
if id:
uri += "/%s" %(id)

argStr = ""
argData = None
encoded_kwargs = urlencode(kwargs.items())
if (method == "GET"):
if kwargs:
argStr = "?%s" %(encoded_kwargs)
else:
argData = encoded_kwargs

headers = {}
if (self.agent):
headers["X-AmyIris-Client"] = self.agent
if (self.username):
headers["Authorization"] = "Basic " + b64encode("%s:%s" %(
self.username, self.password))



req = urllib2.Request(
"http://%s%s.%s/%s" %(self.domain, uri, self.format, argStr),
argData, headers
)
try:
handle = urllib2.urlopen(req)
if "json" == self.format:
return json.loads(handle.read())
else:
return handle.read()
except urllib2.HTTPError, e:
if (e.code == 304):
return []
else:
raise AmyIrisError(
"AmyIris sent status %i for URL: %s.%s using parameters: (%s)\ndetails: %s" %(
e.code, uri, self.format, encoded_kwargs, e.fp.read()))

class AmyIris(AmyIrisCall):
"""
The AmyIris API class.

Get RESTful data by accessing members of this class. The result
is decoded python objects (lists and dicts).

"""
def __init__(
self, username=None, password=None, format="json", domain=DOMAIN,
agent=None):
"""
Create a new amyiris API connector using the specified
credentials (username and password). Format specifies the output
format ("json" (default) or maybe "xml" in the future).
"""
if (format not in ("json", )):
raise AmyIrisError("Unknown data format '%s'" %(format))
AmyIrisCall.__init__(self, username, password, format, domain, "", agent)

__all__ = ["AmyIris", "AmyIrisError"]


class Code():
"""
Code container for submission of Code Snippets.

Use this class for submitting or reviewing Code Snippets and their meta data.

"""
def __init__(self,**kwargs):

self.agree="checked"
self.rowid="None"
self.author="Anonymous User"
self.publish="Test"

for arg in kwargs:
self.__dict__[arg]=kwargs[arg]






I hope you find this interesting and mildly amusing. I'm sure you can imagine, this is just the beginning. Our goal is to have community contributions to create more powerful Conversational Interfaces.

Feedback, please! And Click Here to ReTweet if you like this post!

2 comments:

ithoughts.de said...

I like the idea ... and am happy that you're back to blogging :-)

Amy Iris said...

Thanks, ithoughts!

Yeah, I had to take a bit of a hiatus, but all is well. I think the project looks great now, and I am anxious to share it with the world. Bit by bit.