使用 Toolhouse 的销售潜在客户工作流¶
在本 notebook 中，您将学习如何使用 Toolhouse 和 LlamaIndex 创建销售潜在客户工作流。销售潜在客户发掘使公司能够根据其业务的价值主张和目标市场找到完美的潜在客户。
该工作流将使用单个代理执行以下活动：
- 它将要求代理通过获取其登录页面的内容来确定业务的价值主张。
- 它将搜索互联网，寻找可能从该业务的产品中受益的潜在客户。
- 它将确定最适合联系的公司。
- 它将为选定的公司起草一封个性化电子邮件。
初始设置¶
确保所有必需的库都已安装。本示例使用 Groq 上的 Llama 3.2，但您可以使用 LlamaIndex 支持的任何 LLM。
In [ ]
Copied!
%pip install llama-index
%pip install llama-index-llms-groq
%pip install toolhouse
%pip install llama-index %pip install llama-index-llms-groq %pip install toolhouse
接下来，我们将传入 API 密钥。
获取 Toolhouse API 密钥
- 注册 Toolhouse 或 登录 如果您是现有用户。
- 如果您是新用户，请复制您在入职过程中收到的自动生成的 API 密钥。现有用户可以在 API 密钥页面获取 API 密钥。
- 将 API 密钥粘贴到下方。
获取 Groq API 密钥：在 Groq 上获取访问权限，然后将您的 API 密钥粘贴到下方。
重要提示：在生产环境中安全地存储您的 API 密钥。
In [ ]
Copied!
import os
os.environ[
"TOOLHOUSE_API_KEY"
] = "Get your Toolhouse API key at https://join.toolhouse.ai"
os.environ[
"GROQ_API_KEY"
] = "Get your Groq API key at https://console.groq.com"
import os os.environ[ "TOOLHOUSE_API_KEY" ] = "Get your Toolhouse API key at https://join.toolhouse.ai" os.environ[ "GROQ_API_KEY" ] = "Get your Groq API key at https://console.groq.com"
导入库¶
我们将导入 LlamaIndex 和 Toolhouse。然后我们初始化 Toolhouse 和 Groq LLM。
In [ ]
Copied!
from llama_index.llms.groq import Groq
from llama_index.core.agent import ReActAgent
from llama_index.core.memory import ChatMemoryBuffer
from toolhouse import Toolhouse, Provider
from llama_index.core.workflow import (
Context,
Event,
StartEvent,
StopEvent,
Workflow,
step,
)
from llama_index.llms.groq import Groq from llama_index.core.agent import ReActAgent from llama_index.core.memory import ChatMemoryBuffer from toolhouse import Toolhouse, Provider from llama_index.core.workflow import ( Context, Event, StartEvent, StopEvent, Workflow, step, )
In [ ]
Copied!
llm = Groq(model="llama-3.2-11b-vision-preview")
th = Toolhouse(provider=Provider.LLAMAINDEX)
th.set_metadata("id", "llamaindex_agent")
th.set_metadata("timezone", 0)
llm = Groq(model="llama-3.2-11b-vision-preview") th = Toolhouse(provider=Provider.LLAMAINDEX) th.set_metadata("id", "llamaindex_agent") th.set_metadata("timezone", 0)
In [ ]
Copied!
class WebsiteContentEvent(Event):
contents: str
class WebSearchEvent(Event):
results: str
class RankingEvent(Event):
results: str
class LogEvent(Event):
msg: str
class SalesRepWorkflow(Workflow):
agent = ReActAgent(
tools=th.get_tools(bundle="llamaindex test"),
llm=llm,
memory=ChatMemoryBuffer.from_defaults(),
)
@step
async def get_company_info(
self, ctx: Context, ev: StartEvent
) -> WebsiteContentEvent:
ctx.write_event_to_stream(
LogEvent(msg=f"Getting the contents of {ev.url}…")
)
prompt = f"Get the contents of {ev.url}, then summarize its key value propositions in a few bullet points."
contents = await self.agent.achat(prompt)
return WebsiteContentEvent(contents=str(contents.response))
@step
async def find_prospects(
self, ctx: Context, ev: WebsiteContentEvent
) -> WebSearchEvent:
ctx.write_event_to_stream(
LogEvent(
msg=f"Performing web searches to identify companies who can benefit from the business's offerings."
)
)
prompt = f"With that you know about the business, perform a web search to find 5 tech companies who may benefit from the business's product. Only answer with the names of the companies you chose."
results = await self.agent.achat(prompt)
return WebSearchEvent(results=str(results.response))
@step
async def select_best_company(
self, ctx: Context, ev: WebSearchEvent
) -> RankingEvent:
ctx.write_event_to_stream(
LogEvent(
msg=f"Selecting the best company who can benefit from the business's offering…"
)
)
prompt = "Select one company that can benefit from the business's product. Only use your knowledge to select the company. Respond with just the name of the company. Do not use tools."
results = await self.agent.achat(prompt)
ctx.write_event_to_stream(
LogEvent(
msg=f"The agent selected this company: {results.response}"
)
)
return RankingEvent(results=str(results.response))
@step
async def prepare_email(self, ctx: Context, ev: RankingEvent) -> StopEvent:
ctx.write_event_to_stream(
LogEvent(msg=f"Drafting a short email for sales outreach…")
)
prompt = f"Draft a short cold sales outreach email for the company you picked. Do not use tools."
email = await self.agent.achat(prompt)
ctx.write_event_to_stream(
LogEvent(msg=f"Here is the email: {email.response}")
)
return StopEvent(result=str(email.response))
class WebsiteContentEvent(Event): contents: str class WebSearchEvent(Event): results: str class RankingEvent(Event): results: str class LogEvent(Event): msg: str class SalesRepWorkflow(Workflow): agent = ReActAgent( tools=th.get_tools(bundle="llamaindex test"), llm=llm, memory=ChatMemoryBuffer.from_defaults(), ) @step async def get_company_info( self, ctx: Context, ev: StartEvent ) -> WebsiteContentEvent: ctx.write_event_to_stream( LogEvent(msg=f"Getting the contents of {ev.url}…") ) prompt = f"Get the contents of {ev.url}, then summarize its key value propositions in a few bullet points." contents = await self.agent.achat(prompt) return WebsiteContentEvent(contents=str(contents.response)) @step async def find_prospects( self, ctx: Context, ev: WebsiteContentEvent ) -> WebSearchEvent: ctx.write_event_to_stream( LogEvent( msg=f"Performing web searches to identify companies who can benefit from the business's offerings." ) ) prompt = f"With that you know about the business, perform a web search to find 5 tech companies who may benefit from the business's product. Only answer with the names of the companies you chose." results = await self.agent.achat(prompt) return WebSearchEvent(results=str(results.response)) @step async def select_best_company( self, ctx: Context, ev: WebSearchEvent ) -> RankingEvent: ctx.write_event_to_stream( LogEvent( msg=f"Selecting the best company who can benefit from the business's offering…" ) ) prompt = "Select one company that can benefit from the business's product. Only use your knowledge to select the company. Respond with just the name of the company. Do not use tools." results = await self.agent.achat(prompt) ctx.write_event_to_stream( LogEvent( msg=f"The agent selected this company: {results.response}" ) ) return RankingEvent(results=str(results.response)) @step async def prepare_email(self, ctx: Context, ev: RankingEvent) -> StopEvent: ctx.write_event_to_stream( LogEvent(msg=f"Drafting a short email for sales outreach…") ) prompt = f"Draft a short cold sales outreach email for the company you picked. Do not use tools." email = await self.agent.achat(prompt) ctx.write_event_to_stream( LogEvent(msg=f"Here is the email: {email.response}") ) return StopEvent(result=str(email.response))
运行工作流¶
只需实例化工作流并传入公司 URL 即可开始。
In [ ]
Copied!
workflow = SalesRepWorkflow(timeout=None)
handler = workflow.run(url="https://toolhouse.ai")
async for event in handler.stream_events():
if isinstance(event, LogEvent):
print(event.msg)
workflow = SalesRepWorkflow(timeout=None) handler = workflow.run(url="https://toolhouse.ai") async for event in handler.stream_events(): if isinstance(event, LogEvent): print(event.msg)
Getting the contents of https://toolhouse.ai… Performing web searches to identify companies who can benefit from the business's offerings. Selecting the best company who can benefit from the business's offering… The agent selected this company: Cohere Drafting a short email for sales outreach… Here is the email: Subject: Streamline Your LLM Function Calling with Toolhouse Hi [Cohere Team], I noticed Cohere is leading the way in providing enterprise-ready LLM solutions. Given that your Command-r model already supports function calling, I thought you'd be interested in Toolhouse's developer toolkit that could enhance your clients' implementation experience. Toolhouse offers a unified SDK that streamlines LLM function calling across multiple models, including Cohere's. Our platform provides: - Pre-built, production-ready tools that reduce development time - Built-in analytics for easier debugging - A single integration point for multiple LLM tools Would you be open to a 15-minute call to discuss how Toolhouse could help Cohere's enterprise clients implement function calling more efficiently? Best regards, [Name]
