diff --git a/main.py b/main.py index 9f7f44e..31d61fd 100644 --- a/main.py +++ b/main.py @@ -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()