Project Setup & Architecture
You'll need Python 3.10+, LangChain 0.3+, FastAPI, and uvicorn. Install:
pip install langchain langchain-community langchain-openai fastapi uvicorn pydanticSet your OpenAI API key as environment variable OPENAIAPIKEY. The project structure:
agent_task_manager/
├── main.py
├── tools.py
├── agent.py
└── models.pyWe build a REST API that accepts a user message and returns an agentic response. The agent has access to four tools: createplayertask, listalltasks, gettaskdetail, and updatetaskstatus. Internally we use an in-memory store (a Python dict) representing the team's task board.
Defining Tools with Pydantic
Each tool is a structured tool with a Pydantic model for inputs. This is a best practice from LangChain docs: structured tools allow the LLM to understand expected arguments and improve reliability.
# models.py
from pydantic import BaseModel, Field
from typing import Optional
class PlayerTaskInput(BaseModel):
player_name: str = Field(description="Player's name, e.g. 'Lebron'")
task_description: str = Field(description="Task details, e.g. 'Finish shooting drill'")
due_date: Optional[str] = Field(None, description="ISO date, e.g. '2025-05-01'")
class TaskQueryInput(BaseModel):
task_id: int = Field(description="Integer task ID")Now create tools in tools.py. We use a global dict task_board acting as the database.
# tools.py
from typing import Dict, List
from pydantic import BaseModel
from langchain_core.tools import StructuredTool
from models import PlayerTaskInput, TaskQueryInput
task_board: Dict[int, dict] = {}
next_id = 1
def create_player_task(player_name: str, task_description: str, due_date: str = None) -> str:
global next_id
task_id = next_id
next_id += 1
task_board[task_id] = {
"player": player_name,
"description": task_description,
"due_date": due_date or "No due date",
"status": "pending"
}
return f"Task {task_id} created for {player_name}: {task_description}"
def list_all_tasks() -> str:
if not task_board:
return "No tasks on the board."
lines = []
for tid, t in task_board.items():
lines.append(f"ID {tid}: {t['player']} - {t['description']} [{t['status']}]")
return "\n".join(lines)
def get_task_detail(task_id: int) -> str:
t = task_board.get(task_id)
if not t:
return f"Task {task_id} not found."
return f"Task {task_id}: Player {t['player']}, Desc: {t['description']}, Due: {t['due_date']}, Status: {t['status']}"
def update_task_status(task_id: int, status: str) -> str:
if task_id not in task_board:
return f"Task {task_id} not found."
task_board[task_id]["status"] = status
return f"Task {task_id} status updated to {status}."
create_tool = StructuredTool.from_function(
func=create_player_task,
name="create_player_task",
description="Add a new task for a player. Use when user wants to assign a task.",
args_schema=PlayerTaskInput
)
list_tool = StructuredTool.from_function(
func=list_all_tasks,
name="list_all_tasks",
description="Retrieve all tasks currently on the board.",
args_schema=None
)
get_tool = StructuredTool.from_function(
func=get_task_detail,
name="get_task_detail",
description="Get details of a specific task by ID.",
args_schema=TaskQueryInput
)
update_tool = StructuredTool.from_function(
func=update_task_status,
name="update_task_status",
description="Update the status of a task (pending, in_progress, completed).",
args_schema=TaskQueryInput # We'll extend later
)Note: updatetaskstatus expects two arguments; we should create a proper Pydantic model with task_id and status. Keep it simple for brevity, but in production you'd validate with Pydantic.
Building the Agent with Memory
We use LangChain's createreactagent (ReAct agent) with a ChatOpenAI model. Add conversation memory so the agent can refer to previous tasks. We'll use ConversationBufferMemory wrapped into the agent via MemorySaver.
# agent.py
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import PromptTemplate
from tools import create_tool, list_tool, get_tool, update_tool
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [create_tool, list_tool, get_tool, update_tool]
# Prompt template tailored to sports task manager
prompt = PromptTemplate.from_template(
"""You are a helpful assistant for a sports team's task board. You have access to tools that manage player tasks. Always answer helpfully.
Previous conversation:
{chat_history}
User: {input}
Agent scratchpad: {agent_scratchpad}"""
)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)We enable handleparsingerrors and max_iterations to prevent infinite loops—a common pitfall highlighted in LangChain community discussions.
Error Handling & Retry Logic
Tool execution can fail (e.g., invalid input). Wrap tool functions with try/except and return a descriptive error string. The agent will then re-prompt or ask for clarification. This pattern is recommended in engineering blogs (e.g., from LangChain docs).
# Within tools.py, modify functions:
def get_task_detail(task_id: int) -> str:
try:
t = task_board.get(task_id)
if not t:
return f"Task {task_id} not found."
return f"Task {task_id}: Player {t['player']}, Desc: {t['description']}, Due: {t['due_date']}, Status: {t['status']}"
except Exception as e:
return f"Error retrieving task: {str(e)}"For the agent, set max_retries=2 on the ChatOpenAI invocation to handle transient API errors.
Integrating with FastAPI
We expose a single endpoint that takes a user query and returns the agent's response. We use recursion depth limits and async support.
# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import agent_executor
app = FastAPI(title="Team Task Agent API")
class QueryRequest(BaseModel):
message: str
class QueryResponse(BaseModel):
response: str
@app.post("/ask", response_model=QueryResponse)
async def ask_agent(request: QueryRequest):
try:
result = await agent_executor.ainvoke({"input": request.message})
return QueryResponse(response=result["output"])
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Run with uvicorn main:app --reload. The agent is stateful per process; for production you'd persist memory to a database (e.g., Redis).
Running & Testing
Start the server, then send a request:
curl -X POST http://localhost:8000/ask \
-H "Content-Type: application/json" \
-d '{"message": "Assign Lebron to practice drills due 2025-06-01"}'Response should confirm creation. Then ask: "List all tasks." The agent uses memory to track the conversation.
Common Issues
- Tool recursion: When the LLM repeatedly calls a tool with the same input. Fix by setting
maxiterationsandearlystopping_method="generate". - Token limits: Long chat history can exceed context. Use
ConversationSummaryMemoryorConversationBufferWindowMemory(keep last k messages). - Structured tool parsing errors: The LLM may output malformed JSON. Use
handleparsingerrors=Trueto capture and retry. - Concurrent requests with stateful memory: Our in-memory store is not thread-safe. For production, use per-session memory stored in a database or external cache.
This architecture demonstrates the core patterns used in real-world AI agents at companies like Replit and Notion. Extend with authentication, persistent storage, and custom LLM calls.
Memuat komentar...