diff --git a/examples/research/researcher.py b/examples/research/researcher.py
new file mode 100644
index 000000000..b5513a054
--- /dev/null
+++ b/examples/research/researcher.py
@@ -0,0 +1,91 @@
+from operator import itemgetter
+
+from langchain.chat_models import ChatOpenAI, ChatAnthropic
+from langchain.prompts import SystemMessagePromptTemplate, ChatPromptTemplate
+from langchain.schema.output_parser import StrOutputParser
+from langchain.runnables.openai_functions import OpenAIFunctionsRouter
+import requests
+from fastapi import FastAPI
+
+
+from permchain.connection_inmemory import InMemoryPubSubConnection
+from permchain.pubsub import PubSub
+from permchain.topic import Topic
+from langchain.output_parsers.openai_functions import JsonKeyOutputFunctionsParser
+
+template = """Write between 2 and 5 sub questions that serve as google search queries to search online that form an objective opinion from the following: {question}"""
+functions = [
+    {
+                "name": "sub_questions",
+                "description": "List of sub questions",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "questions": {
+                            "type": "array",
+                            "description": "List of sub questions to ask.",
+                              "items": {
+                                "type": "string"
+                              }
+                        },
+                    },
+                },
+            },
+]
+prompt = ChatPromptTemplate.from_template(template)
+question_chain = prompt | ChatOpenAI(temperature=0).bind(functions=functions, function_call={"name":"sub_questions"}) | JsonKeyOutputFunctionsParser(key_name="questions")
+
+template = """You are tasked with writing a research report to answer the following question:
+
+<question>
+{question}
+</question>
+
+In order to do that, you first came up with several sub questions and researched those. please find those below:
+
+<research>
+{research}
+</research>
+
+Now, write your final report answering the original question!"""
+prompt = ChatPromptTemplate.from_template(template)
+report_chain = prompt | ChatOpenAI() | StrOutputParser()
+
+research_inbox = Topic("research")
+writer_inbox = Topic("writer_inbox")
+
+def web_researcher(questions):
+    response = requests.post("http://127.0.0.1:8081/batch", json={"questions": questions})
+    return response.json()
+
+subquestion_actor = (
+    # Listed in inputs
+    Topic.IN.subscribe()
+    | question_chain
+    # The draft always goes to the editors inbox
+    | research_inbox.publish()
+)
+research_actor = (
+    research_inbox.subscribe()
+    | {
+        "research": lambda x: web_researcher(x),
+        #"research": (lambda x: [web_researcher(i) for i in x]),
+        "question": Topic.IN.current() | itemgetter("question"),
+    }
+    | writer_inbox.publish()
+)
+write_actor = (
+    writer_inbox.subscribe()
+    | {"response": report_chain}
+    | Topic.OUT.publish()
+)
+
+longer_researcher = PubSub(
+    processes=(subquestion_actor, research_actor, write_actor),
+    connection=InMemoryPubSubConnection(),
+)
+
+app = FastAPI()
+@app.get("/report")
+def read_item(question: str):
+    return longer_researcher.invoke({"question":question})
\ No newline at end of file
diff --git a/examples/research/single_question_researcher.py b/examples/research/single_question_researcher.py
new file mode 100644
index 000000000..08ef58d0e
--- /dev/null
+++ b/examples/research/single_question_researcher.py
@@ -0,0 +1,57 @@
+from operator import itemgetter
+from typing import List
+from langchain.chat_models import ChatOpenAI, ChatAnthropic
+from langchain.prompts import SystemMessagePromptTemplate, ChatPromptTemplate
+from langchain.schema.output_parser import StrOutputParser
+from langchain.runnables.openai_functions import OpenAIFunctionsRouter
+from pydantic import BaseModel
+import requests
+from fastapi import FastAPI
+
+
+from permchain.connection_inmemory import InMemoryPubSubConnection
+from permchain.pubsub import PubSub
+from permchain.topic import Topic
+
+prompt = ChatPromptTemplate.from_template("Answer the user's question given the search results\n\n<question>{question}</question><search_results>{search_results}</search_results>")
+
+summarizer_chain = prompt | ChatOpenAI(max_retries=0).with_fallbacks([ChatOpenAI(model="gpt-3.5-turbo-16k"), ChatAnthropic(model="claude-2")]) | StrOutputParser()
+
+
+def retrieve_documents(query):
+    response = requests.get("http://127.0.0.1:8080/query", params={"query": query})
+    return response.json()
+
+
+summarizer_inbox = Topic("summarizer")
+
+search_actor = (
+    Topic.IN.subscribe()
+    | {
+        "search_results": retrieve_documents,
+        "question": Topic.IN.current(),
+    }
+    | summarizer_inbox.publish()
+)
+
+summ_actor = (
+    summarizer_inbox.subscribe()
+    | {"answer":summarizer_chain }
+    |  Topic.OUT.publish()
+)
+
+web_researcher = PubSub(
+    processes=(search_actor, summ_actor),
+    connection=InMemoryPubSubConnection(),
+)
+
+app = FastAPI()
+class Data(BaseModel):
+    questions: List[str]
+@app.get("/invoke")
+def read_item(question: str):
+    return web_researcher.invoke(question)
+
+@app.post("/batch")
+def batch(data: Data):
+    return web_researcher.batch(data.questions)
diff --git a/examples/research/webscraper.py b/examples/research/webscraper.py
new file mode 100644
index 000000000..ecb4cb857
--- /dev/null
+++ b/examples/research/webscraper.py
@@ -0,0 +1,36 @@
+# main.py
+
+from fastapi import FastAPI
+from langchain.document_loaders import AsyncHtmlLoader
+from langchain.document_transformers import Html2TextTransformer
+from duckduckgo_search import DDGS
+
+ddgs = DDGS()
+
+app = FastAPI()
+
+@app.get("/")
+def read_root():
+    return {"Hello": "World"}
+
+@app.get("/query")
+def read_item(query: str):
+    query = query.strip().strip('"')
+    search_results = ddgs.text(query)
+    urls_to_look = []
+    for res in search_results:
+        if res.get("href", None):
+            urls_to_look.append(res["href"])
+        if len(urls_to_look) >= 4:
+            break
+
+    # Relevant urls
+    # Load, split, and add new urls to vectorstore
+    if urls_to_look:
+        loader = AsyncHtmlLoader(urls_to_look)
+        html2text = Html2TextTransformer()
+        docs = loader.load()
+        docs = list(html2text.transform_documents(docs))
+    else:
+        docs = []
+    return docs
diff --git a/examples/web-research.ipynb b/examples/web-research.ipynb
index b1e26c7a9..8191bb4fc 100644
--- a/examples/web-research.ipynb
+++ b/examples/web-research.ipynb
@@ -176,21 +176,21 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 28,
+   "execution_count": 43,
    "id": "2d4a6b51-bd93-47c2-a301-9593a47df7d4",
    "metadata": {},
    "outputs": [],
    "source": [
     "summ_actor = (\n",
     "    summarizer_inbox.subscribe()\n",
-    "    | summarizer_chain\n",
+    "    | {\"answer\": summarizer_chain}\n",
     "    | Topic.OUT.publish()\n",
     ")"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 29,
+   "execution_count": 44,
    "id": "11d3b066-7f95-4ad9-82d0-ba64bbf3e3fa",
    "metadata": {},
    "outputs": [],
@@ -203,12 +203,68 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 30,
+   "execution_count": 45,
    "id": "bc022d51-69f0-4da9-8025-70afdc3cc6a8",
    "metadata": {},
-   "outputs": [],
+   "outputs": [
+    {
+     "name": "stderr",
+     "output_type": "stream",
+     "text": [
+      "Fetching pages: 100%|#############################################################################################################################################################################################| 4/4 [00:00<00:00,  6.44it/s]\n"
+     ]
+    },
+    {
+     "data": {
+      "text/plain": [
+       "[{'answer': 'LangSmith is a platform for building production-grade language model applications. It helps trace and evaluate language model applications and intelligent agents, making it easier to move from prototype to production. LangSmith is developed by LangChain, the company behind the open source LangChain framework. More information can be found in the LangSmith documentation.'}]"
+      ]
+     },
+     "execution_count": 45,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
    "source": [
-    "#web_researcher.invoke(\"What is langsmith?\")"
+    "web_researcher.invoke(\"What is langsmith?\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 47,
+   "id": "000f4f24-15ba-476f-8a33-d023081b18d2",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stderr",
+     "output_type": "stream",
+     "text": [
+      "Fetching pages:   0%|                                                                                                                                                                                                     | 0/4 [00:00<?, ?it/s]\n",
+      "Fetching pages: 100%|#############################################################################################################################################################################################| 4/4 [00:00<00:00,  6.93it/s]\u001b[A\n",
+      "Fetching pages: 100%|#############################################################################################################################################################################################| 4/4 [00:00<00:00,  4.79it/s]\n"
+     ]
+    },
+    {
+     "ename": "TypeError",
+     "evalue": "unsupported operand type(s) for +=: 'dict' and 'dict'",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
+      "Cell \u001b[0;32mIn[47], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43mweb_researcher\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mbatch\u001b[49m\u001b[43m(\u001b[49m\u001b[43m[\u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mwhat is langsmith\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[38;5;124;43mwhat is llama\u001b[39;49m\u001b[38;5;124;43m\"\u001b[39;49m\u001b[43m]\u001b[49m\u001b[43m)\u001b[49m\n",
+      "File \u001b[0;32m~/workplace/langchain/libs/langchain/langchain/schema/runnable/base.py:102\u001b[0m, in \u001b[0;36mRunnable.batch\u001b[0;34m(self, inputs, config, max_concurrency)\u001b[0m\n\u001b[1;32m     99\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m [\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39minvoke(inputs[\u001b[38;5;241m0\u001b[39m], configs[\u001b[38;5;241m0\u001b[39m])]\n\u001b[1;32m    101\u001b[0m \u001b[38;5;28;01mwith\u001b[39;00m ThreadPoolExecutor(max_workers\u001b[38;5;241m=\u001b[39mmax_concurrency) \u001b[38;5;28;01mas\u001b[39;00m executor:\n\u001b[0;32m--> 102\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mlist\u001b[39;49m\u001b[43m(\u001b[49m\u001b[43mexecutor\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mmap\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43minvoke\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43minputs\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mconfigs\u001b[49m\u001b[43m)\u001b[49m\u001b[43m)\u001b[49m\n",
+      "File \u001b[0;32m~/.pyenv/versions/3.10.1/lib/python3.10/concurrent/futures/_base.py:608\u001b[0m, in \u001b[0;36mExecutor.map.<locals>.result_iterator\u001b[0;34m()\u001b[0m\n\u001b[1;32m    605\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m fs:\n\u001b[1;32m    606\u001b[0m     \u001b[38;5;66;03m# Careful not to keep a reference to the popped future\u001b[39;00m\n\u001b[1;32m    607\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m timeout \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[0;32m--> 608\u001b[0m         \u001b[38;5;28;01myield\u001b[39;00m \u001b[43mfs\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpop\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mresult\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    609\u001b[0m     \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    610\u001b[0m         \u001b[38;5;28;01myield\u001b[39;00m fs\u001b[38;5;241m.\u001b[39mpop()\u001b[38;5;241m.\u001b[39mresult(end_time \u001b[38;5;241m-\u001b[39m time\u001b[38;5;241m.\u001b[39mmonotonic())\n",
+      "File \u001b[0;32m~/.pyenv/versions/3.10.1/lib/python3.10/concurrent/futures/_base.py:445\u001b[0m, in \u001b[0;36mFuture.result\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    443\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m CancelledError()\n\u001b[1;32m    444\u001b[0m \u001b[38;5;28;01melif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_state \u001b[38;5;241m==\u001b[39m FINISHED:\n\u001b[0;32m--> 445\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m__get_result\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    446\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    447\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mTimeoutError\u001b[39;00m()\n",
+      "File \u001b[0;32m~/.pyenv/versions/3.10.1/lib/python3.10/concurrent/futures/_base.py:390\u001b[0m, in \u001b[0;36mFuture.__get_result\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    388\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_exception:\n\u001b[1;32m    389\u001b[0m     \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m--> 390\u001b[0m         \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_exception\n\u001b[1;32m    391\u001b[0m     \u001b[38;5;28;01mfinally\u001b[39;00m:\n\u001b[1;32m    392\u001b[0m         \u001b[38;5;66;03m# Break a reference cycle with the exception in self._exception\u001b[39;00m\n\u001b[1;32m    393\u001b[0m         \u001b[38;5;28mself\u001b[39m \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m\n",
+      "File \u001b[0;32m~/.pyenv/versions/3.10.1/lib/python3.10/concurrent/futures/thread.py:58\u001b[0m, in \u001b[0;36m_WorkItem.run\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m     55\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m\n\u001b[1;32m     57\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m---> 58\u001b[0m     result \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mfn\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43margs\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[38;5;241;43m*\u001b[39;49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mkwargs\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m     59\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mBaseException\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m exc:\n\u001b[1;32m     60\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mfuture\u001b[38;5;241m.\u001b[39mset_exception(exc)\n",
+      "File \u001b[0;32m~/workplace/permchain/permchain/pubsub.py:67\u001b[0m, in \u001b[0;36mPubSub.invoke\u001b[0;34m(self, input, config)\u001b[0m\n\u001b[1;32m     65\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21minvoke\u001b[39m(\u001b[38;5;28mself\u001b[39m, \u001b[38;5;28minput\u001b[39m: Any, config: Optional[RunnableConfig] \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m Any:\n\u001b[1;32m     66\u001b[0m     collected \u001b[38;5;241m=\u001b[39m []\n\u001b[0;32m---> 67\u001b[0m     \u001b[38;5;28;01mfor\u001b[39;00m chunk \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mstream(\u001b[38;5;28minput\u001b[39m, config):\n\u001b[1;32m     68\u001b[0m         collected\u001b[38;5;241m.\u001b[39mappend(chunk)\n\u001b[1;32m     69\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m collected\n",
+      "File \u001b[0;32m~/workplace/permchain/permchain/pubsub.py:176\u001b[0m, in \u001b[0;36mPubSub.stream\u001b[0;34m(self, input, config, max_concurrency)\u001b[0m\n\u001b[1;32m    174\u001b[0m             final_output \u001b[38;5;241m=\u001b[39m chunk\n\u001b[1;32m    175\u001b[0m         \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m--> 176\u001b[0m             final_output \u001b[38;5;241m+\u001b[39m\u001b[38;5;241m=\u001b[39m chunk\n\u001b[1;32m    177\u001b[0m \u001b[38;5;28;01mfinally\u001b[39;00m:\n\u001b[1;32m    178\u001b[0m     \u001b[38;5;66;03m# Cleanup\u001b[39;00m\n\u001b[1;32m    179\u001b[0m     \u001b[38;5;28;01mfor\u001b[39;00m fut \u001b[38;5;129;01min\u001b[39;00m inflight:\n",
+      "\u001b[0;31mTypeError\u001b[0m: unsupported operand type(s) for +=: 'dict' and 'dict'"
+     ]
+    }
+   ],
+   "source": [
+    "web_researcher.batch([\"what is langsmith\", \"what is llama\"])"
    ]
   },
   {