122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
"""Разбор сырого текста задания в плоскую карточку (LangGraph + Pydantic + OpenRouter)."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any, TypedDict
|
|
|
|
from dotenv import load_dotenv
|
|
from langchain_core.output_parsers import PydanticOutputParser
|
|
from langchain_core.prompts import PromptTemplate
|
|
from langchain_openai import ChatOpenAI
|
|
from langgraph.constants import END, START
|
|
from langgraph.graph import StateGraph
|
|
from pydantic import BaseModel, Field
|
|
|
|
load_dotenv()
|
|
|
|
OPENROUTER_BASE_URL = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
|
|
OPENROUTER_MODEL = os.getenv("OPENROUTER_MODEL", "poolside/laguna-m.1:free")
|
|
|
|
SAMPLE_TASK_TEXT = (
|
|
"Сдайте к пятнице мини-отчёт по LangChain: 2 страницы, упор на агентов. "
|
|
"Оценка: за полноту и за пример кода."
|
|
)
|
|
|
|
|
|
class TaskCard(BaseModel):
|
|
"""Плоская карточка задания."""
|
|
|
|
title: str = Field(description="Краткий заголовок задания")
|
|
subject: str = Field(description="Предмет или тема (LangChain, Python и т.д.)")
|
|
deadline_hint: str = Field(
|
|
description="Подсказка по сроку в свободной форме (к пятнице, до 20 марта)"
|
|
)
|
|
deliverable_type: str = Field(
|
|
description="Что нужно сдать: отчёт, код, презентация и т.п."
|
|
)
|
|
grading_hints: list[str] = Field(
|
|
description="Список критериев или подсказок по оценке"
|
|
)
|
|
|
|
|
|
class GraphState(TypedDict):
|
|
task_text: str
|
|
verdict_row: dict[str, Any] | None
|
|
|
|
|
|
def build_chain():
|
|
parser = PydanticOutputParser(pydantic_object=TaskCard)
|
|
prompt = PromptTemplate(
|
|
template=(
|
|
"Ты помощник студента. Из текста ниже извлеки структурированную карточку задания.\n"
|
|
"Верни только данные в формате из инструкций, без пояснений.\n\n"
|
|
"Текст задания:\n{task_text}\n\n"
|
|
"{format_instructions}"
|
|
),
|
|
input_variables=["task_text"],
|
|
partial_variables={"format_instructions": parser.get_format_instructions()},
|
|
)
|
|
llm = ChatOpenAI(
|
|
model=OPENROUTER_MODEL,
|
|
base_url=OPENROUTER_BASE_URL,
|
|
api_key=os.getenv("OPENAI_API_KEY"),
|
|
temperature=0,
|
|
)
|
|
return prompt | llm | parser
|
|
|
|
|
|
def parse_task_text(task_text: str) -> TaskCard:
|
|
chain = build_chain()
|
|
return chain.invoke({"task_text": task_text})
|
|
|
|
|
|
def parse_to_verdict_node(state: GraphState) -> dict[str, Any]:
|
|
"""Узел графа: заполняет verdict_row для автопроверки."""
|
|
card = parse_task_text(state["task_text"])
|
|
return {"verdict_row": card.model_dump()}
|
|
|
|
|
|
def build_graph():
|
|
"""Собирает и компилирует граф (ожидается автопроверкой)."""
|
|
builder = StateGraph(GraphState)
|
|
builder.add_node("parse", parse_to_verdict_node)
|
|
builder.add_edge(START, "parse")
|
|
builder.add_edge("parse", END)
|
|
return builder.compile()
|
|
|
|
|
|
# Экспорт для автопроверки: graph.invoke(...) -> {"verdict_row": {...}}
|
|
graph = build_graph()
|
|
|
|
|
|
def print_summary(card: TaskCard) -> None:
|
|
print("\n--- Краткая сводка ---")
|
|
print(f"title: {card.title}")
|
|
print(f"subject: {card.subject}")
|
|
print(f"deadline_hint: {card.deadline_hint}")
|
|
print(f"deliverable_type: {card.deliverable_type}")
|
|
print(f"grading_hints: {card.grading_hints}")
|
|
|
|
|
|
def main() -> None:
|
|
print(f"Модель OpenRouter: {OPENROUTER_MODEL}")
|
|
print(f"\nВход:\n{SAMPLE_TASK_TEXT}\n")
|
|
|
|
result = graph.invoke(
|
|
{"task_text": SAMPLE_TASK_TEXT, "verdict_row": None},
|
|
)
|
|
|
|
if not result.get("verdict_row"):
|
|
raise RuntimeError("Граф не вернул verdict_row")
|
|
|
|
card = TaskCard.model_validate(result["verdict_row"])
|
|
|
|
print("\n--- Валидированный объект (model_dump) ---")
|
|
print(card.model_dump())
|
|
|
|
print_summary(card)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|