run-llama/llama-lab
1
0
Fork 0
mirror of https://github.com/run-llama/llama-lab.git synced 2026-07-01 20:14:02 -04:00
Code Issues 12 Packages Projects Releases Wiki Activity

[llama_agi] Create llama_agi python package (#18)

Browse Source 
* Refactor into python package

* Add steamlit runner and example

* Update frontpage readme

* Update README

* pin langchain + llama_index versions

* Add some docstrings

* Use dataclass instead of typed dict
This commit is contained in:
Logan
2023-04-23 19:29:05 -06:00
committed by GitHub
parent 639df52277
commit 8bd7bd0129
30 changed files with 930 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -0,0 +1,3 @@
.mypy_cache
.ruff_cache
__pycache__
README.md
+2 -2
View File
@@ -33,11 +33,11 @@ This will run in a loop until the task list is empty (or maybe you run out of Op
Example Usage:
```python
cd llama_agi
cd llama_agi/examples
export OPENAI_API_KEY="key"
export GOOGLE_API_KEY="key"
export GOOGLE_CSE_ID="id"
python ./run_llama_agi.py --initial-task "Create a task list" --objective "Solve world hunger" --sleep 2
python ./auto_runner_example.py --initial-task "Create a task list" --objective "Solve world hunger" --sleep 2
```
```bash
llama_agi/LICENSE
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2023 Logan Markewich
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.
llama_agi/Makefile
+9
View File
@@ -0,0 +1,9 @@
.PHONY: format lint
format:
black .
lint:
mypy .
black . --check
ruff check .
llama_agi/README.md
+60
View File
@@ -0,0 +1,60 @@
# 🤖 Llama AGI 🦙
This python package allows you to quickly create Auto-GPT-like agents, using LlamaIndex and Langchain.
## Setup
Install using pip:
```bash
pip install llama-agi
```
Or install from source:
```bash
git clone https://github.com/run-llama/llama-lab.git
cd llama-lab/llama_agi
pip install -e .
```
## Example Usage
The following shows an example of setting up the `AutoAGIRunner`, which will continue completing tasks (nearly) indefinitely, trying to achieve it's initial objective of "Solve world hunger."
```python
from langchain.agents import load_tools
from langchain.llms import OpenAI
from llama_agi.execution_agent import ToolExecutionAgent
from llama_agi.runners import AutoAGIRunner
from llama_agi.task_manager import LlamaTaskManager
from llama_agi.tools import search_notes, record_note, search_webpage
from llama_index import ServiceContext, LLMPredictor
# LLM setup
llm = OpenAI(temperature=0, model_name='text-davinci-003')
service_context = ServiceContext.from_defaults(llm_predictor=LLMPredictor(llm=llm), chunk_size_limit=512)
# llama_agi setup
task_manager = LlamaTaskManager([args.initial_task], task_service_context=service_context)
tools = load_tools(["google-search-results-json"])
tools = tools + [search_notes, record_note, search_webpage]
execution_agent = ToolExecutionAgent(llm=llm, tools=tools)
# launch the auto runner
runner = AutoAGIRunner(task_manager, execution_agent)
objective = "Solve world hunger"
initial_task = "Create a list of tasks"
sleep_time = 2
runner.run(objective, initial_task, sleep_time)
```
More examples can be found in the `examples` folder!
## Llama Ecosystem
- LlamaIndex (connecting your LLMs to data): https://github.com/jerryjliu/llama_index
- LlamaHub (community library of data loaders): https://llamahub.ai
llama_agi/agi/ExecutionAgent.py
-98
View File
@@ -1,98 +0,0 @@
from langchain.agents import AgentExecutor, ZeroShotAgent, load_tools
from langchain.chains import LLMChain
from langchain.llms import OpenAI, BaseLLM
from langchain.chat_models.base import BaseChatModel
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from typing import Optional, Union
from agi.task_prompts import LC_EXECUTION_PROMPT, LC_PREFIX, LC_SUFFIX
from agi.tools.NoteTakingTools import record_note, search_notes
from agi.tools.WebpageSearchTool import search_webpage
class BaseExecutionAgent:
def __init__(
self,
llm: Optional[Union[BaseLLM, BaseChatModel]] = None,
model_name: str = "text-davinci-003",
) -> None:
if llm:
self._llm = llm
elif model_name == "text-davinci-003":
self._llm = OpenAI(temperature=0, model_name=model_name, max_tokens=512)
else:
self._llm = ChatOpenAI(temperature=0, model_name=model_name, max_tokens=512)
def execute_task(
self, objective: str, task: str, completed_tasks_summary: str
) -> str:
raise NotImplementedError("execute_task not implemented in BaseExecutionAgent")
class SimpleExecutionAgent(BaseExecutionAgent):
def __init__(
self,
llm: Optional[Union[BaseLLM, BaseChatModel]] = None,
model_name: str = "text-davinci-003",
) -> None:
super().__init__(llm=llm, model_name=model_name)
self._prompt_template = PromptTemplate(
template=LC_EXECUTION_PROMPT,
input_variables=["task", "objective", "completed_tasks_summary"],
)
self._execution_chain = LLMChain(llm=self._llm, prompt=self._prompt_template)
def execute_task(
self, objective: str, task: str, completed_tasks_summary: str
) -> str:
result = self._execution_chain.predict(
objective=objective,
task=task,
completed_tasks_summary=completed_tasks_summary,
)
return result
class ToolExecutionAgent(BaseExecutionAgent):
def __init__(
self,
llm: Optional[Union[BaseLLM, BaseChatModel]] = None,
model_name: str = "text-davinci-003",
) -> None:
super().__init__(llm=llm, model_name=model_name)
# use some default langchain tools
self._tools = load_tools(["google-search-results-json"])
# add our custom tools
self._tools = self._tools + [search_notes, record_note, search_webpage]
# create the agent
self._agent_prompt = ZeroShotAgent.create_prompt(
self._tools,
prefix=LC_PREFIX,
suffix=LC_SUFFIX,
input_variables=[
"objective",
"task",
"agent_scratchpad",
"completed_tasks_summary",
],
)
self._llm_chain = LLMChain(llm=self._llm, prompt=self._agent_prompt)
self._agent = ZeroShotAgent(
llm_chain=self._llm_chain, tools=self._tools, verbose=True
)
self._execution_chain = AgentExecutor.from_agent_and_tools(
agent=self._agent, tools=self._tools, verbose=True
)
def execute_task(
self, objective: str, task: str, completed_tasks_summary: str
) -> str:
result = self._execution_chain.run(
objective=objective,
task=task,
completed_tasks_summary=completed_tasks_summary,
)
return result
llama_agi/agi/tools/__init__.py
View File
llama_agi/agi/utils.py
-49
View File
@@ -1,49 +0,0 @@
import os
from llama_index import (
GPTSimpleVectorIndex,
GPTListIndex,
ServiceContext,
)
def initialize_task_list_index(
documents, llm_predictor=None, index_path="./task_index.json", chunk_size_limit=2000
):
service_context = ServiceContext.from_defaults(
llm_predictor=llm_predictor, chunk_size_limit=chunk_size_limit
)
if os.path.exists(index_path):
return GPTListIndex.load_from_disk(index_path, service_context=service_context)
else:
return GPTListIndex.from_documents(documents, service_context=service_context)
def initialize_search_index(
documents,
llm_predictor=None,
index_path="./search_index.json",
chunk_size_limit=2000,
):
service_context = ServiceContext.from_defaults(
llm_predictor=llm_predictor, chunk_size_limit=chunk_size_limit
)
if os.path.exists(index_path):
return GPTSimpleVectorIndex.load_from_disk(
index_path, service_context=service_context
)
else:
return GPTSimpleVectorIndex.from_documents(
documents, service_context=service_context
)
def log_current_status(cur_task, result, completed_tasks_summary, task_list):
status_string = f"""
==================================
Completed Tasks Summary: {completed_tasks_summary.strip()}
Current Task: {cur_task.strip()}
Result: {result.strip()}
Task List: {", ".join([x.get_text().strip() for x in task_list])}
==================================
"""
print(status_string, flush=True)
llama_agi/examples/auto_runner_example.py
+56
View File
@@ -0,0 +1,56 @@
import argparse
from langchain.agents import load_tools
from langchain.llms import OpenAI
from llama_agi.execution_agent import ToolExecutionAgent
from llama_agi.runners import AutoAGIRunner
from llama_agi.task_manager import LlamaTaskManager
from llama_agi.tools import search_notes, record_note, search_webpage
from llama_index import ServiceContext, LLMPredictor
if __name__ == "__main__":
parser = argparse.ArgumentParser(
prog="Llama AGI",
description="A baby-agi/auto-gpt inspired application, powered by Llama Index!",
)
parser.add_argument(
"-it",
"--initial-task",
default="Create a list of tasks",
help="The initial task for the system to carry out. Default='Create a list of tasks'",
)
parser.add_argument(
"-o",
"--objective",
default="Solve world hunger",
help="The overall objective for the system. Default='Solve world hunger'",
)
parser.add_argument(
"--sleep-time",
default=2,
help="Sleep time (in seconds) between each task loop. Default=2",
type=int,
)
args = parser.parse_args()
# LLM setup
llm = OpenAI(temperature=0, model_name="text-davinci-003")
service_context = ServiceContext.from_defaults(
llm_predictor=LLMPredictor(llm=llm), chunk_size_limit=512
)
# llama_agi setup
task_manager = LlamaTaskManager(
[args.initial_task], task_service_context=service_context
)
tools = load_tools(["google-search-results-json"])
tools = tools + [search_notes, record_note, search_webpage]
execution_agent = ToolExecutionAgent(llm=llm, tools=tools)
# launch the auto runner
runner = AutoAGIRunner(task_manager, execution_agent)
runner.run(args.objective, args.initial_task, args.sleep_time)
llama_agi/examples/streamlit_runner_example.py
+87
View File
@@ -0,0 +1,87 @@
import os
import streamlit as st
from langchain.agents import load_tools
from langchain.chat_models import ChatOpenAI
from langchain.llms import OpenAI
from llama_agi.execution_agent import ToolExecutionAgent
from llama_agi.runners import AutoStreamlitAGIRunner
from llama_agi.task_manager import LlamaTaskManager
from llama_agi.tools import search_notes, record_note, search_webpage
from llama_index import ServiceContext, LLMPredictor
st.set_page_config(layout="wide")
st.header("🤖 Llama AGI 🦙")
st.text("Use the setup tab to configure your LLM settings and initial objective+tasks.")
st.text("Then use the Launch tab to run the AGI. Kill the process by clicking 'Stop' in the top right.")
setup_tab, launch_tab = st.tabs(["Setup", "Launch"])
with setup_tab:
st.subheader("LLM Setup")
col1, col2, col3 = st.columns(3)
with col1:
openai_api_key = st.text_input("Enter your OpenAI API key here", type="password")
llm_name = st.selectbox(
"Which LLM?", ["text-davinci-003", "gpt-3.5-turbo", "gpt-4"]
)
with col2:
google_api_key = st.text_input("Enter your Google API key here", type="password")
model_temperature = st.slider(
"LLM Temperature", min_value=0.0, max_value=1.0, step=0.1, value=0.0
)
with col3:
google_cse_id = st.text_input("Enter your Google CSE ID key here", type="password")
max_tokens = st.slider(
"LLM Max Tokens", min_value=256, max_value=1024, step=8, value=512
)
st.subheader("AGI Setup")
objective = st.text_input("Objective:", value="Solve world hunger")
initial_task = st.text_input("Initial Task:", value="Create a list of tasks")
if st.button('Initialize?'):
os.environ['OPENAI_API_KEY'] = openai_api_key
os.environ['GOOGLE_API_KEY'] = google_api_key
os.environ['GOOGLE_CSE_ID'] = google_cse_id
if llm_name == "text-davinci-003":
llm = OpenAI(
temperature=model_temperature, model_name=llm_name, max_tokens=max_tokens
)
else:
llm= ChatOpenAI(
temperature=model_temperature, model_name=llm_name, max_tokens=max_tokens
)
service_context = ServiceContext.from_defaults(
llm_predictor=LLMPredictor(llm=llm), chunk_size_limit=512
)
st.session_state['task_manager'] = LlamaTaskManager(
[initial_task], task_service_context=service_context
)
tools = load_tools(["google-search-results-json"])
tools = tools + [search_notes, record_note, search_webpage]
st.session_state['execution_agent'] = ToolExecutionAgent(llm=llm, tools=tools)
st.session_state['initial_task'] = initial_task
st.session_state['objective'] = objective
st.session_state['init'] = True
st.success("Initialized!")
with launch_tab:
st.subheader("AGI Status")
if st.button("Launch"):
if st.session_state.get('init', False):
# launch the auto runner
with st.spinner("Running!"):
runner = AutoStreamlitAGIRunner(st.session_state['task_manager'], st.session_state['execution_agent'])
runner.run(st.session_state['objective'], st.session_state['initial_task'], 2)
llama_agi/agi/__init__.py → llama_agi/llama_agi/__init__.py
View File
llama_agi/agi/task_prompts.py → llama_agi/llama_agi/default_task_prompts.py
+2 -12
View File
@@ -14,27 +14,17 @@ NO_COMPLETED_TASKS_SUMMARY = "You haven't completed any tasks yet."
#############################################
##### Langchain - Execution Agent (Unused Currently) #####
##### Langchain - Execution Agent #####
#############################################
LC_PREFIX = PREFIX + "You have access to the following tools:"
LC_FORMAT_INSTRUCTIONS = """Use the following format:
Task: the current task you must complete
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I have now completed the task
Final Answer: the final answer to the original input task"""
LC_SUFFIX = (
"This is your current objective: {objective}\n"
"Take into account what you have already achieved: {completed_tasks_summary}\n"
"Using your current objective, your previously completed tasks, and your available tools,"
"Complete the current task.\n"
"Begin!\n"
"Task: {task}\n"
"Task: {cur_task}\n"
"Thought: {agent_scratchpad}"
)
llama_agi/llama_agi/execution_agent/SimpleExecutionAgent.py
+65
View File
@@ -0,0 +1,65 @@
from typing import Any, Dict, List, Optional, Union
from string import Formatter
from langchain.agents.tools import Tool
from langchain.chains import LLMChain
from langchain.llms import BaseLLM
from langchain.chat_models.base import BaseChatModel
from langchain.prompts import PromptTemplate
from llama_agi.execution_agent.base import BaseExecutionAgent, LlamaAgentPrompts
class SimpleExecutionAgent(BaseExecutionAgent):
"""Simple Execution Agent
This agent uses an LLM to execute a basic action without tools.
The LlamaAgentPrompts.execution_prompt defines how this execution agent
behaves.
Usually, this is used for simple tasks, like generating the initial list of tasks.
The execution template kwargs are automatically extracted and expected to be
specified in execute_task().
Args:
llm (Union[BaseLLM, BaseChatModel]): The langchain LLM class to use.
model_name: (str): The name of the OpenAI model to use, if the LLM is
not provided.
max_tokens: (int): The maximum number of tokens the LLM can generate.
prompts: (LlamaAgentPrompts): The prompt templates used during execution.
The only prompt used byt the SimpleExecutionAgent is
LlamaAgentPrompts.execution_prompt.
"""
def __init__(
self,
llm: Optional[Union[BaseLLM, BaseChatModel]] = None,
model_name: str = "text-davinci-003",
max_tokens: int = 512,
prompts: LlamaAgentPrompts = LlamaAgentPrompts(),
tools: Optional[List[Tool]] = None,
) -> None:
super().__init__(
llm=llm,
model_name=model_name,
max_tokens=max_tokens,
prompts=prompts,
tools=tools,
)
self.execution_prompt = self.prompts.execution_prompt
input_variables = [
fn
for _, fn, _, _ in Formatter().parse(self.execution_prompt)
if fn is not None
]
self._prompt_template = PromptTemplate(
template=self.execution_prompt,
input_variables=input_variables,
)
self._execution_chain = LLMChain(llm=self._llm, prompt=self._prompt_template)
def execute_task(self, **prompt_kwargs: Any) -> Dict[str, str]:
"""Execute a task."""
result = self._execution_chain.predict(**prompt_kwargs)
return {"output": result}
llama_agi/llama_agi/execution_agent/ToolExecutionAgent.py
+80
View File
@@ -0,0 +1,80 @@
from typing import Any, Dict, List, Optional, Union
from string import Formatter
from langchain.agents import AgentExecutor, ZeroShotAgent
from langchain.agents.tools import Tool
from langchain.chains import LLMChain
from langchain.llms import BaseLLM
from langchain.chat_models.base import BaseChatModel
from llama_agi.execution_agent.base import BaseExecutionAgent, LlamaAgentPrompts
class ToolExecutionAgent(BaseExecutionAgent):
"""Tool Execution Agent
This agent is a wrapper around the zero-shot agent from Langchain. Using
a set of tools, the agent is expected to carry out and complete some task
that will help achieve an overall objective.
The agents overall behavior is controlled by the LlamaAgentPrompts.agent_prefix
and LlamaAgentPrompts.agent_suffix prompt templates.
The execution template kwargs are automatically extracted and expected to be
specified in execute_task().
execute_task() also returns the intermediate steps, for additional debugging and is
used for the streamlit example.
Args:
llm (Union[BaseLLM, BaseChatModel]): The langchain LLM class to use.
model_name: (str): The name of the OpenAI model to use, if the LLM is
not provided.
max_tokens: (int): The maximum number of tokens the LLM can generate.
prompts: (LlamaAgentPrompts): The prompt templates used during execution.
The Tool Execution Agent uses LlamaAgentPrompts.agent_prefix and
LlamaAgentPrompts.agent_suffix.
tools: (List[Tool]): The list of langchain tools for the execution agent to use.
"""
def __init__(
self,
llm: Optional[Union[BaseLLM, BaseChatModel]] = None,
model_name: str = "text-davinci-003",
max_tokens: int = 512,
prompts: LlamaAgentPrompts = LlamaAgentPrompts(),
tools: Optional[List[Tool]] = None,
) -> None:
super().__init__(
llm=llm,
model_name=model_name,
max_tokens=max_tokens,
prompts=prompts,
tools=tools,
)
self.agent_prefix = self.prompts.agent_prefix
self.agent_suffix = self.prompts.agent_suffix
# create the agent
input_variables = [
fn for _, fn, _, _ in Formatter().parse(self.agent_prefix) if fn is not None
] + [
fn for _, fn, _, _ in Formatter().parse(self.agent_suffix) if fn is not None
]
self._agent_prompt = ZeroShotAgent.create_prompt(
self.tools,
prefix=self.agent_prefix,
suffix=self.agent_suffix,
input_variables=input_variables,
)
self._llm_chain = LLMChain(llm=self._llm, prompt=self._agent_prompt)
self._agent = ZeroShotAgent(
llm_chain=self._llm_chain, tools=self.tools, verbose=True
)
self._execution_chain = AgentExecutor.from_agent_and_tools(
agent=self._agent, tools=self.tools, verbose=True, return_intermediate_steps=True
)
def execute_task(self, **prompt_kwargs: Any) -> Dict[str, str]:
"""Execute a task, using tools."""
result = self._execution_chain(prompt_kwargs)
return result
llama_agi/llama_agi/execution_agent/__init__.py
+7
View File
@@ -0,0 +1,7 @@
from .SimpleExecutionAgent import SimpleExecutionAgent
from .ToolExecutionAgent import ToolExecutionAgent
__all__ = [
SimpleExecutionAgent,
ToolExecutionAgent
]
llama_agi/llama_agi/execution_agent/base.py
+58
View File
@@ -0,0 +1,58 @@
from abc import abstractmethod
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Union
from langchain.agents.tools import Tool
from langchain.llms import OpenAI, BaseLLM
from langchain.chat_models.base import BaseChatModel
from langchain.chat_models import ChatOpenAI
from llama_agi.default_task_prompts import (
LC_PREFIX, LC_SUFFIX, LC_EXECUTION_PROMPT
)
@dataclass
class LlamaAgentPrompts:
execution_prompt: str = LC_EXECUTION_PROMPT
agent_prefix: str = LC_PREFIX
agent_suffix: str = LC_SUFFIX
class BaseExecutionAgent:
"""Base Execution Agent
Args:
llm (Union[BaseLLM, BaseChatModel]): The langchain LLM class to use.
model_name: (str): The name of the OpenAI model to use, if the LLM is
not provided.
max_tokens: (int): The maximum number of tokens the LLM can generate.
prompts: (LlamaAgentPrompts): The prompt templates used during execution.
tools: (List[Tool]): The list of langchain tools for the execution
agent to use.
"""
def __init__(
self,
llm: Optional[Union[BaseLLM, BaseChatModel]] = None,
model_name: str = "text-davinci-003",
max_tokens: int = 512,
prompts: LlamaAgentPrompts = LlamaAgentPrompts(),
tools: Optional[List[Tool]] = None,
) -> None:
if llm:
self._llm = llm
elif model_name == "text-davinci-003":
self._llm = OpenAI(
temperature=0, model_name=model_name, max_tokens=max_tokens
)
else:
self._llm = ChatOpenAI(
temperature=0, model_name=model_name, max_tokens=max_tokens
)
self.max_tokens = max_tokens
self.prompts = prompts
self.tools = tools if tools else []
@abstractmethod
def execute_task(self, **prompt_kwargs: Any) -> Dict[str, str]:
"""Execute a task."""
llama_agi/llama_agi/runners/AutoAGIRunner.py
+81
View File
@@ -0,0 +1,81 @@
import time
from typing import List, Optional
from llama_agi.runners.base import BaseAGIRunner
from llama_agi.execution_agent.SimpleExecutionAgent import SimpleExecutionAgent
from llama_agi.utils import log_current_status
class AutoAGIRunner(BaseAGIRunner):
def run(
self,
objective: str,
initial_task: str,
sleep_time: int,
initial_task_list: Optional[List[str]] = None,
) -> None:
# get initial list of tasks
if initial_task_list:
self.task_manager.add_new_tasks(initial_task_list)
else:
initial_completed_tasks_summary = (
self.task_manager.get_completed_tasks_summary()
)
initial_task_prompt = initial_task + "\nReturn the list as an array."
# create simple execution agent using current agent
simple_execution_agent = SimpleExecutionAgent(
llm=self.execution_agent._llm,
max_tokens=self.execution_agent.max_tokens,
prompts=self.execution_agent.prompts,
)
initial_task_list_result = simple_execution_agent.execute_task(
objective=objective,
task=initial_task_prompt,
completed_tasks_summary=initial_completed_tasks_summary,
)
initial_task_list = self.task_manager.parse_task_list(initial_task_list_result['output'])
# add tasks to the task manager
self.task_manager.add_new_tasks(initial_task_list)
# prioritize initial tasks
self.task_manager.prioritize_tasks(objective)
completed_tasks_summary = initial_completed_tasks_summary
while True:
# Get the next task
cur_task = self.task_manager.get_next_task()
# Execute current task
result = self.execution_agent.execute_task(
objective=objective,
cur_task=cur_task,
completed_tasks_summary=completed_tasks_summary,
)['output']
# store the task and result as completed
self.task_manager.add_completed_task(cur_task, result)
# generate new task(s), if needed
self.task_manager.generate_new_tasks(objective, cur_task, result)
# Summarize completed tasks
completed_tasks_summary = self.task_manager.get_completed_tasks_summary()
# log state of AGI to terminal
log_current_status(
cur_task,
result,
completed_tasks_summary,
self.task_manager.current_tasks,
)
# Quit the loop?
if len(self.task_manager.current_tasks) == 0:
print("Out of tasks! Objective Accomplished?")
break
# wait a bit to let you read what's happening
time.sleep(sleep_time)
llama_agi/llama_agi/runners/AutoStreamlitAGIRunner.py
+123
View File
@@ -0,0 +1,123 @@
import json
import streamlit as st
import time
from typing import List, Optional
from llama_agi.runners.base import BaseAGIRunner
from llama_agi.execution_agent.SimpleExecutionAgent import SimpleExecutionAgent
from llama_agi.utils import log_current_status
def make_intermediate_steps_pretty(json_str: str) -> List[str]:
steps = json.loads(json_str)
output = []
for action_set in steps:
for step in action_set:
if isinstance(step, list):
output.append(step[-1])
else:
output.append(step)
return output
class AutoStreamlitAGIRunner(BaseAGIRunner):
def run(
self,
objective: str,
initial_task: str,
sleep_time: int,
initial_task_list: Optional[List[str]] = None,
) -> None:
logs_col, state_col = st.columns(2)
logs = []
with logs_col:
st.subheader("Execution Log")
st_logs = st.empty()
st_logs.write("No logs yet!")
with state_col:
st.subheader("AGI State")
st_state = st.empty()
st_state.write("No state yet!")
# get initial list of tasks
if initial_task_list:
self.task_manager.add_new_tasks(initial_task_list)
else:
initial_completed_tasks_summary = (
self.task_manager.get_completed_tasks_summary()
)
initial_task_prompt = initial_task + "\nReturn the list as an array."
# create simple execution agent using current agent
simple_execution_agent = SimpleExecutionAgent(
llm=self.execution_agent._llm,
max_tokens=self.execution_agent.max_tokens,
prompts=self.execution_agent.prompts,
)
initial_task_list_result = simple_execution_agent.execute_task(
objective=objective,
task=initial_task_prompt,
completed_tasks_summary=initial_completed_tasks_summary,
)
initial_task_list = self.task_manager.parse_task_list(initial_task_list_result['output'])
# add tasks to the task manager
self.task_manager.add_new_tasks(initial_task_list)
# prioritize initial tasks
self.task_manager.prioritize_tasks(objective)
completed_tasks_summary = initial_completed_tasks_summary
# update streamlit state
state_str = log_current_status(initial_task, initial_task_list_result['output'], completed_tasks_summary, self.task_manager.current_tasks, return_str=True)
if state_str:
st_state.markdown(state_str.replace("\n", "\n\n"))
while True:
# Get the next task
cur_task = self.task_manager.get_next_task()
# Execute current task
result_dict = self.execution_agent.execute_task(
objective=objective,
cur_task=cur_task,
completed_tasks_summary=completed_tasks_summary,
)
result = result_dict['output']
# update logs
log = make_intermediate_steps_pretty(json.dumps(result_dict['intermediate_steps'])) + [result]
logs.append(log)
st_logs.write(log)
# store the task and result as completed
self.task_manager.add_completed_task(cur_task, result)
# generate new task(s), if needed
self.task_manager.generate_new_tasks(objective, cur_task, result)
# Summarize completed tasks
completed_tasks_summary = self.task_manager.get_completed_tasks_summary()
# log state of AGI to streamlit
state_str = log_current_status(
cur_task,
result,
completed_tasks_summary,
self.task_manager.current_tasks,
return_str=True
)
if state_str is not None:
st_state.markdown(state_str.replace("\n", "\n\n"))
# Quit the loop?
if len(self.task_manager.current_tasks) == 0:
print("Out of tasks! Objective Accomplished?")
break
# wait a bit to let you read what's happening
time.sleep(sleep_time)
llama_agi/llama_agi/runners/__init__.py
+7
View File
@@ -0,0 +1,7 @@
from .AutoAGIRunner import AutoAGIRunner
from .AutoStreamlitAGIRunner import AutoStreamlitAGIRunner
__all__ = [
AutoAGIRunner,
AutoStreamlitAGIRunner
]
llama_agi/llama_agi/runners/base.py
+23
View File
@@ -0,0 +1,23 @@
from abc import abstractmethod
from typing import List, Optional
from llama_agi.execution_agent.base import BaseExecutionAgent
from llama_agi.task_manager.base import BaseTaskManager
class BaseAGIRunner:
def __init__(
self, task_manager: BaseTaskManager, execution_agent: BaseExecutionAgent
) -> None:
self.task_manager = task_manager
self.execution_agent = execution_agent
@abstractmethod
def run(
self,
objective: str,
initial_task: str,
sleep_time: int,
initial_task_list: Optional[List[str]] = None,
) -> None:
"""Run the task manager and execution agent in a loop."""
llama_agi/agi/TaskManager.py → llama_agi/llama_agi/task_manager/LlamaTaskManager.py
+69 -25
View File
@@ -1,34 +1,63 @@
import re
import json
from typing import List, Tuple
from llama_index import Document
from typing import List, Tuple, Optional
from llama_index import Document, ServiceContext
from llama_index.prompts.prompts import QuestionAnswerPrompt, RefinePrompt
from agi.utils import initialize_task_list_index
from agi.task_prompts import (
DEFAULT_TASK_PRIORITIZE_TMPL,
DEFAULT_REFINE_TASK_PRIORITIZE_TMPL,
DEFAULT_TASK_CREATE_TMPL,
DEFAULT_REFINE_TASK_CREATE_TMPL,
NO_COMPLETED_TASKS_SUMMARY,
)
from llama_agi.task_manager.base import BaseTaskManager, LlamaTaskPrompts
from llama_agi.utils import initialize_task_list_index
from llama_agi.default_task_prompts import NO_COMPLETED_TASKS_SUMMARY
class TaskManager:
def __init__(self, tasks: List[str]) -> None:
self.current_tasks = [Document(x) for x in tasks]
self.completed_tasks: List[Document] = []
class LlamaTaskManager(BaseTaskManager):
"""Llama Task Manager
This task manager uses LlamaIndex to create and prioritize tasks. Using
the LlamaTaskPrompts, the task manager will create tasks that work
towards achieving an overall objective.
New tasks are created based on the prev task+result, completed tasks summary,
and the overall objective.
Tasks are then prioritized using the overall objective and current list of tasks.
Args:
tasks (List[str]): The initial list of tasks to complete.
prompts: (LlamaTaskPrompts): The prompts to control the task creation
and prioritization.
tasK_service_context (ServiceContext): The LlamaIndex service context to use
for task creation and prioritization.
"""
def __init__(
self,
tasks: List[str],
prompts: LlamaTaskPrompts = LlamaTaskPrompts(),
task_service_context: Optional[ServiceContext] = None,
) -> None:
super().__init__(
tasks=tasks, prompts=prompts, task_service_context=task_service_context
)
self.current_tasks_index = initialize_task_list_index(
self.current_tasks, index_path="current_tasks_index.json"
self.current_tasks, service_context=self.task_service_context
)
self.completed_tasks_index = initialize_task_list_index(
self.completed_tasks, index_path="completed_tasks_index.json"
self.completed_tasks, service_context=self.task_service_context
)
self.task_create_qa_template = self.prompts.task_create_qa_template
self.task_create_refine_template = self.prompts.task_create_refine_template
self.task_prioritize_qa_template = self.prompts.task_prioritize_qa_template
self.task_prioritize_refine_template = self.prompts.task_prioritize_refine_template
def _get_task_create_templates(
self, prev_task: str, prev_result: str
) -> Tuple[QuestionAnswerPrompt, RefinePrompt]:
text_qa_template = DEFAULT_TASK_CREATE_TMPL.format(
"""Fetch the task create prompts as llama_index objects."""
text_qa_template = self.task_create_qa_template.format(
prev_result=prev_result,
prev_task=prev_task,
query_str="{query_str}",
@@ -36,7 +65,7 @@ class TaskManager:
)
llama_text_qa_template = QuestionAnswerPrompt(text_qa_template)
refine_template = DEFAULT_REFINE_TASK_CREATE_TMPL.format(
refine_template = self.task_create_refine_template.format(
prev_result=prev_result,
prev_task=prev_task,
query_str="{query_str}",
@@ -50,13 +79,14 @@ class TaskManager:
def _get_task_prioritize_templates(
self,
) -> Tuple[QuestionAnswerPrompt, RefinePrompt]:
"""Fetch the task prioritize prompts as llama_index objects."""
return (
QuestionAnswerPrompt(DEFAULT_TASK_PRIORITIZE_TMPL),
RefinePrompt(DEFAULT_REFINE_TASK_PRIORITIZE_TMPL),
QuestionAnswerPrompt(self.task_prioritize_qa_template),
RefinePrompt(self.task_prioritize_refine_template),
)
def parse_task_list(self, task_list_str: str) -> List[str]:
# Try to parse lists with json, fallback to regex
"""Parse new tasks generated by the agent."""
new_tasks: List[str] = []
try:
new_tasks = json.loads(task_list_str)
@@ -71,6 +101,7 @@ class TaskManager:
return new_tasks
def get_completed_tasks_summary(self) -> str:
"""Generate a summary of completed tasks."""
if len(self.completed_tasks) == 0:
return NO_COMPLETED_TASKS_SUMMARY
summary = self.completed_tasks_index.query(
@@ -79,6 +110,7 @@ class TaskManager:
return str(summary)
def prioritize_tasks(self, objective: str) -> None:
"""Prioritize the current list of incomplete tasks."""
(text_qa_template, refine_template) = self._get_task_prioritize_templates()
prioritized_tasks = self.current_tasks_index.query(
objective,
@@ -92,11 +124,14 @@ class TaskManager:
if len(task) > 10:
new_tasks.append(task)
self.current_tasks = [Document(x) for x in new_tasks]
self.current_tasks_index = initialize_task_list_index(self.current_tasks)
self.current_tasks_index = initialize_task_list_index(
self.current_tasks, service_context=self.task_service_context
)
def generate_new_tasks(
self, objective: str, prev_task: str, prev_result: str
) -> None:
"""Generate new tasks given the previous task and result."""
(text_qa_template, refine_template) = self._get_task_create_templates(
prev_task, prev_result
)
@@ -109,17 +144,26 @@ class TaskManager:
self.add_new_tasks(new_tasks)
def get_next_task(self) -> str:
"""Get the next task to complete."""
next_task = self.current_tasks.pop().get_text()
self.current_tasks_index = initialize_task_list_index(self.current_tasks)
self.current_tasks_index = initialize_task_list_index(
self.current_tasks, service_context=self.task_service_context
)
return next_task
def add_new_tasks(self, tasks: List[str]) -> None:
"""Add new tasks to the task manager."""
for task in tasks:
if task not in self.current_tasks and task not in self.completed_tasks:
self.current_tasks.append(Document(task))
self.current_tasks_index = initialize_task_list_index(self.current_tasks)
self.current_tasks_index = initialize_task_list_index(
self.current_tasks, service_context=self.task_service_context
)
def add_completed_task(self, task: str, result: str) -> None:
"""Add a task as completed."""
document = Document(f"Task: {task}\nResult: {result}\n")
self.completed_tasks.append(document)
self.completed_tasks_index = initialize_task_list_index(self.completed_tasks)
self.completed_tasks_index = initialize_task_list_index(
self.completed_tasks, service_context=self.task_service_context
)
llama_agi/llama_agi/task_manager/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from .LlamaTaskManager import LlamaTaskManager
__all__ = [
LlamaTaskManager,
]
llama_agi/llama_agi/task_manager/base.py
+73
View File
@@ -0,0 +1,73 @@
from abc import abstractmethod
from dataclasses import dataclass
from typing import List, Optional
from llama_index import Document, ServiceContext
from llama_agi.default_task_prompts import (
DEFAULT_TASK_PRIORITIZE_TMPL,
DEFAULT_REFINE_TASK_PRIORITIZE_TMPL,
DEFAULT_TASK_CREATE_TMPL,
DEFAULT_REFINE_TASK_CREATE_TMPL,
)
@dataclass
class LlamaTaskPrompts:
task_create_qa_template: str = DEFAULT_TASK_CREATE_TMPL
task_create_refine_template: str = DEFAULT_REFINE_TASK_CREATE_TMPL
task_prioritize_qa_template: str = DEFAULT_TASK_PRIORITIZE_TMPL
task_prioritize_refine_template: str = DEFAULT_REFINE_TASK_PRIORITIZE_TMPL
class BaseTaskManager:
"""Base Task Manager
Args:
tasks (List[str]): The initial list of tasks to complete.
prompts: (LlamaTaskPrompts): The prompts to control the task creation
and prioritization.
tasK_service_context (ServiceContext): The LlamaIndex service context to use
for task creation and prioritization.
"""
def __init__(
self,
tasks: List[str],
prompts: LlamaTaskPrompts = LlamaTaskPrompts(),
task_service_context: Optional[ServiceContext] = None,
) -> None:
self.current_tasks = [Document(x) for x in tasks]
self.completed_tasks: List[Document] = []
self.prompts = prompts
self.task_service_context = task_service_context
@abstractmethod
def parse_task_list(self, task_list_str: str) -> List[str]:
"""Parse new tasks generated by the agent."""
@abstractmethod
def get_completed_tasks_summary(self) -> str:
"""Generate a summary of completed tasks."""
@abstractmethod
def prioritize_tasks(self, objective: str) -> None:
"""Prioritize the current list of incomplete tasks."""
@abstractmethod
def generate_new_tasks(
self, objective: str, prev_task: str, prev_result: str
) -> None:
"""Generate new tasks given the previous task and result."""
@abstractmethod
def get_next_task(self) -> str:
"""Get the next task to complete."""
@abstractmethod
def add_new_tasks(self, tasks: List[str]) -> None:
"""Add new tasks to the task manager."""
@abstractmethod
def add_completed_task(self, task: str, result: str) -> None:
"""Add a task as completed."""
llama_agi/agi/tools/NoteTakingTools.py → llama_agi/llama_agi/tools/NoteTakingTools.py
+1 -1
View File
@@ -1,6 +1,6 @@
from langchain.agents import tool
from llama_index import Document
from agi.utils import initialize_search_index
from llama_agi.utils import initialize_search_index
note_index = initialize_search_index([])
llama_agi/agi/tools/WebpageSearchTool.py → llama_agi/llama_agi/tools/WebpageSearchTool.py
+4 -3
View File
@@ -1,7 +1,7 @@
from langchain.agents import tool
from llama_index import download_loader
from llama_index import download_loader, ServiceContext
from agi.utils import initialize_search_index
from llama_agi.utils import initialize_search_index
BeautifulSoupWebReader = download_loader("BeautifulSoupWebReader")
@@ -18,7 +18,8 @@ def search_webpage(prompt: str) -> str:
try:
documents = loader.load_data(urls=[url])
index = initialize_search_index(documents, chunk_size_limit=512)
service_context = ServiceContext.from_defaults(chunk_size_limit=512)
index = initialize_search_index(documents, service_context=service_context)
query_result = index.query(
query_str, similarity_top_k=3, response_mode="compact"
)
llama_agi/llama_agi/tools/__init__.py
+8
View File
@@ -0,0 +1,8 @@
from .NoteTakingTools import record_note, search_notes
from .WebpageSearchTool import search_webpage
__all__ = [
record_note,
search_notes,
search_webpage
]
llama_agi/llama_agi/utils.py
+36
View File
@@ -0,0 +1,36 @@
from typing import Any, List, Optional
from llama_index import GPTSimpleVectorIndex, GPTListIndex, ServiceContext, Document
from llama_index.indices.base import BaseGPTIndex
def initialize_task_list_index(
documents: List[Document], service_context: Optional[ServiceContext] = None
) -> BaseGPTIndex[Any]:
return GPTListIndex.from_documents(documents, service_context=service_context)
def initialize_search_index(
documents: List[Document], service_context: Optional[ServiceContext] = None
) -> BaseGPTIndex[Any]:
return GPTSimpleVectorIndex.from_documents(
documents, service_context=service_context
)
def log_current_status(
cur_task: str, result: str, completed_tasks_summary: str, task_list: List[Document], return_str: bool = False
) -> Optional[str]:
status_string = f"""
__________________________________
Completed Tasks Summary: {completed_tasks_summary.strip()}
Current Task: {cur_task.strip()}
Result: {result.strip()}
Task List: {", ".join([x.get_text().strip() for x in task_list])}
__________________________________
"""
if return_str:
return status_string
else:
print(status_string, flush=True)
return None
llama_agi/pyproject.toml
+49
View File
@@ -0,0 +1,49 @@
[tool.poetry]
name = "llama_agi"
version = "0.1.0"
description = "Building AGI loops using LlamaIndex and Langchain"
authors = []
license = "MIT"
readme = "README.md"
repository = "https://github.com/run-llama/llama-lab/tree/main/llama_agi"
include = [
"LICENSE",
]
keywords = ["LLM", "LlamaIndex", "Langchain", "AGI"]
[tool.poetry.dependencies]
python = ">=3.8.1,<4.0"
langchain = "==0.0.141"
llama-index = "==0.5.16"
streamlit = ">=1.21.0"
[tool.poetry.group.lint.dependencies]
ruff = "^0.0.249"
black = "^23.1.0"
[tool.poetry.group.typing.dependencies]
mypy = "^0.991"
[tool.poetry.group.dev.dependencies]
setuptools = "^67.6.1"
[tool.mypy]
ignore_missing_imports = "True"
disallow_untyped_defs = "True"
exclude = ["notebooks", "build", "examples"]
[tool.ruff]
exclude = [
".venv",
"__pycache__",
".ipynb_checkpoints",
".mypy_cache",
".ruff_cache",
"examples",
"notebooks",
]
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
llama_agi/requirements.txt
+1
View File
@@ -1,2 +1,3 @@
langchain==0.0.141
llama-index==0.5.16
streamlit==1.21.0
llama_agi/run_llama_agi.py
-90
View File
@@ -1,90 +0,0 @@
import logging
import sys
import argparse
import json
import time
from agi.ExecutionAgent import SimpleExecutionAgent, ToolExecutionAgent
from agi.TaskManager import TaskManager
from agi.utils import log_current_status
def run_llama_agi(objective: str, initial_task: str, sleep_time: int) -> None:
task_manager = TaskManager([initial_task])
simple_execution_agent = SimpleExecutionAgent()
tool_execution_agent = ToolExecutionAgent()
# get initial list of tasks
initial_completed_tasks_summary = task_manager.get_completed_tasks_summary()
initial_task_prompt = initial_task + "\nReturn the list as an array."
initial_task_list_str = simple_execution_agent.execute_task(
objective, initial_task_prompt, initial_completed_tasks_summary
)
initial_task_list = task_manager.parse_task_list(initial_task_list_str)
# add tasks to the task manager
task_manager.add_new_tasks(initial_task_list)
# prioritize initial tasks
task_manager.prioritize_tasks(objective)
completed_tasks_summary = initial_completed_tasks_summary
while True:
# Get the next task
cur_task = task_manager.get_next_task()
# Execute current task
result = tool_execution_agent.execute_task(
objective, cur_task, completed_tasks_summary
)
# store the task and result as completed
task_manager.add_completed_task(cur_task, result)
# generate new task(s), if needed
task_manager.generate_new_tasks(objective, cur_task, result)
# Summarize completed tasks
completed_tasks_summary = task_manager.get_completed_tasks_summary()
# log state of AGI to terminal
log_current_status(
cur_task, result, completed_tasks_summary, task_manager.current_tasks
)
# Quit the loop?
if len(task_manager.current_tasks) == 0:
print("Out of tasks! Objective Accomplished?")
break
# wait a bit to let you read what's happening
time.sleep(sleep_time)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
prog="Llama AGI",
description="A baby-agi/auto-gpt inspired application, powered by Llama Index!",
)
parser.add_argument(
"-it",
"--initial-task",
default="Create a list of tasks",
help="The initial task for the system to carry out. Default='Create a list of tasks'",
)
parser.add_argument(
"-o",
"--objective",
default="Solve world hunger",
help="The overall objective for the system. Default='Solve world hunger'",
)
parser.add_argument(
"--sleep",
default=2,
help="Sleep time (in seconds) between each task loop. Default=2",
type=int,
)
args = parser.parse_args()
run_llama_agi(args.objective, args.initial_task, args.sleep)