LangChain LCEL و استریم پاسخ با API پیکو هوش

در پشت صحنه LangChain درخواست stream: true می‌فرستد و پیکو هوش پاسخ را با قالب SSE برمی‌گرداند. برای فهم قالب خام به راهنمای استریم SSE مراجعه کنید.

استریم async (مناسب FastAPI)

import asyncio

async def main():
    async for chunk in chain.astream({"topic": "ایندکس پایگاه داده", "lines": 3}):
        print(chunk, end="", flush=True)

asyncio.run(main())

نمونه‌ی ساده با FastAPI:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/ask")
async def ask(topic: str):
    async def gen():
        async for c in chain.astream({"topic": topic, "lines": 3}):
            yield c
    return StreamingResponse(gen(), media_type="text/plain; charset=utf-8")

اجرای موازی با RunnableParallel

from langchain_core.runnables import RunnableParallel

pros = ChatPromptTemplate.from_template("سه مزیت {tech} را فهرست کن.") | llm | StrOutputParser()
cons = ChatPromptTemplate.from_template("سه عیب {tech} را فهرست کن.") | llm | StrOutputParser()

both = RunnableParallel(pros=pros, cons=cons)
print(both.invoke({"tech": "میکروسرویس"}))

دو درخواست هم‌زمان ارسال می‌شود؛ مراقب سقف درخواست در دقیقه‌ی کلید خود باشید.

batch با کنترل هم‌زمانی

inputs = [{"topic": t, "lines": 2} for t in ["DNS", "TLS", "CDN", "NAT"]]
results = chain.batch(inputs, config={"max_concurrency": 2})
for r in results:
    print("-", r)

max_concurrency از رسیدن به خطای 429 جلوگیری می‌کند.

خروجی ساخت‌یافته

from pydantic import BaseModel

class Summary(BaseModel):
    title: str
    bullets: list[str]

structured = llm.with_structured_output(Summary)
print(structured.invoke("درباره‌ی HTTP/3 خلاصه بنویس"))

with_structured_output به قابلیت tool calling یا خروجی JSON مدل وابسته است. اگر مدل انتخابی از آن پشتیبانی نکند خطا می‌گیرید؛ در آن صورت مدل دیگری از کاتالوگ امتحان کنید یا از PydanticOutputParser استفاده کنید.

شمارش مصرف در حالت استریم

هزینه بر اساس توکن‌های واقعی محاسبه می‌شود. برای دیدن مصرف توکن در خود LangChain هنگام استریم، stream_usage=True را در ChatOpenAI بدهید (طبق مستندات LangChain، اگر متغیر OPENAI_BASE_URL تنظیم باشد این گزینه به‌طور پیش‌فرض خاموش است؛ و اگر مدل/مسیر گزارش usage در استریم را نپشتیبانی کند، مقدار برنمی‌گردد). سپس usage در آخرین chunk (فیلد usage_metadata) خواهد بود. فرمول دقیق در مقاله‌ی تخمین هزینه آمده است.

عیب‌یابی

افزودن حافظه‌ی گفتگو

LCEL به‌تنهایی حافظه ندارد. ساده‌ترین راه، نگه‌داشتن تاریخچه و ارسال آن در هر دور است:

from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

prompt = ChatPromptTemplate.from_messages([
    ("system", "پاسخ‌ها کوتاه و فارسی باشند."),
    MessagesPlaceholder("history"),
    ("human", "{input}"),
])
chat = prompt | llm | StrOutputParser()

history = []
for q in ["React چیست؟", "تفاوتش با Vue؟"]:
    a = chat.invoke({"history": history, "input": q})
    history += [HumanMessage(q), AIMessage(a)]
    print(a)

به یاد داشته باشید که تاریخچه‌ی بلندتر یعنی توکن ورودی بیشتر و هزینه‌ی بالاتر؛ در گفتگوهای طولانی، پیام‌های قدیمی را خلاصه یا حذف کنید.

ثبت رویدادها برای دیباگ

برای دیدن مراحل میانی زنجیره می‌توانید از astream_events یا callbackها استفاده کنید. نام و قالب رویدادها بین نسخه‌ها فرق می‌کند، بنابراین به مستندات نسخه‌ی نصب‌شده رجوع کنید:

async for ev in chain.astream_events({"topic": "Redis", "lines": 2}, version="v2"):
    if ev["event"] == "on_chat_model_stream":
        print(ev["data"]["chunk"].content, end="")

الگوی تلاش مجدد و fallback

LCEL امکان with_retry و with_fallbacks را می‌دهد. مثلاً اگر مدل اصلی خطای موقت داد، به مدل دوم برگردید:

backup = ChatOpenAI(model="claude-haiku-5-5", base_url="https://api.picoai.ir/v1",
                    api_key=os.environ["PICO_API_KEY"])
safe_llm = llm.with_retry(stop_after_attempt=3).with_fallbacks([backup])
chain = prompt | safe_llm | StrOutputParser()

پیکو هوش خودش هم مسیرهای پردازشی را در پشت صحنه جایگزین می‌کند، اما fallback در سطح برنامه برای قطعی کامل یک مدل یا محدودیت‌های حساب مفید است.

سوالات متداول

آیا استریم هزینه‌ی بیشتری دارد؟

خیر؛ هزینه فقط به توکن‌های ورودی و خروجی بستگی دارد، چه استریم باشد چه نباشد.

اگر استریم وسط راه قطع شود چه می‌شود؟

خطاهای قبل از شروع پاسخ بدون هزینه‌اند. برای قطع‌شدن‌های میانه‌ی راه، سمت کلاینت با تلاش مجدد و ذخیره‌ی متن دریافتی رفتار کنید؛ جزئیات در راهنمای خطاها.

کدام مدل برای استریم مناسب است؟

همه‌ی مدل‌های چت کاتالوگ استریم را پشتیبانی می‌کنند؛ برای سرعت می‌توانید gpt-6-luna یا claude-haiku-5-5 را امتحان کنید. مشخصات در قیمت‌ها.

راهنمای کلی API کجاست؟

در مستندات و صفحه‌ی LangChain.