وكلاء الذكاء الاصطناعي من الصفر: التكرارات والسلاسل

import ollama
llm = "qwen2.5" 

import yfinance as yf

stock = "MSFT"
yf.Ticker(ticker=stock).history(period='5d') #1d,5d,1mo,3mo,6mo,1y,2y,5y,10y,ytd,max

import matplotlib.pyplot as plt

def get_stock(ticker:str, period:str, col:str):
    data = yf.Ticker(ticker=ticker).history(period=period)
    if len(data) > 0:
        data[col].plot(color="black", legend=True, xlabel='', title=f"{ticker.upper()} ({period})").grid()
        plt.show()
        return 'ok'
    else:
        return 'no'

tool_get_stock = {'type':'function', 'function':{
  'name': 'get_stock',
  'description': 'Download stock data',
  'parameters': {'type': 'object',
                'required': ['ticker','period','col'],
                'properties': {
                    'ticker': {'type':'str', 'description':'the ticker symbol of the stock.'},
                    'period': {'type':'str', 'description':"for 1 month input '1mo', for 6 months input '6mo', for 1 year input '1y'. Use '1y' if not specified."},
                    'col': {'type':'str', 'description':"one of 'Open','High','Low','Close','Volume'. Use 'Close' if not specified."},
}}}}

## test
get_stock(ticker="msft", period="1y", col="Close")

def use_tool(agent_res:dict, dic_tools:dict) -> dict:
    ## use tool
    if "tool_calls" in agent_res["message"].keys():
        for tool in agent_res["message"]["tool_calls"]:
            t_name, t_inputs = tool["function"]["name"], tool["function"]["arguments"]
            if f := dic_tools.get(t_name):
                ### calling tool
                print('🔧 >', f"\x1b[1;31m{t_name} -> Inputs: {t_inputs}\x1b[0m")
                ### tool output
                t_output = f(**tool["function"]["arguments"])
                print(t_output)
                ### final res
                res = t_output
            else:
                print('🤬 >', f"\x1b[1;31m{t_name} -> NotFound\x1b[0m")
    ## don't use tool
    if agent_res['message']['content'] != '':
        res = agent_res["message"]["content"]
        t_name, t_inputs = '', ''
    return {'res':res, 'tool_used':t_name, 'inputs_used':t_inputs}

prompt = '''You are a financial analyst, assist the user using your available tools.'''
messages = [{"role":"system", "content":prompt}]
dic_tools = {'get_stock':get_stock}

while True:
    ## user input
    try:
        q = input('🙂 >')
    except EOFError:
        break
    if q == "quit":
        break
    if q.strip() == "":
        continue
    messages.append( {"role":"user", "content":q} )
   
    ## model
    agent_res = ollama.chat(model=llm, messages=messages,
                            tools=[tool_get_stock])
    dic_res = use_tool(agent_res, dic_tools)
    res, tool_used, inputs_used = dic_res["res"], dic_res["tool_used"], dic_res["inputs_used"]
   
    ## final response
    print("👽 >", f"\x1b[1;30m{res}\x1b[0m")
    messages.append( {"role":"assistant", "content":res} )

max_i, i = 3, 0
    while res == 'no' and i < max_i:
        comment = f'''I used tool '{tool_used}' with inputs {inputs_used}. But it didn't work, so I must try again with different inputs'''
        messages.append( {"role":"assistant", "content":comment} )
        agent_res = ollama.chat(model=llm, messages=messages,
                                tools=[tool_get_stock])
        dic_res = use_tool(agent_res, dic_tools)
        res, tool_used, inputs_used = dic_res["res"], dic_res["tool_used"], dic_res["inputs_used"]
       
        i += 1
        if i == max_i:
            res = f'I tried {i} times but something is wrong'

    ## final response
    print("👽 >", f"\x1b[1;30m{res}\x1b[0m")
    messages.append( {"role":"assistant", "content":res} )

from langchain_community.tools import DuckDuckGoSearchResults

def search_web(query:str) -> str:
  return DuckDuckGoSearchResults(backend="news").run(query)

tool_search_web = {'type':'function', 'function':{
  'name': 'search_web',
  'description': 'Search the web',
  'parameters': {'type': 'object',
                'required': ['query'],
                'properties': {
                    'query': {'type':'str', 'description':'the topic or subject to search on the web'},
}}}}

## test
search_web(query="facebook stock")

prompt = '''
[GOAL] You are a financial analyst, assist the user using your available tools.

[RETURN] You must return the stock data that the user asks for.

[WARNINGS] In order to retrieve stock data, you need to know the ticker symbol of the company.

[CONTEXT] First ALWAYS try to use the tool 'get_stock'.
If it doesn't work, you can use the tool 'search_web' and search 'company name stock'.
Get information about the stock and deduct what is the right ticker symbol of the company.
Then, you can use AGAIN the tool 'get_stock' with the ticker you got using the previous tool.
'''

max_i, i = 3, 0
    while res in ['no',''] and i < max_i:
        comment = f'''I used tool '{tool_used}' with inputs {inputs_used}. But it didn't work, so I must try a different way.'''
        messages.append( {"role":"assistant", "content":comment} )
        agent_res = ollama.chat(model=llm, messages=messages,
                                tools=[tool_get_stock, tool_search_web])
        dic_res = use_tool(agent_res, dic_tools)
        res, tool_used, inputs_used = dic_res["res"], dic_res["tool_used"], dic_res["inputs_used"]

        ## chain: output of previous tool = input of next tool
        if tool_used == 'search_web':
            query = q+". You must return just the compay ticker.\nContext: "+res
            llm_res = ollama.generate(model=llm, prompt=query)["response"]
            messages.append( {"role":"user", "content":f"try ticker: {llm_res}"} )
           
            print("👽 >", f"\x1b[1;30mI can try with {llm_res}\x1b[0m")
           
            agent_res = ollama.chat(model=llm, messages=messages, tools=[tool_get_stock])
            dic_res = use_tool(agent_res, dic_tools)
            res, tool_used, inputs_used = dic_res["res"], dic_res["tool_used"], dic_res["inputs_used"]
        i += 1        if i == max_i:            res = f'I tried {i} times but something is wrong'
    
 ## final response    
 print("👽 >", f"\x1b[1;30m{res}\x1b[0m")    
 messages.append( {"role":"assistant", "content":res} )