銷售與 CRM 助手

import json
import os
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI

# One API key per sales team (e.g. TUTUJIN_KEY_EAST): usage details can then be filtered and exported per key.
TEAM = os.environ.get("SALES_TEAM", "EAST")
client = OpenAI(base_url="https://www.tutujin.com/v1", api_key=os.environ["TUTUJIN_KEY_" + TEAM], max_retries=3)

# Set by the sales lead; every lead is scored on the same criteria.
CRITERIA = [
    {"id": "budget", "points": 30, "text": "Mentions a budget or a buying timeline"},
    {"id": "authority", "points": 25, "text": "The contact decides or names who decides"},
    {"id": "need", "points": 30, "text": "Describes a problem our product solves"},
    {"id": "fit", "points": 15, "text": "Company has 50 to 2,000 employees"},
]


def ask(model, system, user, temperature=0):
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
        temperature=temperature,
    )
    return resp.choices[0].message.content


def as_json(raw):
    return json.loads(raw[raw.find("{"): raw.rfind("}") + 1])


def call_notes(transcript):
    return as_json(ask(
        "qwen3.5-plus",
        "Turn the sales call transcript into CRM notes. Use only what was said. Reply with JSON only: "
        '{"summary": "", "pain_points": [], "objections": [], "decision_makers": [], "next_step": "", "next_step_date": "YYYY-MM-DD or empty"}',
        transcript,
    ))


def score_lead(lead):
    answers = as_json(ask(
        "gemini-2.5-flash-lite",
        'For each criterion answer yes, no or unknown, using only the lead data. Reply with JSON only, e.g. {"budget": "unknown"}',
        "Criteria:\n" + json.dumps(CRITERIA) + "\n\nLead:\n" + json.dumps(lead, ensure_ascii=False),
    ))
    return {"id": lead["id"], "score": sum(c["points"] for c in CRITERIA if answers.get(c["id"]) == "yes"), "answers": answers}


def follow_up_draft(notes, contact):
    # saved as a draft: the rep checks and sends it
    return ask(
        "deepseek-v4.1-flash",
        "Write a short follow-up email for the sales rep. Restate the agreed next step and date. "
        "Never promise prices, discounts or delivery dates that are not in the notes.",
        "Contact: " + json.dumps(contact, ensure_ascii=False) + "\nNotes: " + json.dumps(notes, ensure_ascii=False),
        temperature=0.5,
    )


# nightly batch: score yesterday's leads with a few requests in flight at a time
leads = [{"id": 1, "company": "Example Co", "employees": 300, "message": "I run support and we must replace our ticket tool before Q3."}]
with ThreadPoolExecutor(max_workers=8) as pool:
    ranked = sorted(pool.map(score_lead, leads), key=lambda r: r["score"], reverse=True)
print(ranked)

with open("call-0412.txt", encoding="utf-8") as f:
    notes = call_notes(f.read())
print(follow_up_draft(notes, {"name": "Ms. Chen", "company": "Example Co"}))