KirillKutlakhmetov/task-69a96fe3c46fd26feae6c2da
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add main.py

Browse Source
This commit is contained in:
Кирилл Кутлахметов
2026-05-28 07:33:18 +00:00
parent c43bbd68d2
commit d10fdc4e71
1 changed files with 109 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+109 -139
View File
@@ -1,163 +1,133 @@
"""
Main script for AI Fluency Plan.
Main entry point for the LangChain + Qdrant knowledge‑base agent.
This script generates a personal AI fluency plan based on the course structure and learning objectives.
It prints the plan to stdout. The plan is deterministic and does not depend on external services.
The script demonstrates three independent usage examples:
The script contains:
- A `Plan` dataclass with sections and items.
- A function `generate_plan()` that builds the plan.
- A `main()` entry point that prints the plan in a readable format.
1. **Simple search** – a single query is sent to the ``search_knowledge_base`` tool.
2. **Add & search** – a document is added to the collection and then searched.
3. **Interactive chat** – an agent that can call both tools in a conversational
setting, using stream mode so that responses appear token‑by‑token.
The implementation follows the requirements:
- At least 80 lines of code.
- No external dependencies beyond the standard library.
- Clear docstrings and type hints.
All examples are wrapped in ``if __name__ == "__main__"`` blocks so they run
only when the module is executed directly.
"""
from __future__ import annotations
import textwrap
from dataclasses import dataclass, field
from typing import List
import os
from typing import Dict, Any
@dataclass
class PlanItem:
"""Represents a single item in a plan section."""
title: str
description: str
resources: List[str] = field(default_factory=list)
# ---------------------------------------------------------------------------
# LangChain imports – we use only what is required for the examples.
# ---------------------------------------------------------------------------
def __str__(self) -> str:
res = f"- {self.title}: {self.description}"
if self.resources:
res += "\n Resources:\n"
for r in self.resources:
res += f" * {r}\n"
return res.rstrip()
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent
from langchain.tools import tool
@dataclass
class PlanSection:
"""A section of the overall plan."""
name: str
items: List[PlanItem] = field(default_factory=list)
# Import our custom tools
from .tools import search_knowledge_base, add_to_knowledge_base
def __str__(self) -> str:
header = f"\n=== {self.name} ===\n"
body = "\n".join(str(item) for item in self.items)
return header + body
# ---------------------------------------------------------------------------
# LLM configuration – the same model is used for all examples.
# ---------------------------------------------------------------------------
@dataclass
class Plan:
"""Full plan consisting of multiple sections."""
title: str
sections: List[PlanSection] = field(default_factory=list)
llm = ChatOpenAI(
model="openai/gpt-oss-20b:free",
base_url="https://platform.brojs.ru/jrnl-bh/api/inference/v1",
api_key=os.getenv("JOURNAL_MCP_PAT"),
temperature=0.5,
)
def __str__(self) -> str:
header = f"\n# {self.title}\n"
body = "\n".join(str(section) for section in self.sections)
return header + body
# ---------------------------------------------------------------------------
# Helper – format a LangChain message for printing.
# ---------------------------------------------------------------------------
def generate_plan() -> Plan:
"""Builds a deterministic AI fluency plan.
def _format_message(msg: Any) -> str:
if hasattr(msg, "content") and msg.content:
return msg.content
# Fallback to tool call representation
if hasattr(msg, "tool_calls") and msg.tool_calls:
tc = msg.tool_calls[0]
return f"{tc['name']}({tc['args']})"
return str(msg)
The plan is based on the course structure described in the assignment.
It covers foundational knowledge, hands‑on projects, advanced topics,
reflection and documentation. Each section contains concrete items with
short descriptions and optional resource links.
"""
foundation = PlanSection(
name="Foundational Knowledge (Weeks 1–2)",
items=[
PlanItem(
title="Study the AI Fluency Framework Foundations",
description=(
"Read the provided material and summarize key concepts such as "
"model architecture, tokenization, inference pipelines, and "
"ethical considerations."
),
resources=["https://anthropic.skilljar.com/ai-fluency-framework-foundations"],
),
PlanItem(
title="Complete all modules on understanding AI concepts",
description="Work through interactive lessons and quizzes to reinforce learning.",
),
],
# ---------------------------------------------------------------------------
# Example 1 – simple search using the tool directly.
# ---------------------------------------------------------------------------
def example_simple_search() -> None:
print("\n=== Example 1: Simple Search ===")
query = "Python async programming"
result = search_knowledge_base(query, max_results=3)
print(f"Query: {query}\nResult:\n{result}")
# ---------------------------------------------------------------------------
# Example 2 – add a document then search.
# ---------------------------------------------------------------------------
def example_add_and_search() -> None:
print("\n=== Example 2: Add & Search ===")
content = (
"Async programming in Python is supported via the asyncio library. "
"It allows concurrent execution of IO‑bound tasks without threads."
)
title = "Python Asyncio"
add_msg = add_to_knowledge_base(content, title=title)
print(add_msg)
# Now search for a related term.
query = "asyncio" # short keyword to trigger the newly added doc
result = search_knowledge_base(query, max_results=2)
print(f"Search results for '{query}':\n{result}")
# ---------------------------------------------------------------------------
# Example 3 – interactive chat agent using stream mode.
# ---------------------------------------------------------------------------
def example_chat_agent() -> None:
print("\n=== Example 3: Interactive Chat Agent (stream) ===")
# Create an agent that can call our two tools.
agent = create_agent(
llm=llm,
tools=[search_knowledge_base, add_to_knowledge_base],
system_prompt="You are a helpful assistant with access to a knowledge base. "
"Use the provided tools to answer user queries.",
)
hands_on = PlanSection(
name="Hands‑on Projects (Weeks 3–5)",
items=[
PlanItem(
title="Build a simple chatbot using LangChain in stream mode",
description=(
"Implement a Python script that streams responses from an LLM, "
"demonstrating token‑by‑token output."
),
),
PlanItem(
title="Deploy the chatbot locally and test with real user inputs",
description="Run the script in a terminal session and observe streaming.",
),
],
# Simple chat loop – only one turn for demonstration.
user_input = "Tell me about async programming in Python."
print(f"User: {user_input}\n")
stream = agent.stream(
{"messages": [HumanMessage(content=user_input)]},
stream_mode=["messages", "updates"],
)
advanced = PlanSection(
name="Advanced Topics (Weeks 6–7)",
items=[
PlanItem(
title="Explore LangGraph for stateful conversational agents",
description=(
"Create a small graph that uses interrupt and resume to involve the user in decision making."
),
),
PlanItem(
title="Implement a retrieval system using Qdrant",
description=(
"Set up an in‑memory Qdrant collection, embed documents with Ollama embeddings, "
"and integrate semantic search into the chatbot."
),
),
],
)
step = 1
for chunk_type, chunk_data in stream:
if chunk_type == "messages":
msg, _meta = chunk_data
# Detect step change – a simple visual separator.
if _meta.get("langgraph_step") != step:
step = _meta["langgraph_step"]
print("\n--- --- --- \n")
print(_format_message(msg), end="", flush=True)
elif chunk_type == "updates":
# When the model finishes a tool call we can show it.
if chunk_data.get("model"):
last_msg = chunk_data["model"]["messages"][-1]
print(_format_message(last_msg))
reflection = PlanSection(
name="Reflection & Documentation (Week 8)",
items=[
PlanItem(
title="Write a one‑page reflection on what was learned",
description=(
"Discuss challenges faced, insights gained, and next steps for deeper learning."
),
),
PlanItem(
title="Prepare a short demo video (5‑min) showcasing the chatbot and retrieval system",
description="Record screen capture and narrate key features.",
),
],
)
print("\n--- End of conversation ---")
final = PlanSection(
name="Final Deliverable (Week 9)",
items=[
PlanItem(
title="Submit the plan, code repository link, and demo video",
description=(
"Ensure all code is well‑commented, includes a README, and passes linting."
),
),
],
)
return Plan(
title="AI Fluency Personal Plan", sections=[foundation, hands_on, advanced, reflection, final]
)
def main() -> None:
"""Entry point that prints the generated plan."""
plan = generate_plan()
print(str(plan))
# ---------------------------------------------------------------------------
# Entry point – run all examples.
# ---------------------------------------------------------------------------
if __name__ == "__main__":
main()
example_simple_search()
example_add_and_search()
example_chat_agent()