Merge pull request #1 from aseifert/codex/create-sales_lead_researcher-example

Add sales_lead_researcher example

Author: aseifert

docs/examples.md

@@ -37,6 +37,7 @@ Check out a variety of sample implementations of the SDK in the examples section
 
  - **customer_service**: Example customer service system for an airline.
  - **research_bot**: Simple deep research clone.
+ - **sales_lead_researcher**: Collect CRM information about publishing contacts.
 
 - **[voice](https://github.com/openai/openai-agents-python/tree/main/examples/voice):**
  See examples of voice agents, using our TTS and STT models.

docs/ja/examples.md

@@ -38,8 +38,9 @@ search:
 - **[customer_service](https://github.com/openai/openai-agents-python/tree/main/examples/customer_service)** と **[research_bot](https://github.com/openai/openai-agents-python/tree/main/examples/research_bot):** 
  より実践的なユースケースを示す、拡張された 2 つの例です。
 
- - **customer_service**: 航空会社向けカスタマーサービスシステムの例 
- - **research_bot**: シンプルなディープリサーチクローン 
+ - **customer_service**: 航空会社向けカスタマーサービスシステムの例
+ - **research_bot**: シンプルなディープリサーチクローン
+ - **sales_lead_researcher**: 出版社向けリード情報を収集する例
 
 - **[voice](https://github.com/openai/openai-agents-python/tree/main/examples/voice):** 
  TTS と STT モデルを用いた音声エージェントの例をご覧ください。

examples/sales_lead_researcher/README.md

@@ -0,0 +1,11 @@
+# Sales lead researcher
+
+This example shows how to build a small multi‑agent workflow for gathering sales leads. The flow mirrors the `research_bot` example but is specialized for collecting CRM information about publishing contacts.
+
+Run it with:
+
+```bash
+python -m examples.sales_lead_researcher.main
+```
+
+You will be prompted to enter company and person pairs. Provide either `Company, Person` or just the company name. The agents will search the web (LinkedIn and company websites) in parallel and return structured lead information.

examples/sales_lead_researcher/__init__.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+
+from agents import Agent, WebSearchTool
+from agents.model_settings import ModelSettings
+
+INSTRUCTIONS = (
+ "You find the best editorial contact at a publishing company. Given just the company's name, "
+ "search the web to identify a likely decision maker we could reach out to. "
+ "Prefer people responsible for content or editorial decisions. "
+ "Return their full name and title. If unsure, make your best guess based on available information."
+)
+
+
+class ContactInfo(BaseModel):
+ full_name: str
+ title: str
+
+
+contact_finder_agent = Agent(
+ name="ContactFinderAgent",
+ instructions=INSTRUCTIONS,
+ tools=[WebSearchTool()],
+ model_settings=ModelSettings(tool_choice="required"),
+ output_type=ContactInfo,
+)

examples/sales_lead_researcher/agents/__init__.py

@@ -0,0 +1,44 @@
+from pydantic import BaseModel
+
+from agents import Agent, WebSearchTool
+from agents.model_settings import ModelSettings
+
+INSTRUCTIONS = (
+ "You gather sales lead information for our CRM. "
+ "Given a person's name and company, search the web (LinkedIn and the company's website) "
+ "and fill out the following fields as best you can. Use concise language."
+)
+
+
+class LeadInfo(BaseModel):
+ # Basic contact data
+ full_name: str
+ title: str
+ position: str
+ email: str | None
+ phone: str | None
+ linkedin: str | None
+
+ # Company data
+ company: str
+ company_type: str | None
+ company_size: str | None
+ headquarters: str | None
+ website: str | None
+ languages: str | None
+ publication_formats: str | None
+
+ # Publication/content data
+ publication_types: str | None
+ publication_frequency: str | None
+ topics: str | None
+ audience: str | None
+
+
+info_agent = Agent(
+ name="LeadInfoAgent",
+ instructions=INSTRUCTIONS,
+ tools=[WebSearchTool()],
+ model_settings=ModelSettings(tool_choice="required"),
+ output_type=LeadInfo,
+)

examples/sales_lead_researcher/agents/contact_finder_agent.py

@@ -0,0 +1,26 @@
+import asyncio
+
+from .manager import SalesLeadResearcher
+
+
+async def main() -> None:
+ print("Enter targets as 'Company, Person' or just 'Company'. Enter an empty line to finish.")
+ targets: list[tuple[str, str | None]] = []
+ while True:
+ line = input()
+ if not line.strip():
+ break
+ if "," in line:
+ company, person = [part.strip() for part in line.split(",", 1)]
+ else:
+ company, person = line.strip(), None
+ targets.append((company, person))
+
+ results = await SalesLeadResearcher().run(targets)
+ print("\n=====RESULTS=====")
+ for res in results:
+ print(res.json())
+
+
+if __name__ == "__main__":
+ asyncio.run(main())

examples/sales_lead_researcher/agents/info_agent.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import asyncio
+
+from rich.console import Console
+
+from agents import Runner, gen_trace_id, trace
+
+from .agents.contact_finder_agent import ContactInfo, contact_finder_agent
+from .agents.info_agent import LeadInfo, info_agent
+from .printer import Printer
+
+
+class SalesLeadResearcher:
+ def __init__(self) -> None:
+ self.console = Console()
+ self.printer = Printer(self.console)
+
+ async def run(self, targets: list[tuple[str, str | None]]) -> list[LeadInfo]:
+ trace_id = gen_trace_id()
+ with trace("Sales lead research", trace_id=trace_id):
+ self.printer.update_item(
+ "trace_id",
+ f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}",
+ is_done=True,
+ hide_checkmark=True,
+ )
+ tasks = [
+ asyncio.create_task(self._process(company, person)) for company, person in targets
+ ]
+ results: list[LeadInfo] = []
+ num_done = 0
+ for task in asyncio.as_completed(tasks):
+ result = await task
+ results.append(result)
+ num_done += 1
+ self.printer.update_item("progress", f"Processed {num_done}/{len(tasks)} leads")
+ self.printer.mark_item_done("progress")
+ self.printer.end()
+ return results
+
+ async def _process(self, company: str, person: str | None) -> LeadInfo:
+ if not person:
+ result = await Runner.run(contact_finder_agent, company)
+ info = result.final_output_as(ContactInfo)
+ person = info.full_name
+ input_data = f"Person: {person}\nCompany: {company}"
+ result = await Runner.run(info_agent, input_data)
+ return result.final_output_as(LeadInfo)

examples/sales_lead_researcher/main.py

@@ -0,0 +1,41 @@
+from typing import Any
+
+from rich.console import Console, Group
+from rich.live import Live
+from rich.spinner import Spinner
+
+
+class Printer:
+ def __init__(self, console: Console):
+ self.live = Live(console=console)
+ self.items: dict[str, tuple[str, bool]] = {}
+ self.hide_done_ids: set[str] = set()
+ self.live.start()
+
+ def end(self) -> None:
+ self.live.stop()
+
+ def hide_done_checkmark(self, item_id: str) -> None:
+ self.hide_done_ids.add(item_id)
+
+ def update_item(
+ self, item_id: str, content: str, is_done: bool = False, hide_checkmark: bool = False
+ ) -> None:
+ self.items[item_id] = (content, is_done)
+ if hide_checkmark:
+ self.hide_done_ids.add(item_id)
+ self.flush()
+
+ def mark_item_done(self, item_id: str) -> None:
+ self.items[item_id] = (self.items[item_id][0], True)
+ self.flush()
+
+ def flush(self) -> None:
+ renderables: list[Any] = []
+ for item_id, (content, is_done) in self.items.items():
+ if is_done:
+ prefix = "✅ " if item_id not in self.hide_done_ids else ""
+ renderables.append(prefix + content)
+ else:
+ renderables.append(Spinner("dots", text=content))
+ self.live.update(Group(*renderables))